数据模块搭建：Tick级数据获取、清洗与对齐、存储方案
做高频回测，数据就是你的命根子。我见过太多人策略写得漂亮，最后死在数据上——不是缺了字段，就是时间戳对不上，回测结果跟实盘差了十万八千里。今天咱们就把数据模块彻底讲透。
一、Tick级数据获取：从哪里来，怎么拿
先说说数据源。我个人习惯用三类来源：
- 交易所直连：延迟最低，但门槛高，适合机构
- 数据服务商：如Wind、聚宽、RiceQuant，省心但贵
- 开源数据：比如Binance的API，免费但需要自己清洗
拿Binance的Tick数据举个例子。你想想看，每秒可能产生上千条成交记录，怎么高效拉取？
import requests
import pandas as pd
from datetime import datetime
def fetch_binance_ticks(symbol, start_time, end_time):
"""
获取Binance的Tick级成交数据
"""
url = f"https://api.binance.com/api/v3/aggTrades"
params = {
'symbol': symbol,
'startTime': int(start_time.timestamp() * 1000),
'endTime': int(end_time.timestamp() * 1000),
'limit': 1000 # 单次最多1000条
}
resp = requests.get(url, params=params)
data = resp.json()
# 转成DataFrame
df = pd.DataFrame(data)
df['timestamp'] = pd.to_datetime(df['T'], unit='ms')
df = df.rename(columns={
'a': 'agg_trade_id',
'p': 'price',
'q': 'quantity',
'T': 'trade_time'
})
return df[['timestamp', 'price', 'quantity', 'agg_trade_id']]
这里有个坑：Binance的aggTrades接口一次最多返回1000条。如果你要拉一整天的数据，得做分页。我在项目中遇到过，直接循环请求会被限流，后来加了随机延时才搞定。
二、数据清洗：脏数据是回测的隐形杀手
拿到原始Tick数据后，你会发现什么妖魔鬼怪都有。我总结了几类常见问题：
| 问题类型 | 表现 | 处理方法 |
|---|---|---|
| 时间戳错乱 | 后一条的时间比前一条还早 | 按时间排序后，删除逆序记录 |
| 价格异常 | 价格突然变成0或负数 | 设定合理阈值，超出则剔除 |
| 重复数据 | 同一条成交记录出现多次 | 按trade_id去重 |
| 缺失字段 | 某些Tick缺少成交量 | 用前一条数据填充或直接丢弃 |
嗯，这里要重点说说时间戳对齐。不同交易所的时间戳精度不一样，有的到毫秒，有的到微秒。你如果直接拿来做回测，会发现买卖单对不上。
def clean_ticks(df):
"""
清洗Tick数据：去重、排序、异常值处理
"""
# 1. 去重
df = df.drop_duplicates(subset=['agg_trade_id'])
# 2. 排序
df = df.sort_values('timestamp').reset_index(drop=True)
# 3. 价格异常过滤
median_price = df['price'].median()
df = df[df['price'].between(median_price * 0.9, median_price * 1.1)]
# 4. 时间戳对齐到毫秒
df['timestamp_ms'] = df['timestamp'].dt.floor('ms')
return df
三、数据对齐：让买卖双方对上话
高频回测里，最头疼的就是数据对齐。你想想看，买一价和卖一价的时间戳可能差了几微秒，但就是这几微秒，决定了你的订单能不能成交。
我常用的对齐策略有三种：
- 精确对齐：按时间戳精确匹配，适合纳秒级数据
- 插值对齐：用前后数据插值，适合毫秒级数据
- 聚合对齐：按固定时间窗口（如1ms）聚合，适合大多数场景
def align_ticks(bid_df, ask_df, freq='1ms'):
"""
将买卖Tick数据对齐到统一时间轴
"""
# 设置时间索引
bid_df = bid_df.set_index('timestamp_ms')
ask_df = ask_df.set_index('timestamp_ms')
# 重采样到固定频率
bid_resampled = bid_df['price'].resample(freq).last().ffill()
ask_resampled = ask_df['price'].resample(freq).last().ffill()
# 合并
aligned = pd.DataFrame({
'bid_price': bid_resampled,
'ask_price': ask_resampled
})
# 计算中间价
aligned['mid_price'] = (aligned['bid_price'] + aligned['ask_price']) / 2
return aligned.dropna()
说白了，对齐的本质就是「把不同频率的数据强行塞进同一个时间框架」。但要注意，ffill（向前填充）会引入未来信息，回测时容易过拟合。我建议用bfill（向后填充）或者只保留完全对齐的时间点。
四、数据存储：HDF5 vs Parquet
数据清洗对齐完了，得存起来。高频数据量巨大，一天可能几个GB，选错存储格式会让你欲哭无泪。
我对比过HDF5和Parquet，直接上结论：
| 特性 | HDF5 | Parquet |
|---|---|---|
| 压缩率 | 中等（默认zlib） | 高（snappy/zstd） |
| 读取速度 | 快（支持列切片） | 非常快（列式存储） |
| 写入速度 | 慢（追加模式还行） | 快（批量写入） |
| 跨语言支持 | Python为主 | 多语言（Spark、Java等） |
| 文件大小 | 较大 | 小（通常小30-50%） |
我个人习惯：日常研究用HDF5，因为Pandas直接支持，读写方便。生产环境用Parquet，压缩率高，而且后续如果要上Spark做大规模回测，直接就能用。
# HDF5存储示例
def save_to_hdf5(df, filepath, key='ticks'):
"""
将Tick数据存储为HDF5格式
"""
with pd.HDFStore(filepath, mode='a') as store:
store.put(key, df, format='table',
complib='zlib', complevel=5)
store.get_storer(key).attrs.description = 'Tick级成交数据'
# Parquet存储示例
def save_to_parquet(df, filepath, partition_cols=None):
"""
将Tick数据存储为Parquet格式
"""
df.to_parquet(filepath,
engine='pyarrow',
compression='snappy',
partition_cols=partition_cols)
五、完整的数据流水线
最后，我把整个数据模块串起来，画了一张流程图。你看一眼就明白了：
这条流水线跑下来，你的数据模块基本就稳了。记住：数据质量决定了回测的天花板。策略再牛，数据是脏的，结果就是垃圾进垃圾出。
数据模块搭建完了，接下来就是回测引擎的核心逻辑了。不过那是另一章的内容，咱们先把数据这关过了再说。