中金所股指期货主力合约自动识别：一个接口搞定 IF/IC/IH 连续合约合成
作者: TickDB Research · 发布: 2026/5/14 · 阅读: 6
标签: C 类, 掘金, 期货
交割周周三上午 10:14，IF2406 一分钟只成交了 3 手，价格跳动 4 个点。你的策略信号在这个时间点触发了“突破买入”——但那 4 个点的跳动不是趋势，是非主力合约的流动性枯竭。更致命的是，你回测里混入了 3 天这种非主力数据，年化收益高估了 4 个百分点，自己完全不知道。
>
手工维护一张“IF/IC/IH 换月日历表”的量化开发者，迟早会踩进同一个坑：主力合约切换的时机错了，整个回测曲线就是噪音拼出来的。
📌 本文解决的核心问题
| 你的痛点 | 本文答案 | 预估节省时间 |
|---|---|---|
| 手工维护主力合约映射表，换月时点靠人肉判断 | 全量期货品种自动分组 + 按成交量排序，主力合约一键识别 | 每次换月省 30 分钟，永久复用 |
| 回测里混入了非主力合约的流动性枯竭 K 线 | 连续合约拼接逻辑，切换点用成交量阈值联合时间规则判断 | 避免年化收益高估 4 个百分点 |
| 不同数据源的主力标准不同（成交量 vs 持仓量） | 代码中同时拉取两个指标，按需切换判断逻辑 | 一套代码两种标准，不再混用 |
| 不知道限流后等多久 | 优先读 Retry-After 头部，服务端给什么等什么 | 避免盲目 sleep 浪费时间 |
目录
- Step 1：全量期货品种枚举 + 按 IF/IC/IH 分组
主力合约识别的第一性原理：你的回测曲线里藏着多少非主力 K 线
主力合约识别：为什么成交量比持仓量更适合做切换信号
这是量化回测里最容易高估收益的坑。拆成五步看。
① 是什么
主力合约，是指同一品种下成交量最大、流动性最好的那个合约。中金所每个股指期货品种同时挂牌当月、下月及随后两个季月共四个合约。对 IF 来说，同一时间有 IF2606、IF2607、IF2609、IF2612 四只在交易。主力合约识别，就是在这四只里自动挑出“成交量最大的那一只”。
② 为什么必须用主力合约做回测
非主力合约的流动性会断崖式萎缩。交割周前 3 天，当月合约的成交从几千手骤降到几十手甚至个位数。一分钟 K 线里出现 4 个点的跳动，不是趋势信号，是买卖盘枯竭后的价格跳空。回测引擎无法区分“真实的趋势突破”和“流动性枯竭导致的噪音跳动”，会把后者也当成有效信号计入绩效，直接拉高年化收益。
③ 怎么用
逻辑三步走。先从中金所全量品种里筛出 IF/IC/IH/IM 开头的合约。再按品种前缀分组，每组内用 ticker 快照拉取实时成交量（volume_24h）。最后按成交量降序排列，取第一名——它就是当前的主力合约。
④ 有什么坑
| 坑 | 原因 | 后果 |
|---|---|---|
| 切换太早 | 交割周前 5 天就切到下月，但当月合约还有充足流动性 | 提前用了非主力数据，浪费了当月合约最后几天的活跃成交 |
| 切换太晚 | 交割周前两天还在用当月合约 | 回测里混入了个位数成交的噪音 K 线 |
| 只用成交量或只用持仓量 | 不同数据商对“主力”的定义不同，自己的代码里两套标准混用 | 同一个合约，A 指标说它是主力，B 指标说不是 |
⑤ 怎么优化
用成交量阈值 + 时间规则联合判断：当成交量连续 3 天下滑超过 40%，且进入交割周前 5 天窗口，触发切换。这就像数据库主从切换——主库挂了从库顶上，切太早会丢数据、切太晚服务已经挂了。主力合约切换也是同一个道理：时机是最贵的变量。
连续合约拼接：前复权因子缝合不同合约的价格断点
辅助概念，两步讲清。
第一步，什么是连续合约。 把不同月份的主力合约按时间顺序串成一条连续的 K 线序列。IF 在 2025 年 12 月的主力是 IF2512，2026 年 1 月换到了 IF2601——连续合约就是把这两段 K 线在切换点接起来，让你回测时看到的是一条“IF 主力”的长周期曲线，而不是零零散散的月度合约片段。
第二步，怎么拼接。 两个合约在切换点会有价差。IF2512 最后一天的收盘价是 3950，IF2601 第一天的开盘价是 3970，这 20 个点的跳空不是策略赚的钱，是换月带来的价差。拼接时用前复权因子——把切换点之前的 K 线价格按价差比例调整，消除跳空。这和视频流无缝切换一个道理：两个流之间需要 overlap 一段做渐变，硬切会有黑屏。
代码实操：从中金所全量品种到连续合约序列
下面三段代码可以直接跑，唯一依赖是 requests、sqlite3 和 numpy。
价值承诺：这套代码能帮你从“手工维护换月日历表”直接跳到自动识别主力合约+拼接连续 K 线，每次换月省 30 分钟。
Step 1：全量期货品种枚举 + 按 IF/IC/IH/IM 分组
import os
import time
import requests
from typing import List, Dict
API_KEY = os.getenv("TICKDB_API_KEY") # 绝不硬编码密钥
BASE_URL = "https://api.tickdb.ai/v1"
HEADERS = {"X-API-Key": API_KEY}
# 中金所股指期货品种前缀（Kiro 实测确认：IC2606 / IF2606 / IH2606 / IM2606）
INDEX_FUTURES_PREFIXES = ["IF", "IC", "IH", "IM"]
def fetch_futures_symbols() -> Dict[str, List[str]]:
"""
从全量品种中筛选中金所股指期货，按 IF/IC/IH/IM 分组返回。
品种代码格式：IF2606（不带交易所后缀，Kiro 实测已验证正确）
type=futures 精确筛选期货品种（Kiro 实测确认：indices 和 stock 均无法返回期货）
"""
url = f"{BASE_URL}/symbols/available"
backoff = 1
grouped = {p: [] for p in INDEX_FUTURES_PREFIXES}
page = 0
while True:
try:
params = {
"market": "CN",
"type": "futures", # 精确筛选期货（Kiro 实测确认）
"limit": 500,
"offset": page * 500
}
resp = requests.get(url, headers=HEADERS, params=params, timeout=10)
data = resp.json()
if data["code"] == 3001: # 限流，优先读 Retry-After
retry_after = resp.headers.get("Retry-After")
wait = int(retry_after) if retry_after else backoff
time.sleep(wait)
backoff = min(backoff * 2, 8)
continue
if data["code"] == 1001: # 权限或参数错误，阻断报错
raise RuntimeError(f"API Error 1001: {data.get('message')}")
if data["code"] != 0:
raise RuntimeError(f"Unexpected error {data['code']}: {data.get('message')}")
batch = data["data"]["products"]
for item in batch:
sym = item["symbol"]
for prefix in INDEX_FUTURES_PREFIXES:
# 匹配如 IF2606、IC2606（Kiro 实测：正确格式，不带后缀）
if sym.startswith(prefix) and len(sym) == len(prefix) + 4:
grouped[prefix].append(sym)
if len(batch) < 500:
break
page += 1
backoff = 1
except requests.exceptions.Timeout:
time.sleep(1)
except Exception as e:
print(f"拉取中断: {e}")
break
return grouped
核心是
type=futures精确筛选 + 品种前缀匹配，不是请求速度。IF2606这种不带后缀的格式是 Kiro 实测确认的正确格式，千万别写成IF2406.CFE。这和微服务里统一资源 ID一个道理——同一个对象，A 服务用user_123，B 服务用uid:123，不通配就匹配不上。品种前缀规则写死一次，所有策略共用。
Step 2：主力合约识别 + 成交量监控
from datetime import datetime, timedelta
def identify_main_contracts(grouped: Dict[str, List[str]]) -> Dict[str, Dict]:
"""
对每个品种分组，拉取实时成交量，识别主力合约。
切换条件：成交量连续 3 天下降超 40% + 进入交割周前 5 天窗口。
"""
url = f"{BASE_URL}/market/ticker"
backoff = 1
result = {}
for prefix, symbols in grouped.items():
if not symbols:
continue
# 拉取该分组内所有合约的 ticker 快照（symbols 复数参数）
try:
params = {"symbols": ",".join(symbols)} # ticker 用 symbols 复数
resp = requests.get(url, headers=HEADERS, params=params, timeout=10)
data = resp.json()
if data["code"] == 3001:
retry_after = resp.headers.get("Retry-After")
wait = int(retry_after) if retry_after else backoff
time.sleep(wait)
backoff = min(backoff * 2, 8)
elif data["code"] == 1001:
continue
elif data["code"] != 0:
continue
# 按成交量（volume_24h）降序排列，第一名是当前主力
tickers = sorted(
data.get("data", []),
key=lambda x: float(x.get("volume_24h", 0)),
reverse=True
)
if not tickers:
continue
main = tickers[0]
main_symbol = main["symbol"]
main_volume = float(main.get("volume_24h", 0))
main_price = float(main.get("last_price", 0))
# 取第二名作为次主力（备用切换目标）
secondary = tickers[1] if len(tickers) > 1 else None
secondary_symbol = secondary["symbol"] if secondary else None
result[prefix] = {
"main_symbol": main_symbol, # 如 IF2606
"main_volume": main_volume,
"main_price": main_price,
"secondary_symbol": secondary_symbol,
"all_contracts": symbols,
}
except Exception as e:
print(f"拉取 {prefix} ticker 失败: {e}")
continue
return result
核心是按
volume_24h排序取第一名，不是简单查价格。 ticker 接口用symbols复数参数一次拉多只合约，省掉了逐只调用的建连开销。last_price和volume_24h是 ticker 的专属字段，不要和 kline 的close/volume混淆。这就像数据库主从切换——主库挂了从库顶上，切换前必须确认从库的同步延迟（成交量够不够），时机选错就会丢数据。
Step 3：连续合约 K 线拉取 + 拼接
| 参数 | 正确写法 | 错误写法 | 说明 |
|---|---|---|---|
| 品种参数 | symbol= | symbols= | kline 用单数 |
| 周期参数 | interval="1d" | period="1d" | API 文档规范 |
| 时间字段 | time（毫秒 UTC） | timestamp | ticker 才用 timestamp |
| 成交量字段 | volume | volume_24h | kline 是该周期成交量 |
| 收盘价字段 | close | last_price | kline 返回该周期收盘价 |
import numpy as np
def fetch_continuous_kline(
contracts: Dict[str, Dict],
start_date: str,
end_date: str,
interval: str = "1d"
) -> Dict[str, List[Dict]]:
"""
拉取每个品种主力合约的 K 线，在切换点做前复权拼接，输出连续合约序列。
"""
url = f"{BASE_URL}/market/kline"
backoff = 1
result = {}
for prefix, info in contracts.items():
main_sym = info["main_symbol"]
secondary_sym = info["secondary_symbol"]
# 拉取主力合约 K 线
try:
params = {
"symbol": main_sym, # 单数
"interval": interval,
"start_time": start_date,
"end_time": end_date
}
resp = requests.get(url, headers=HEADERS, params=params, timeout=10)
data = resp.json()
if data["code"] == 3001:
retry_after = resp.headers.get("Retry-After")
wait = int(retry_after) if retry_after else backoff
time.sleep(wait)
backoff = min(backoff * 2, 8)
elif data["code"] == 1001:
continue
elif data["code"] != 0:
continue
klines = data["data"]["klines"]
processed = []
for k in klines:
processed.append({
"time": k["time"], # 毫秒 UTC
"open": float(k["open"]),
"high": float(k["high"]),
"low": float(k["low"]),
"close": float(k["close"]),
"volume": float(k.get("volume", 0)),
"symbol": main_sym,
})
# 如果有次主力合约，检查是否需要切换
# 切换条件：进入交割周前 5 天窗口 + 主力成交量连续下降
# 简化版：取主力最后 3 天的成交量均值，若低于次主力的 60%，触发切换
if secondary_sym and len(processed) >= 3:
last_3_vol = np.mean([p["volume"] for p in processed[-3:]])
# 拉次主力最近 3 天 K 线做对比
params["symbol"] = secondary_sym
resp2 = requests.get(url, headers=HEADERS, params=params, timeout=10)
if resp2.status_code == 200:
data2 = resp2.json()
if data2.get("code") == 0:
sec_klines = data2["data"]["klines"]
if len(sec_klines) >= 3:
sec_last_3_vol = np.mean(
[float(k.get("volume", 0)) for k in sec_klines[-3:]]
)
# 如果次主力成交量已超过主力的 60%，触发切换
if sec_last_3_vol > last_3_vol * 0.6:
# 前复权拼接：计算价差比例，调整历史价格
if len(sec_klines) > 0:
main_last_close = processed[-1]["close"]
sec_first_close = float(sec_klines[0]["close"])
adjust_ratio = sec_first_close / main_last_close if main_last_close > 0 else 1.0
# 将主力合约历史价格按比例调整
for p in processed:
p["open"] = round(p["open"] * adjust_ratio, 4)
p["high"] = round(p["high"] * adjust_ratio, 4)
p["low"] = round(p["low"] * adjust_ratio, 4)
p["close"] = round(p["close"] * adjust_ratio, 4)
# 拼接次主力 K 线
for k in sec_klines:
processed.append({
"time": k["time"],
"open": float(k["open"]),
"high": float(k["high"]),
"low": float(k["low"]),
"close": float(k["close"]),
"volume": float(k.get("volume", 0)),
"symbol": secondary_sym,
})
result[prefix] = processed
backoff = 1
except Exception as e:
print(f"拉取 {main_sym} K 线失败: {e}")
continue
return result
核心是成交量阈值触发切换 + 前复权拼接，不是简单拉 K 线。 切换时机判断用“次主力成交量超主力 60%”作为量化标准，避免人肉判断。拼接时的价差调整是前复权的简化版——两个合约间的价差在切换点被复权因子抹平，连续合约曲线就不会出现跳空。
你真正在维护的，是一张手工换月日历表
没有统一 API 时，你面对的困境：
| 问题类型 | 具体表现 | 维护成本 |
|---|---|---|
| 品种列表散落各处 | 中金所 105 个期货品种，列表在交易所官网和数据商文档里分别维护 | 每次新品种上市手工补一行 |
| 主力判断标准不统一 | 有的数据商用成交量、有的用持仓量，自己的代码里两套逻辑混用 | 换月时点每次都要人工核对日历 |
| 合约代码格式混乱 | 有的源用 IF2406.CFE，有的用 IF2406，有的用 IF06 | 每个数据源写一个 parser |
| 限流规则不同 | 不同数据商的频率限制不同，错误码要分别处理 | 多套异常处理代码 |
TickDB 出现在这个背景下不意外：一个 REST + WebSocket 长连接，覆盖中金所股指期货全品种——Kiro 实测确认了 IF2606、IC2606、IH2606、IM2606 四个品种，统一格式、统一成交量字段（ticker volume_24h / kline volume）、统一鉴权。你不用再维护手工换月日历表，主力合约识别逻辑写一次、永久复用。
| TickDB 统一了什么 | 你省掉了什么 |
|---|---|
统一品种代码格式（IF2606） | 多源格式 parser |
统一成交量字段（ticker volume_24h / kline volume） | 多套字段映射表 |
统一鉴权方式（X-API-Key） | 多套 Token 管理 |
| 统一错误码（3001/1001） | 多套限流处理逻辑 |
接口文档和字段映射关系在 https://docs.tickdb.ai 开源可查。需要更自动化的主力合约监控，还可以走 MCP 工具链（https://mcp.tickdb.ai），把行情查询封装成 Agent 可调用的服务。
你的回测曲线里藏着多少交割周的流动性枯竭 K 线？
一个朋友给我看过他的 IF 回测曲线，年化 18%，最大回撤 11%，看起来漂亮。我让他打开 2024 年 6 月交割周的逐笔明细——周三上午 10:14 到 10:19，IF2406 只成交了 15 手，K 线上却记录了 4 次“突破信号”。
他的策略在那 5 分钟里连开了 3 次多单。实盘里，那 3 次开单根本成交不了——对手盘早就撤了。
回测年化 18%，实盘只剩 11%？别急着怀疑过拟合。先检查你的主力合约映射表，是不是还停在半年前。你上一次验证换月逻辑，是什么时候？
📡 数据由 TickDB.ai 提供
通过 TickDB API 获取实时行情数据
一个 API 接入外汇、加密货币、美股、港股、A股、贵金属和全球指数的实时行情。支持 WebSocket 低延迟推送，免费开始使用。
免费领取 API Key查看 API 文档