第三章：数据获取与处理——量化交易的“原材料”工程

做量化交易，说白了就是跟数据打交道。策略再牛，模型再炫，数据源要是出了问题，一切都是白搭。我个人习惯把数据获取和处理比作“厨房备菜”——菜没洗干净、切得不均匀，大厨手艺再好也做不出好菜。

这一章，我们就来聊聊怎么用 ccxt 这个库，把交易所的K线数据“拿”下来，再把它收拾得干干净净，存到该存的地方去。

3.1 为什么是ccxt？

你想想看，市面上交易所那么多，币安、OKX、火币、Bybit……每个交易所的API签名方式、请求格式都不一样。如果每个交易所都单独写一套代码，那维护成本可就太高了。

ccxt 这个库，就是来解决这个问题的。它封装了上百家交易所的API，提供了一套统一的接口。你只需要切换一个参数，就能从币安切换到OKX。嗯，就是这么方便。

核心优势：

  • 统一接口：同样的代码，换交易所只需改 exchange_id
  • 支持现货、合约、期权等多种交易类型
  • 内置限频、重试机制，省去很多底层工作
  • 社区活跃，更新及时，新交易所上线很快就能支持

3.2 连接交易所API

先装好库：

pip install ccxt

然后，连接交易所。这里我以币安为例：

import ccxt

# 初始化交易所对象
exchange = ccxt.binance({
    'apiKey': '你的API_KEY',
    'secret': '你的SECRET_KEY',
    'enableRateLimit': True,  # 自动限频，防止被封
    'options': {
        'defaultType': 'spot',  # 现货市场
    }
})

# 测试连接
try:
    markets = exchange.load_markets()
    print(f"成功连接，共加载 {len(markets)} 个交易对")
except Exception as e:
    print(f"连接失败: {e}")

避坑指南：我曾经在项目里忘记设置 enableRateLimit=True，结果请求太频繁，IP直接被交易所封了半小时。那半小时，策略跑着跑着突然没数据了，回测数据也断了，真是欲哭无泪。所以，这个参数一定要加上。

3.3 获取K线数据

K线数据，也叫OHLCV（Open, High, Low, Close, Volume）。这是量化策略最基础的数据源。

# 获取BTC/USDT的1小时K线，最近100根
symbol = 'BTC/USDT'
timeframe = '1h'
limit = 100

ohlcv = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)

# 打印前5条看看
for candle in ohlcv[:5]:
    print(candle)

返回的数据格式是这样的：

时间戳 开盘价 最高价 最低价 收盘价 成交量
1700000000000 50000.0 50100.0 49900.0 50050.0 123.45
1700003600000 50050.0 50200.0 50000.0 50100.0 98.76

注意，时间戳是毫秒级的Unix时间戳。我习惯把它转成可读的日期格式：

import pandas as pd

