数据获取与清洗：实时行情数据源选择、数据清洗与对齐、数据缺失处理
做量化交易，第一步就是搞定数据。我见过太多人把精力全花在策略模型上，结果数据源一塌糊涂，回测跑得飞起，实盘直接翻车。说白了，数据是地基，地基不稳，楼盖得再漂亮也得塌。
今天咱们就聊聊实时行情数据怎么拿、怎么洗、怎么对齐。嗯，这里面的坑，我基本都踩过一遍。
一、实时行情数据源选择
选数据源，我个人的习惯是看三个维度：速度、质量、成本。不可能三角，你懂的。
| 数据源类型 | 延迟 | 质量 | 成本 | 适用场景 |
|---|---|---|---|---|
| 交易所直连（CTP/行情网关） | <1ms | 最高 | 极高（硬件+席位费） | 高频、做市商 |
| 券商/期货公司API | 1-10ms | 高 | 中等（开户即可） | 中频、日内策略 |
| 第三方数据商（如Wind、聚宽） | 10-100ms | 中高 | 低-中 | 回测、低频策略 |
| 免费数据源（Yahoo、新浪） | 秒级 | 低（常有缺失） | 免费 | 学习、个人研究 |
我的建议： 如果你做的是T+0日内策略，至少用券商API。免费数据源做回测可以，但实盘千万别用——我曾经吃过这个亏，某次行情推送延迟了3秒，策略直接买在了最高点。
二、数据清洗与对齐
数据拿到手，第一件事不是分析，是洗。为什么？因为交易所、券商、第三方给你的数据，格式、时间戳、精度全都不一样。你想想看，如果A股行情用毫秒时间戳，期货行情用微秒，你怎么合并？
2.1 时间戳对齐
这是最头疼的环节。我遇到过最离谱的情况：同一个合约，不同数据源的时间戳差了整整1秒。原因是什么？有的数据源用的是撮合时间，有的用的是接收时间。
我的做法是：
- 统一转成UTC纳秒：所有数据源的时间戳，先转成int64的纳秒数。别用字符串，字符串比较慢，而且容易出错。
- 建立基准时钟：以交易所行情网关的时间为基准，其他数据源做偏移校正。
- 丢弃异常时间戳：比如时间戳比当前时间还晚的，直接扔掉。
# 伪代码示例：时间戳对齐
def align_timestamp(raw_ts, source_type):
if source_type == 'exchange':
# 交易所直连，时间戳可信
return raw_ts
elif source_type == 'broker':
# 券商API，可能有几十微秒延迟
return raw_ts - OFFSET_BROKER
elif source_type == 'third_party':
# 第三方数据，延迟较大，做线性插值
return interpolate_ts(raw_ts)
else:
raise ValueError("未知数据源")
2.2 数据对齐（Tick级）
不同品种的Tick频率不一样。比如股指期货每秒可能来几十笔，而某些冷门股票可能几秒才一笔。对齐的时候怎么办？
我个人习惯用快照对齐法：
- 以最快的数据源为基准，比如期货行情。
- 对于慢的数据源，取最近一笔有效数据填充。
- 如果超过100ms没有新数据，标记为缺失。
注意： 千万别用「未来数据」！比如你用下一笔Tick的价格来填充当前时间，回测会看起来很美，实盘直接亏光。我刚开始做的时候犯过这个错，回测年化50%，实盘一个月亏了20%。
三、数据缺失处理
数据缺失是常态。交易所偶尔断流、网络抖动、API限流，都会导致数据掉点。怎么处理？
3.1 缺失类型判断
| 缺失类型 | 特征 | 处理方式 |
|---|---|---|
| 随机缺失 | 零星几个Tick丢失 | 线性插值或前值填充 |
| 连续缺失 | 连续几秒甚至几分钟无数据 | 标记为「无效区间」，策略暂停 |
| 开盘/收盘缺失 | 集合竞价阶段数据不全 | 使用交易所官方开盘价 |
3.2 我的实战处理流程
嗯，这里直接上代码吧。这是我目前在用的一个简化版处理函数：
def handle_missing_data(tick_df, max_gap_ms=100):
"""
tick_df: 包含时间戳和价格的DataFrame
max_gap_ms: 允许的最大缺失间隔（毫秒）
"""
# 1. 计算时间间隔
tick_df['gap_ms'] = tick_df['timestamp'].diff().dt.total_seconds() * 1000
# 2. 判断缺失类型
missing_mask = tick_df['gap_ms'] > max_gap_ms
# 3. 处理
for idx in tick_df[missing_mask].index:
gap = tick_df.loc[idx, 'gap_ms']
if gap < 1000: # 小于1秒，线性插值
tick_df.loc[idx, 'price'] = (
tick_df.loc[idx-1, 'price'] +
tick_df.loc[idx+1, 'price']
) / 2
else: # 大于1秒，标记为无效
tick_df.loc[idx, 'valid'] = False
return tick_df
避坑指南： 我曾经在实盘环境中，因为网络抖动导致连续5秒没有行情数据。当时我的策略没有做缺失检测，直接用了前值填充。结果那5秒里价格已经跌了2%，策略还在按旧价格下单，直接造成了滑点亏损。从那以后，我规定：连续缺失超过1秒，必须暂停交易。
四、知识体系总览
说了这么多，咱们用一张图来总结一下整个数据获取与清洗的流程：
这张图把整个流程串起来了。从数据源选择开始，到获取、缓存，再到清洗对齐，最后处理缺失。每一步都环环相扣，缺一不可。
一个小技巧： 在实际工程中，我建议把数据清洗做成一个独立的微服务。这样即使行情源换了，或者清洗逻辑改了，也不会影响策略模块。我现在的系统就是这样的架构，维护起来轻松很多。
好了，数据获取与清洗这块就聊到这儿。记住一句话：数据质量决定策略上限。再好的策略，喂进去脏数据，出来的也是垃圾结果。别嫌清洗麻烦，这步做好了，后面能省你80%的调试时间。