数据获取与清洗：多合约数据对齐、缺失值处理、异常值检测、数据频率转换

做跨品种套利，最怕什么？

不是策略亏钱，而是数据有问题。我见过太多人，策略逻辑写得漂漂亮亮，回测曲线也好看，结果一上实盘就崩。为什么？数据没对齐，频率不一致，缺失值随便填了个0。嗯，这些坑我都踩过。

今天咱们就聊聊数据获取与清洗。说白了，就是让不同品种的数据「说同一种语言」。

一、多合约数据对齐：让它们站在同一起跑线

跨品种套利，至少涉及两个合约。比如螺纹钢和热卷，或者豆粕和豆油。这些合约的交易时间、节假日、甚至交易所规则都可能不同。

我个人的习惯是：先确定一个「基准合约」，然后让其他合约向它看齐。

举个例子。螺纹钢主力合约和热卷主力合约，虽然都在上期所，但它们的最后交易日、交割规则不同。如果你只按「日期」对齐，可能会发现某一天螺纹钢有数据，热卷却没有——因为热卷那天是交割日，暂停交易了。

怎么做？我一般用 pandas 的 merge_asof 函数。它能按时间戳「最近邻」匹配，比普通的 merge 灵活得多。

import pandas as pd

# 假设 rb_df 是螺纹钢数据，hc_df 是热卷数据
# 两者都有 'datetime' 列

aligned_df = pd.merge_asof(
    rb_df.sort_values('datetime'),
    hc_df.sort_values('datetime'),
    on='datetime',
    direction='nearest',
    tolerance=pd.Timedelta('1min')  # 允许1分钟误差
)

这里有个坑：tolerance 参数别设太大。我曾经设成5分钟，结果回测出来年化收益高得离谱——后来发现是数据错位了，相当于「用未来的价格预测过去」。嗯，这种错误最隐蔽。

二、缺失值处理：别随便填0

数据缺失是常态。节假日、涨跌停、临时停牌，都会导致某条记录为空。

你想想看，如果直接填0，会有什么后果？价差瞬间变成极端值，策略可能直接开仓。我见过有人因此一天亏了3%。

处理缺失值，我按优先级排序：

1. 向前填充（ffill）：用上一个有效值填充。适合非交易时段。
2. 向后填充（bfill）：用下一个有效值填充。适合开盘前数据。
3. 插值法：线性插值或时间加权插值。适合短时间缺失。
4. 删除：如果缺失比例超过5%，直接删掉那段时间。

# 向前填充缺失值
aligned_df['price_rb'] = aligned_df['price_rb'].ffill()
aligned_df['price_hc'] = aligned_df['price_hc'].ffill()

# 如果还有缺失，说明两个合约都停牌了，直接删除
aligned_df = aligned_df.dropna(subset=['price_rb', 'price_hc'])

三、异常值检测：揪出那些「离谱」的数据

数据里总有些「怪物」。比如螺纹钢价格突然跳到10000元/吨，或者价差瞬间变成500点。这些多半是数据录入错误，或者交易所的「乌龙指」。

我常用的方法有三种：

我个人偏爱滚动窗口法。因为它能适应市场波动变化——牛市的「正常波动」和熊市的「正常波动」不一样，用固定阈值会出问题。

# 滚动窗口法检测异常值
window = 20
mean = aligned_df['spread'].rolling(window).mean()
std = aligned_df['spread'].rolling(window).std()

# 超过3倍标准差视为异常
aligned_df['is_outlier'] = (
    (aligned_df['spread'] > mean + 3 * std) |
    (aligned_df['spread'] < mean - 3 * std)
)

# 标记异常值，但不删除，后续人工复核
outliers = aligned_df[aligned_df['is_outlier']]
print(f"发现 {len(outliers)} 个异常值")

四、数据频率转换：统一「节奏」

不同合约的数据频率可能不同。比如螺纹钢是1分钟线，热卷是5分钟线。或者你想做日频套利，但拿到的数据是分钟级的。

频率转换，说白了就是「降频」或「升频」。

• 降频：从高频到低频。比如1分钟转5分钟。用 resample 聚合，取OHLC（开盘、最高、最低、收盘）。
• 升频：从低频到高频。比如日线转小时线。这个要小心，因为会引入「未来信息」——你没法用今天的收盘价去填充昨天的小时线。

我建议：尽量用原始频率做套利。如果非要转换，只做降频，不做升频。

# 将1分钟数据降频为5分钟
rb_5min = rb_df.resample('5T', on='datetime').agg({
    'open': 'first',
    'high': 'max',
    'low': 'min',
    'close': 'last',
    'volume': 'sum'
}).dropna()

这里有个细节：resample 默认按「左闭右开」区间。比如 9:30-9:35 的数据，会归到 9:30 这个时间戳。我个人习惯用「左闭右闭」，这样更符合交易直觉。

# 左闭右闭的resample
rb_5min = rb_df.set_index('datetime').resample('5T', closed='left', label='left').agg(...)

五、知识体系总览

说了这么多，咱们用一张图来总结。这张图展示了数据获取与清洗的完整流程：

这张图把整个流程串起来了。从原始数据开始，经过时间对齐、缺失值处理、异常值检测、频率转换，最后得到干净可用的数据。每一步都有对应的技术方案，咱们刚才都讲到了。

好了，数据清洗这块就聊到这儿。记住一个原则：垃圾进，垃圾出。数据质量决定了策略的天花板。花80%的时间在数据上，都不为过。