3、数据获取与预处理：多市场数据源选择

做跨市场价差策略，第一步就是搞数据。说实话，这一步要是没做好，后面所有模型都是空中楼阁。我在这个坑里摔过好几次，今天把经验掰开了讲。

3.1 多市场数据源怎么选

数据源的选择，说白了就是「快、准、稳」三个字。但不同市场，情况完全不一样。

3.1.1 交易所官方API

我个人最推荐的方式。为什么？因为数据最原始，没有经过第三方加工，延迟最低。

• 国内期货：CTP（综合交易平台）是主流。我2018年做螺纹钢和热卷的跨品种套利时，用的就是CTP直连。速度确实快，但部署起来有点麻烦。
• A股：上交所的L1/L2行情、深交所的行情网关。Level-2数据能看到逐笔委托，做高频价差策略很关键。
• 加密货币：Binance、OKX、Coinbase都有WebSocket接口。嗯，这里要注意，不同交易所的时间戳格式不一样，有的用毫秒，有的用微秒。

3.1.2 第三方数据商

如果你不想自己折腾，第三方数据商是省心的选择。我目前项目中用的比较多的是这几个：

数据商	覆盖市场	数据频率	我的评价
Wind	A股、期货、债券	Tick/分钟/日	国内最全，但贵
聚宽	A股、期货	分钟/日	适合回测，API友好
Quandl	全球市场	日/周/月	国际数据首选
CoinGecko	加密货币	分钟/日	免费但有限流

我曾经在Quandl上吃过亏——他们的数据偶尔会有跳变，后来我加了一层校验才放心。

3.2 数据对齐：时间戳、频率、除权除息

跨市场策略最头疼的问题是什么？数据对不齐。你想想看，A股9:30开盘，期货9:00就开了，加密货币更是7×24小时。怎么对齐？

3.2.1 时间戳对齐

不同交易所的时间戳格式五花八门。我见过Unix秒、毫秒、微秒，还有带时区的字符串。统一转换成UTC时间戳是最稳妥的做法。

import pandas as pd

# 统一转换为UTC时间戳
def align_timestamp(series, source_format='ms'):
    if source_format == 'ms':
        return pd.to_datetime(series, unit='ms', utc=True)
    elif source_format == 's':
        return pd.to_datetime(series, unit='s', utc=True)
    else:
        return pd.to_datetime(series, utc=True)

# 示例：Binance的毫秒时间戳
binance_ts = 1700000000000
print(align_timestamp([binance_ts], 'ms'))

3.2.2 频率对齐

做价差策略，频率必须统一。我一般用1分钟线做基准。为什么？因为1分钟既能捕捉短期价差波动，又不会像Tick数据那样噪声太大。

• 降频：从Tick到1分钟，用resample聚合。注意用OHLC（开盘、最高、最低、收盘）方式。
• 升频：从日线到分钟线，这个要小心。我建议用前向填充，不要插值，因为未来数据会泄露。

3.2.3 除权除息处理

做股票价差策略，这个必须处理。我记得有一次回测螺纹钢和热卷的价差，发现策略表现特别好，后来才发现是没处理分红——价差里包含了分红收益，根本不是真实的套利机会。

常用的复权方式有两种：

• 前复权：调整历史价格，让价格连续。适合回测。
• 后复权：调整当前价格，适合看真实收益。

# 简单前复权示例
def adjust_forward(price_series, dividend, split_ratio=1):
    adj_factor = (1 - dividend/price_series.iloc[-1]) * split_ratio
    return price_series * adj_factor

3.3 缺失值处理与异常值检测

数据拿到手，第一件事不是建模，而是清洗。我见过太多人直接拿原始数据跑策略，结果一塌糊涂。

3.3.1 缺失值处理

缺失值的原因很多：交易所休市、网络中断、数据商漏发。处理方式要看情况：

• 前向填充：用上一个有效值填充。适合非交易时段。
• 线性插值：适合短时间缺失，比如1-2个数据点。
• 直接删除：如果缺失比例超过5%，我建议直接删掉这段数据。

3.3.2 异常值检测

异常值对价差策略的影响特别大。一个极端值就能让价差看起来「回归」了，其实是数据错误。

我常用的方法：

1. Z-score方法：超过3倍标准差视为异常。简单粗暴，但有效。
2. IQR方法：四分位距法，适合非正态分布的数据。
3. 滚动窗口法：用过去N个点的均值和标准差判断。这个我最常用，因为能适应市场变化。

def detect_outliers(series, window=20, threshold=3):
    rolling_mean = series.rolling(window).mean()
    rolling_std = series.rolling(window).std()
    z_scores = (series - rolling_mean) / rolling_std
    return abs(z_scores) > threshold

# 使用示例
price_data = pd.Series([100, 101, 102, 500, 101, 100])
outliers = detect_outliers(price_data)
print(f"异常值位置: {outliers[outliers].index.tolist()}")

3.4 知识体系总览

下面这张图是我自己总结的数据处理流程，每次做新策略都会拿出来对照一遍。

这张图把整个流程串起来了。从数据源到最终可用的价差数据，每一步都不能跳过。我刚开始做的时候，总想省掉清洗步骤，结果回测漂亮、实盘亏钱——数据质量决定了策略的天花板。

好了，数据获取和预处理就讲到这里。记住一句话：垃圾进，垃圾出。数据质量不过关，再好的策略也是白搭。