摘要：本文记录了在中金所股指期货品种上实现主力合约自动识别与连续合约拼接的完整工程方案，按成交量排序确定主力、用成交量阈值联合时间规则触发切换、通过前复权因子缝合不同合约的价格断点。

一、交割周周三上午的那几笔成交

交割周周三上午 10:14，IF2406 一分钟只成交了 3 手，价格跳动 4 个点。

策略信号在这个时间点触发了“突破买入”——但那 4 个点的跳动不是趋势，是非主力合约的流动性枯竭。更麻烦的是，回测里混入了 3 天这种非主力数据，年化收益高估了 4 个百分点。

这是我在维护手工换月日历表的第三年，第一次下决心把这个流程自动化。手工维护一张 IF/IC/IH 换月日历表的量化开发者，迟早会踩进同一个坑：主力合约切换的时机错了，整个回测曲线就是噪音拼出来的。

二、主力合约识别：为什么成交量比持仓量更适合做切换信号

这是量化回测里最容易高估收益的坑。拆成五步看。

是什么。 主力合约，是指同一品种下成交量最大、流动性最好的那个合约。中金所每个股指期货品种同时挂牌当月、下月及随后两个季月共四个合约。对 IF 来说，同一时间有 IF2606、IF2607、IF2609、IF2612 四只在交易。主力合约识别，就是在这四只里自动挑出成交量最大的那一只。

为什么必须用主力合约做回测。 非主力合约的流动性会断崖式萎缩。交割周前 3 天，当月合约的成交从几千手骤降到几十手甚至个位数。一分钟 K 线里出现几个点的跳动，不是趋势信号，是买卖盘枯竭后的价格跳空。回测引擎无法区分“真实的趋势突破”和“流动性枯竭导致的噪音跳动”，会把后者也当成有效信号计入绩效，直接拉高年化收益。

怎么用。 逻辑三步走。先从中金所全量品种里筛出 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。

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：主力合约识别

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

        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,
                "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 线拉取与拼接

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"]

        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"],
                    "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,
                })

            # 检查是否需要切换至次主力
            if secondary_sym and len(processed) >= 3:
                last_3_vol = np.mean([p["volume"] for p in processed[-3:]])
                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:]]
                            )
                            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)

                                    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 的环境里做期货策略，日常维护的不是策略逻辑，而是一张换月日历表：

TickDB 的出现正好解决了这个问题。Kiro 实测确认了 IF2606、IC2606、IH2606、IM2606 四个品种的代码格式和字段名体系——统一 REST 与 WebSocket 接口、统一品种代码格式、统一成交量字段（ticker 的 volume_24h 和 kline 的 volume）、统一鉴权方式、统一错误码语义。主力合约识别逻辑写一次，永久复用。

接口文档和字段映射关系可在 TickDB 官方文档中查阅，需更自动化的主力合约监控还可以走 MCP 工具链（https://mcp.tickdb.ai）。

六、你的回测曲线里藏着多少交割周的流动性枯竭 K 线

一个朋友给我看过他的 IF 回测曲线，年化 18%，最大回撤 11%，看起来漂亮。我让他打开 2024 年 6 月交割周的逐笔明细——周三上午 10:14 到 10:19，IF2406 只成交了 15 手，K 线上却记录了 4 次“突破信号”。

他的策略在那 5 分钟里连开了 3 次多单。实盘里，那 3 次开单根本成交不了——对手盘早就撤了。

回测年化 18%，实盘只剩 11%？别急着怀疑过拟合。先检查你的主力合约映射表，是不是还停在半年前。你上一次验证换月逻辑，是什么时候？欢迎在评论区聊聊你踩过的换月坑。

📡 数据由 TickDB.ai 提供

本文不构成任何投资建议。文中期货品种代码 IF2606、IC2606、IH2606、IM2606 仅为代码示例。API 端点与字段定义以 TickDB 官方文档为准，可搜索查阅。