第二章：价差数据获取——交易所数据接口介绍、历史价差数据下载、实时价差数据订阅、数据清洗与对齐

做跨期套利，说白了就是跟价差打交道。但价差这东西不会自己从天上掉下来，你得自己去拿。我见过太多新手，策略写得天花乱坠，结果数据源一塌糊涂，回测跑得飞起，实盘直接翻车。嗯，数据获取是地基，地基不稳，楼盖得再高也得塌。

这一章，我就把数据获取这摊子事给你捋清楚。从交易所接口到历史数据下载，从实时订阅到数据清洗，咱们一步步来。

2.1 交易所数据接口：你得知道去哪拿数据

国内期货交易所主要有四家：上期所、大商所、郑商所、中金所。每家都有自己的数据接口，但说实话，直接对接交易所的CTP（综合交易平台）接口，是咱们做套利最常用的路子。

CTP接口的特点：

  • 实时行情推送：Tick级数据，毫秒级延迟。做高频套利，这个必须得有。
  • 历史数据查询：支持查询历史K线，但一般只保留最近几个交易日的数据。想拿更久远的？得自己存。
  • 多合约支持：一个接口就能拿到所有合约的行情，省事。
我的个人习惯：CTP接口虽然强大，但部署起来有点麻烦。如果你只是做研究，可以用一些第三方数据服务商，比如万得、聚宽、米筐。它们封装好了接口，直接调API就行。但实盘交易，我建议还是直接上CTP，少一层转发就少一层延迟。

除了CTP，还有FIX协议、各交易所的专线接口。不过对于咱们做跨期套利，CTP基本够用了。你想想看，价差套利对时效性要求高，CTP的Tick数据完全能满足。

2.2 历史价差数据下载：回测的命根子

做策略回测，没有历史数据就是巧妇难为无米之炊。我个人习惯把历史数据存在本地数据库里，用的时候直接查。

下载历史数据的步骤：

  1. 确定合约对：比如螺纹钢RB2410和RB2501，这两个合约的价差就是你要研究的对象。
  2. 选择时间范围：至少一年以上的数据，才能覆盖不同的市场环境。我建议至少两年。
  3. 调用接口下载：用CTP的QueryHistoryData接口，或者第三方API。

下面是一个用Python下载历史数据的示例，我用的是聚宽的API，简单易懂：

import jqdatasdk as jq
import pandas as pd

# 登录聚宽
jq.auth('你的账号', '你的密码')

# 定义合约对
contract1 = 'RB2410.XSGE'
contract2 = 'RB2501.XSGE'

# 下载日线数据
df1 = jq.get_price(contract1, start_date='2023-01-01', end_date='2024-01-01', frequency='daily')
df2 = jq.get_price(contract2, start_date='2023-01-01', end_date='2024-01-01', frequency='daily')

# 计算价差
spread = df1['close'] - df2['close']

# 保存到本地
spread.to_csv('rb_spread_historical.csv')
print('历史价差数据下载完成！')
注意：不同交易所的合约代码格式不一样。上期所是XXXXXX.XSGE，大商所是XXXXXX.XDCE，郑商所是XXXXXX.XZCE。搞错了代码，数据是拿不到的。我曾经因为这个坑，浪费了半天时间排查。

2.3 实时价差数据订阅：实盘交易的生命线

实盘交易跟回测完全是两码事。回测你可以慢慢算，实盘你得实时盯着价差的变化。实时数据订阅，就是让你能第一时间捕捉到价差的异常波动。

实时订阅的实现方式：

  • CTP行情订阅：通过MdApi订阅合约的Tick数据，然后自己计算价差。
  • WebSocket推送：一些第三方服务商提供WebSocket接口，实时推送行情。
  • 本地缓存+计算：把实时数据缓存到内存里，每次新数据来了就更新价差。

我个人比较推荐CTP的行情订阅，因为它延迟最低。下面是一个简单的CTP订阅示例（伪代码，实际CTP接口比较复杂）：

