3. 数据模块搭建：历史行情数据获取、数据清洗与对齐、数据存储格式选择
做市策略回测，说白了就是拿历史数据模拟交易。数据要是脏的、歪的、对不齐的，后面算出来的夏普比率再漂亮也是白搭。我见过太多人花80%时间调参数，却不愿意花20%时间把数据弄干净。嗯，今天咱们就把这块硬骨头啃下来。
3.1 历史行情数据获取：从哪里来，拿什么
数据源的选择，直接决定了回测的可靠性。我个人习惯把数据源分成三档：
| 数据源 | 特点 | 适用场景 |
|---|---|---|
| 交易所官方API | 最准确，但限流严重 | 小规模回测、实盘验证 |
| 第三方数据商（如QuantConnect、Polygon） | 清洗过，有Tick级数据 | 专业回测、高频策略 |
| 开源数据包（如ccxt、yfinance） | 免费，但质量参差不齐 | 快速原型、学习验证 |
我在项目中遇到过最坑的事：用某个免费数据源做回测，年化收益跑出30%，结果一查，数据里混了停牌期间的零成交数据。所以我的建议是——永远不要只依赖一个数据源。至少拿两个源交叉验证一下。
核心原则：数据获取阶段，宁可慢一点，也要保证每根K线的时间戳、开盘价、收盘价、成交量四个字段是完整的。缺一不可。
代码层面，我习惯这样封装一个数据获取器：
import ccxt
import pandas as pd
from datetime import datetime
class DataFetcher:
def __init__(self, exchange_id='binance'):
self.exchange = getattr(ccxt, exchange_id)({
'enableRateLimit': True,
'rateLimit': 1200
})
def fetch_ohlcv(self, symbol, timeframe='1m', since=None, limit=1000):
"""获取K线数据，自动处理分页"""
all_data = []
if since is None:
since = self.exchange.parse8601('2020-01-01T00:00:00Z')
while True:
data = self.exchange.fetch_ohlcv(
symbol, timeframe, since=since, limit=limit
)
if len(data) == 0:
break
all_data.extend(data)
since = data[-1][0] + 1 # 避免重复
print(f'已获取 {len(all_data)} 条数据...')
df = pd.DataFrame(all_data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
return df.set_index('timestamp')
小技巧：获取数据时，记得把时间戳转成UTC。我吃过亏——本地时区跟交易所时区不一致，导致回测里出现了「未来数据」。你想想看，这多可怕。
3.2 数据清洗与对齐：脏数据比没数据更可怕
数据拿到手，第一件事不是跑策略，而是洗数据。我总结了一套「三步清洗法」：
- 去重：同一个时间戳出现两条数据，保留第一条还是最后一条？我建议保留最后一条，因为可能是重传的修正数据。
- 补缺失：某根K线缺失了，是前向填充还是插值？做市策略里，我一般用前向填充——毕竟真实交易中，缺失的那根K线你也没法交易。
- 去异常：价格突然跳空100倍？成交量突然为0但价格没变？这些都得标记出来。
我曾经遇到过一个案例：某币种在凌晨3点突然出现一根涨幅500%的K线，后来发现是交易所数据推送的bug。如果没清洗，策略会以为这是个「突破信号」然后追高买入——结果可想而知。
数据对齐这块，做市策略尤其讲究。因为你要同时监控多个交易对，它们的K线时间戳可能不完全对齐。我的做法是：
def align_data(data_dict, freq='1min'):
"""
将多个交易对的数据对齐到统一时间轴
data_dict: {'BTC/USDT': df1, 'ETH/USDT': df2, ...}
"""
# 生成统一时间轴
all_timestamps = pd.concat([df.index for df in data_dict.values()])
unified_index = pd.date_range(
start=all_timestamps.min(),
end=all_timestamps.max(),
freq=freq
)
# 重采样并对齐
aligned = {}
for symbol, df in data_dict.items():
# 先重采样到目标频率
resampled = df.resample(freq).ffill()
# 再对齐到统一时间轴
aligned[symbol] = resampled.reindex(unified_index, method='ffill')
return aligned
注意：对齐时用ffill（前向填充）还是bfill（后向填充）？做市策略里，你只能知道过去的价格，所以必须用ffill。用bfill等于偷看了未来数据，回测结果会虚高。
3.3 数据存储格式选择：HDF5 vs Parquet
数据清洗完了，存成什么格式？我试过CSV、SQLite、HDF5、Parquet。说实话，CSV在小规模数据时还行，一旦数据量上到几十GB，读写速度会让你崩溃。
我个人最常用的是Parquet，原因有三：
- 压缩率高：同样的数据，Parquet比CSV小5-10倍。我有个项目，原始CSV有120GB，转成Parquet只剩18GB。
- 列式存储：做回测时，你经常只需要读取「close」这一列。Parquet可以只加载需要的列，速度飞快。
- 兼容性好：Pandas、Spark、Dask都能直接读写，不用额外装驱动。
但HDF5也有它的优势：
| 特性 | HDF5 | Parquet |
|---|---|---|
| 写入速度 | 快（尤其单次写入大量数据） | 中等 |
| 读取速度 | 中等 | 快（列式读取优势明显） |
| 压缩率 | 高（但依赖压缩算法） | 非常高（默认snappy压缩） |
| 跨语言支持 | 好（C/Fortran/Python/Java） | 非常好（几乎所有大数据框架都支持） |
| 适合场景 | 科学计算、大量小文件 | 数据分析、数据仓库 |
我的建议是：做市策略回测，优先选Parquet。因为回测过程中，你需要频繁读取不同时间窗口的数据，Parquet的列式存储和谓词下推能大幅提升效率。
下面是我常用的存储和读取封装：
import pandas as pd
import pyarrow.parquet as pq
class DataStore:
def __init__(self, base_path='./data'):
self.base_path = base_path
def save_parquet(self, df, symbol, timeframe):
"""按交易对和时间粒度存储"""
path = f'{self.base_path}/{symbol}/{timeframe}.parquet'
df.to_parquet(path, compression='snappy', index=True)
print(f'数据已保存至 {path}')
def load_parquet(self, symbol, timeframe, columns=None,
start=None, end=None):
"""按需加载，支持列筛选和时间范围过滤"""
path = f'{self.base_path}/{symbol}/{timeframe}.parquet'
# 只加载需要的列
if columns:
table = pq.read_table(path, columns=columns)
else:
table = pq.read_table(path)
df = table.to_pandas()
# 时间范围过滤
if start:
df = df[df.index >= start]
if end:
df = df[df.index <= end]
return df
经验之谈：存储时，把数据按「交易对/时间粒度」分目录存放。比如 ./data/BTC-USDT/1m.parquet。这样后续扩展新交易对时，不用动已有数据。我一开始把所有数据塞一个文件里，后来加个新币种就得重写整个存储逻辑——血的教训。
3.4 数据模块的整体架构
最后，我把整个数据模块的流程画成了一张图。你看一眼就能明白数据是怎么流转的：
数据模块是整个回测框架的地基。你想想看，如果地基是歪的，上面盖的楼再漂亮也没用。我见过太多人花大量时间优化策略逻辑，却不愿意花半天时间把数据弄干净。结果回测跑出来曲线漂亮得不行，一上实盘就崩——十有八九是数据的问题。
嗯，数据模块就讲到这里。记住一句话：垃圾进，垃圾出。数据干净了，策略才能发挥真正的实力。