df = pd.DataFrame(ohlcv, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
print(df.head())

3.4 数据清洗与预处理

数据拿到手，别急着用。交易所的数据偶尔也会“抽风”——比如某根K线的价格突然变成0，或者时间戳错乱。我遇到过好几次，回测跑得挺漂亮，实盘一跑就亏钱，最后发现是数据里有脏数据。

所以，数据清洗是必须的。我一般做这几步：

  1. 检查缺失值：看看有没有NaN或者0值
  2. 检查时间连续性：K线应该是等间隔的，不能有跳空
  3. 检查异常值：价格突然暴涨暴跌几十倍，多半是数据错误
  4. 处理复权：如果涉及分红、拆分，需要做复权处理（数字货币一般不需要）
# 数据清洗示例
def clean_ohlcv(df):
    # 1. 删除空值
    df = df.dropna()
    
    # 2. 删除价格为0的行
    df = df[(df['open'] > 0) & (df['high'] > 0) & (df['low'] > 0) & (df['close'] > 0)]
    
    # 3. 检查时间连续性
    time_diff = df['timestamp'].diff().dropna()
    expected_diff = pd.Timedelta(hours=1)  # 1小时K线
    if (time_diff != expected_diff).any():
        print("警告：存在时间不连续的K线")
    
    # 4. 删除价格异常值（比如涨跌幅超过50%）
    df['return'] = df['close'].pct_change()
    df = df[df['return'].abs() < 0.5]
    df = df.drop(columns=['return'])
    
    return df

df_clean = clean_ohlcv(df)

个人经验：我建议在清洗时保留一份原始数据备份。万一清洗逻辑有bug，还能回退重来。我就吃过这个亏——清洗完直接覆盖了原文件，后来发现过滤条件太严，把正常数据也删了，又得重新下载。

3.5 数据存储

数据清洗完了，得存起来。存哪里？两种主流方案：CSV和数据库。

3.5.1 CSV存储

简单、轻量、方便查看。适合小规模数据或者个人研究。

# 保存为CSV
df_clean.to_csv('btc_usdt_1h.csv', index=False)

# 读取CSV
df_read = pd.read_csv('btc_usdt_1h.csv', parse_dates=['timestamp'])

3.5.2 数据库存储

数据量大、需要频繁查询、或者多人协作时，数据库是更好的选择。我个人偏爱SQLite，因为它不需要单独安装数据库服务，文件即数据库，特别适合个人量化项目。

import sqlite3

# 连接数据库（如果不存在会自动创建）
conn = sqlite3.connect('crypto_data.db')
cursor = conn.cursor()

# 创建表
cursor.execute('''
    CREATE TABLE IF NOT EXISTS ohlcv (
        symbol TEXT,
        timeframe TEXT,
        timestamp DATETIME,
        open REAL,
        high REAL,
        low REAL,
        close REAL,
        volume REAL,
        PRIMARY KEY (symbol, timeframe, timestamp)
    )
''')

# 插入数据
df_clean.to_sql('ohlcv', conn, if_exists='append', index=False)

# 查询数据
query = "SELECT * FROM ohlcv WHERE symbol='BTC/USDT' AND timeframe='1h' ORDER BY timestamp DESC LIMIT 10"
result = pd.read_sql(query, conn)

conn.close()

存储方案对比：

方案 优点 缺点 适用场景
CSV 简单、可读性强、无需额外依赖 查询慢、不支持并发、数据量大时效率低 个人研究、小规模数据
SQLite 轻量、支持SQL查询、单文件 并发写入性能一般 个人/小团队量化项目
PostgreSQL/MySQL 高性能、支持并发、功能强大 需要安装维护、配置复杂 团队协作、生产环境

3.6 本章知识体系总览

下面这张图，帮你把本章的核心逻辑串起来：

数据获取与处理流程 1. 连接交易所 ccxt初始化 + API认证 2. 获取K线数据 fetch_ohlcv() 3. 数据清洗 去空值、去异常、检查连续性 4. 数据存储 CSV文件 数据库 SQLite / PostgreSQL / MySQL 支持SQL查询、并发访问 CSV文件 轻量、可读性强 适合个人研究、小规模数据 核心工具：ccxt + pandas + sqlite3

3.7 避坑总结

最后，把我这些年踩过的坑总结一下，你遇到了可以直接对照：

  • 限频问题：一定要设置 enableRateLimit=True，否则IP被封是迟早的事
  • 数据完整性：交易所偶尔会返回不完整的数据，比如某根K线只有开盘价没有收盘价。我建议每次获取数据后都做一次完整性校验
  • 时间戳时区：交易所返回的时间戳通常是UTC时间。如果你在本地查看，记得做时区转换，否则K线时间对不上
  • 数据备份：原始数据永远保留一份。清洗逻辑可以迭代，但原始数据丢了就得重新下载，浪费时间
  • 存储格式：如果数据量超过10万行，CSV的读写速度会明显变慢。这时候就该考虑数据库了

好了，数据获取与处理这块，核心内容就这些。记住一句话：数据质量决定策略上限。花80%的时间把数据搞干净，剩下20%的时间写策略，你会发现回测结果靠谱得多。

公众号：蓝海数据掘金营，微信 deep3321

← 上一章 📖 返回目录 下一章 →