# 伪代码示例
class CTPMdSpi:
    def OnRtnDepthMarketData(self, pDepthMarketData):
        # 收到行情数据
        if pDepthMarketData.InstrumentID == 'RB2410':
            self.last_price_1 = pDepthMarketData.LastPrice
        elif pDepthMarketData.InstrumentID == 'RB2501':
            self.last_price_2 = pDepthMarketData.LastPrice
        
        # 计算实时价差
        if self.last_price_1 and self.last_price_2:
            spread = self.last_price_1 - self.last_price_2
            print(f'实时价差: {spread}')
关键点：实时价差计算一定要快。我建议在行情回调里直接算，不要开线程去轮询。每多一次线程切换，就多一次延迟。做套利，延迟就是成本。

2.4 数据清洗与对齐：别让脏数据坑了你

数据拿到手，你以为就能直接用了吗？太天真了。我见过太多人，回测跑得漂亮，实盘一塌糊涂，最后发现是数据没对齐。价差数据最怕的就是时间戳对不上。

常见的数据问题：

  • 缺失值：某个合约在某时刻没有成交，导致价差计算不出来。
  • 异常值：比如价格突然跳到一个离谱的数字，可能是交易所的撮合错误。
  • 时间戳不对齐：两个合约的Tick数据到达时间不一样，直接相减会算出错误的价差。

清洗与对齐的步骤：

  1. 去除异常值：用3倍标准差法，或者设定合理的价格范围。比如螺纹钢价格不可能突然从3000跳到10000。
  2. 填充缺失值：用前向填充（ffill）或者插值法。我个人习惯用前向填充，因为期货市场流动性好，缺失值很少。
  3. 时间对齐：把两个合约的数据按时间戳对齐。比如都对齐到最近的秒级时间戳，然后取同时刻的数据计算价差。

下面是一个数据清洗的Python示例：

import pandas as pd
import numpy as np

# 假设df1和df2是两个合约的Tick数据
df1 = pd.read_csv('rb2410_tick.csv')
df2 = pd.read_csv('rb2501_tick.csv')

# 1. 去除异常值
df1 = df1[(df1['price'] > 3000) & (df1['price'] < 5000)]
df2 = df2[(df2['price'] > 3000) & (df2['price'] < 5000)]

# 2. 时间戳对齐到秒
df1['time_sec'] = pd.to_datetime(df1['time']).dt.floor('S')
df2['time_sec'] = pd.to_datetime(df2['time']).dt.floor('S')

# 3. 合并数据
merged = pd.merge(df1, df2, on='time_sec', how='inner', suffixes=('_1', '_2'))

# 4. 计算价差
merged['spread'] = merged['price_1'] - merged['price_2']

# 5. 填充缺失值（如果有的话）
merged['spread'].fillna(method='ffill', inplace=True)

print('数据清洗完成，共{}条有效价差数据'.format(len(merged)))
避坑指南：我曾经在回测时用了未对齐的数据，结果价差序列看起来波动很大，策略收益高得离谱。后来一查，原来是两个合约的时间戳差了0.5秒，导致价差计算错误。从那以后，我每次做回测前都会先检查数据对齐情况。

知识体系总览

下面这张图，把数据获取的整个流程串起来了。你可以把它当作一个检查清单，做数据准备时对照着来：

价差数据获取知识体系 数据来源 获取方式 CTP接口 第三方API WebSocket 数据类型 历史数据 实时数据 数据处理 去除异常值 填充缺失值 时间对齐 计算价差 干净的价差数据 → 策略回测/实盘

数据获取这件事，看起来简单，但细节决定成败。我见过太多人在这上面栽跟头。记住一句话：数据质量决定了策略的天花板。数据是脏的，再好的策略也白搭。

好了，这一章的内容就到这里。数据拿到手、洗干净了，下一步就是分析价差的统计特征，看看它到底有没有规律可循。

专注资料整理

← 上一章 📖 返回目录 下一章 →