4. 数据获取：获取期货行情数据的常用方法与工具

做跨品种套利，第一步不是写策略，而是搞定数据。

我见过太多人，策略逻辑想得天花乱坠，结果数据源一塌糊涂。回测跑得挺美，实盘一跑就崩。说白了，数据是量化交易的「米」，没米下不了锅。

4.1 数据获取的核心痛点

期货行情数据，跟股票不太一样。股票你拿个日线就能做很多分析，期货不行。

为什么？因为期货有到期日，有换月移仓，有夜盘。你想想看，螺纹钢主力合约每个月都在变，你要是拿连续数据不做处理，回测结果能信吗？

我个人习惯把数据获取的难点归纳为三个：

• 连续性：主力合约切换时，价格会跳空，必须做复权或拼接
• 时效性：期货夜盘到凌晨，数据更新频率高，延迟大了没法做高频
• 成本：高质量数据源不便宜，免费数据又容易出坑

4.2 常用数据源对比

我这些年用过不下十种数据源，踩过不少坑。下面这张表是我个人比较下来觉得靠谱的：

数据源	类型	覆盖品种	更新频率	成本	推荐场景
天勤(TQ)	Python SDK	国内全品种	Tick/秒级	免费+付费	入门、快速原型
万得(Wind)	终端/API	国内外全品种	分钟/日线	较贵	机构、研究
聚宽(JQData)	Python SDK	国内期货	分钟/日线	免费额度	回测、策略开发
CTP直连	C++/Python	交易所直连	实时Tick	按流量	实盘、高频
AKShare	Python库	国内期货	日线/分钟	免费	学习、小规模

4.3 实战：用Python获取期货数据

我拿天勤来举个例子。这个工具我用了三年，比较顺手。

先安装：

pip install tqsdk

然后获取螺纹钢和热卷的日线数据：

from tqsdk import TqApi, TqAuth

api = TqApi(auth=TqAuth("你的账号", "密码"))

# 获取螺纹钢主力连续合约
rb_klines = api.get_kline_serial(
    "KQ.m@SHFE.rb", 
    duration_seconds=86400,  # 日线
    data_length=500
)

# 获取热卷主力连续合约
hc_klines = api.get_kline_serial(
    "KQ.m@SHFE.hc", 
    duration_seconds=86400,
    data_length=500
)

# 打印最近5条数据
print(rb_klines.tail())
print(hc_klines.tail())

api.close()

嗯，这里要注意。天勤的「KQ.m@」前缀代表主力连续合约，它会自动处理换月。但自动处理不代表完美，我遇到过几次换月时价格跳空特别大的情况，需要自己再做一次过滤。

4.4 数据清洗与预处理

数据拿到手，别急着用。先做三件事：

1. 去重：有时候网络波动会重复推送同一根K线
2. 补缺失：节假日、夜盘休市，时间戳可能不连续
3. 对齐时间：不同品种的交易时间不一样，比如螺纹钢有夜盘，但有些品种没有。做套利时，时间戳必须对齐到同一秒

我一般这样处理：

import pandas as pd

def clean_futures_data(df):
    # 去重
    df = df.drop_duplicates(subset=['datetime'])
    
    # 按时间排序
    df = df.sort_values('datetime')
    
    # 填充缺失值（用前向填充）
    df = df.fillna(method='ffill')
    
    # 只保留交易时段（过滤掉集合竞价等）
    df = df[df['volume'] > 0]
    
    return df

# 对齐两个品种的时间
def align_two_series(df1, df2):
    merged = pd.merge(
        df1[['datetime', 'close']], 
        df2[['datetime', 'close']], 
        on='datetime', 
        how='inner',
        suffixes=('_rb', '_hc')
    )
    return merged

4.5 数据存储与管理

数据量大了以后，每次从API拉取太慢。我建议本地存一份。

我的做法是：

• 日线数据存 CSV，按品种分文件
• 分钟数据存 SQLite，按日期分表
• Tick 数据存 Parquet，压缩比高

举个例子，存分钟数据到 SQLite：

import sqlite3

conn = sqlite3.connect('futures_data.db')

# 创建表
conn.execute('''
    CREATE TABLE IF NOT EXISTS rb_minute (
        datetime TEXT PRIMARY KEY,
        open REAL,
        high REAL,
        low REAL,
        close REAL,
        volume INTEGER
    )
''')

# 批量插入
df.to_sql('rb_minute', conn, if_exists='append', index=False)

conn.close()

说实话，存储这块不用搞太复杂。刚开始做研究，CSV 完全够用。等数据量上来了，再考虑数据库。

4.6 本章知识体系

下面这张图，是我自己总结的数据获取全流程：

这张图把整个流程串起来了。你从数据源开始，一步步走到本地存储，每一步都有坑，但每一步也都有解法。

---

公众号:蓝海数据掘金营，微信deep3321