第三章：数据获取与清洗——交易所API对接、Tick级与分钟级数据获取、缺失值与异常值处理实战

做基差交易，数据就是你的眼睛。眼睛要是花了，再好的策略也白搭。

我见过太多人，策略逻辑写得漂漂亮亮，回测曲线美如画，一上实盘就崩。为什么？十有八九是数据源出了问题。要么是Tick数据对不上，要么是分钟线有跳空，要么是某个合约的行情突然断了。嗯，今天我们就来聊聊，怎么把这双“眼睛”擦亮。

3.1 交易所API对接：选对路子，少走弯路

国内期货市场，主流交易所就四家：上期所、大商所、郑商所、中金所。每家都有自己的数据接口，但说实话，直接对接交易所的API，成本很高。

我个人习惯，优先选CTP（综合交易平台）接口。CTP是上期技术开发的，几乎覆盖了国内所有期货公司。你只要在期货公司开了户，拿到BrokerID和UserID，就能通过CTP获取实时行情。

举个例子，CTP的行情订阅代码大概长这样：

// C++伪代码示例
CThostFtdcMdApi* pMdApi = CThostFtdcMdApi::CreateFtdcMdApi();
pMdApi->RegisterSpi(this);
pMdApi->RegisterFront("tcp://180.168.146.187:10010");
pMdApi->Init();

// 订阅合约
char* instruments[] = {"rb2401", "hc2401"};
pMdApi->SubscribeMarketData(instruments, 2);

// 回调函数处理Tick
void OnRtnDepthMarketData(CThostFtdcDepthMarketDataField* pData) {
    // pData->LastPrice, pData->Volume, pData->BidPrice1 ...
}

这里有个坑，我一开始也踩过。CTP的行情服务器地址，不同期货公司可能不一样。你得找你的客户经理要最新的地址列表。我曾经因为用了过期的地址，整整两天没收到行情，还以为是代码写错了。

3.2 Tick级数据获取：每一笔成交都是故事

Tick数据，就是每一笔成交的切片。它包含了：最新价、成交量、持仓量、买卖盘口（五档或十档）。对于基差交易来说，Tick数据是计算瞬时基差的基础。

为什么需要Tick级？你想想看，基差的变化往往在几秒钟内完成。如果你用分钟线，可能刚看到基差拉大，还没来得及下单，价差就回归了。Tick数据能让你捕捉到最细微的波动。

获取Tick数据，我推荐两种方式：

• CTP直连： 延迟最低，适合高频交易。但需要自己维护连接，处理断线重连。
• 第三方数据商： 如万得、聚宽、米筐。它们提供封装好的API，用Python就能调。适合策略研发阶段。

用Python从聚宽获取Tick数据的示例：

# Python示例
from jqdatasdk import *
auth('your_account', 'your_password')

# 获取某一天的Tick数据
df = get_price('RB8888', start_date='2024-01-15', end_date='2024-01-15',
               frequency='tick', fields=['time', 'current', 'volume', 'high', 'low'])

print(df.head())

注意，Tick数据量非常大。一个活跃合约一天能产生几万条Tick。如果你同时监控10个合约，一天的数据量就是几十万条。存储和读取都是挑战。我个人习惯用ClickHouse或InfluxDB这类时序数据库来存，MySQL扛不住的。

3.3 分钟级数据获取：效率与精度的平衡

不是所有策略都需要Tick级数据。对于日内中低频策略，1分钟或5分钟的K线就够用了。分钟级数据的好处是：噪声少，信号稳定，存储成本低。

分钟级数据的获取，通常有两种做法：

• 从Tick合成： 自己写程序，把Tick数据按时间窗口聚合成K线。优点是灵活，可以自定义开盘价、收盘价的计算方式。
• 直接拉取： 从数据商直接获取分钟K线。省事，但可能缺失一些细节（比如最高价、最低价是否准确）。

我自己写过一个合成函数，核心逻辑是这样的：

def tick_to_minute(tick_df, freq='1min'):
    """
    将Tick数据合成为分钟K线
    """
    # 按时间分组
    grouped = tick_df.groupby(pd.Grouper(key='time', freq=freq))
    
    # 计算OHLC
    ohlc = grouped['price'].agg([
        ('open', 'first'),
        ('high', 'max'),
        ('low', 'min'),
        ('close', 'last'),
        ('volume', 'sum')
    ])
    
    return ohlc

这里有个细节：分钟线的“开盘价”应该是这一分钟的第一笔成交价，而不是上一分钟的最后一笔。很多新手会搞混，导致回测信号偏移。嗯，这个问题我排查了整整一个下午才找到原因。

3.4 缺失值与异常值处理：数据清洗的硬功夫

数据拿到手，别急着用。先看看有没有“脏数据”。我总结了三类常见问题：

问题类型	表现	原因
缺失值	某段时间内没有行情数据	网络中断、交易所休市、合约不活跃
异常值	价格突然跳变，比如从5000跳到10000	数据商传输错误、撮合异常、涨跌停板
重复值	同一时间戳出现多条相同数据	API重复推送、数据存储逻辑bug

3.4.1 缺失值处理

对于分钟级数据，缺失值通常用“前向填充”或“插值法”。我个人更推荐前向填充，因为期货市场有连续性，前一分钟的收盘价就是下一分钟的开盘价。

# 前向填充
df.fillna(method='ffill', inplace=True)

但要注意：如果连续缺失超过5分钟，就别填充了。说明这段时间可能真的没有交易（比如夜盘休市）。强行填充会引入虚假数据。

3.4.2 异常值处理

异常值检测，我常用“3σ法则”或“IQR（四分位距）”。对于基差数据，还可以用“价差合理性检验”。

举个例子，螺纹钢和热卷的基差正常情况下在-200到+200之间。如果某天突然出现基差=1000，那大概率是数据错了。

def detect_outliers(series, method='iqr'):
    if method == 'iqr':
        Q1 = series.quantile(0.25)
        Q3 = series.quantile(0.75)
        IQR = Q3 - Q1
        lower = Q1 - 1.5 * IQR
        upper = Q3 + 1.5 * IQR
        return (series < lower) | (series > upper)
    elif method == 'zscore':
        mean = series.mean()
        std = series.std()
        return (series - mean).abs() > 3 * std

我曾经在回测中发现一个策略收益异常高，查了半天，原来是某天的数据里有个价格少了一个小数点。5000变成了500，导致基差计算完全错误。从那以后，我每次拿到新数据，都会先跑一遍异常值检测。

3.5 数据清洗的完整流程

说了这么多，总结一下我的数据清洗流水线：

1. 去重： 按时间戳和合约代码去重，保留第一条。
2. 排序： 按时间升序排列，确保时间序列连续。
3. 缺失值检测： 找出缺失的时间段，判断是否合理。
4. 异常值检测： 用IQR或Z-score标记异常点。
5. 人工复核： 对于标记为异常的数据，结合市场新闻判断。比如某天突然暴跌，可能是真实事件，不是数据错误。
6. 存储： 清洗后的数据存入数据库，并打上“clean”标签。

这一步看似繁琐，但能省掉后面90%的调试时间。我见过太多人，策略跑出来不对劲，第一反应是改参数，结果折腾半天才发现是数据有问题。

3.6 本章知识体系

下面这张图，帮你理清数据获取与清洗的完整链路：

这张图展示了从数据源到策略调用的完整链路。每一步都有坑，但只要你按流程走，就能把数据质量控制在可接受范围内。

好了，数据获取与清洗就聊到这儿。记住一句话：花在数据清洗上的时间，永远不会白费。

---