supermind logo
同花顺logo

期货行情API接入：股指、黄金、原油多品种数据管道实战

用户头像Fxdund
2026-10-31 发布

股票和外汇的数据接口，拉回来就能用。期货不一样——同样是"行情"，期货多了一层合约的概念：同一品种有主力合约、连续合约，还分不同到期月份；交易所有好几家，还有夜盘。第一次做期货监控的时候，我以为就是多调几个接口的事，结果被合约体系和换月问题折腾得不轻。这篇把我搭的一套多品种期货数据管道完整记录下来：合约代码怎么认、批量快照怎么拿、实时推送怎么接、主力换月怎么处理。数据源用的 iTick，token 去它官网 itick.org 申请，REST 和 WebSocket 共用一套。

期货合约体系：先认识代码，再谈接入

期货跟股票最大的区别就是"品种"和"合约"是两个概念。股票一个代码就是一个公司，期货一个品种下面有多个合约，接口里用"品种代码"来标识：

品种 代码 市场 region 说明
股指期货（沪深300） IF 中国 CN 中金所
黄金 GC 美国 US COMEX
白银 SI 美国 US COMEX
原油 CL 美国 US NYMEX
大豆 ZS 美国 US CBOT
纳指期货 NQ 美国 US CME
标普期货 ES 美国 US CME

接口统一用 region + code 定位一个品种。认准这几个主流代码，后面调接口就不会发懵。

REST批量快照：一个请求拿全品种

监控面板启动时要初始化所有品种的状态，用批量报价接口一次拉全，别一个个循环调：

import requests

API_BASE = "//api.itick.org"
headers = {"accept": "application/json", "token": "your_token_here"}

def fetch_futures_snapshots(codes="GC,SI,CL,ZS,NQ,ES,IF"):
    # /future/quotes 批量报价，codes 用英文逗号分隔
    # 返回 dict，key 是品种代码
    resp = requests.get(
        f"{API_BASE}/future/quotes",
        headers=headers,
        params={"region": "US", "codes": codes},
    )
    return resp.json()["data"]

snap = fetch_futures_snapshots()
for symbol, q in snap.items():
    print(symbol, q["ld"], q["chp"], q["ts"])

返回的每个品种都是一个报价对象，字段 s/ld/o/p/h/l/t/v/tu/ts/ch/chp：ld 最新价、p 前日收盘价、ts 交易状态、ch/chp 涨跌。有几个点值得注意：

  • 返回是 dict 不是数组，key 是品种代码，遍历时用 for symbol, q in snap.items()
  • p 前收字段是期货报价特有的，算涨跌额直接用 ld - p
  • 批量接口的参数是 codes（复数），写 code 会报参数错误
  • ts 交易状态：期货有夜盘，判断行情是否有效前先看这个字段

历史K线：kType 一个编码走所有周期

回测和画图需要历史数据，期货K线接口的参数跟股票一样：region + code + kType + limit。周期用数字编码，1分钟到月线都有：

def fetch_future_klines(code, region="US", ktype=8, limit=200):
    resp = requests.get(
        f"{API_BASE}/future/kline",
        headers=headers,
        params={
            "region": region,
            "code": code,
            "kType": ktype,     # 1=1分钟 2=5分钟 5=1小时 8=日K 9=周K 10=月K
            "limit": limit,     # 单次最大500
        },
    )
    rows = resp.json()["data"]
    # 返回数组，每条 t/o/h/l/c/v/tu，时间戳毫秒
    return rows

gold_daily = fetch_future_klines("GC", ktype=8, limit=300)
print(gold_daily[-1]["c"])  # 黄金最新一根日K收盘

这里有个期货特有的坑：连续合约和具体月份合约的K线不一样。接口返回的一般是主力/连续合约的标准化数据，做回测时如果自己拼接具体月份合约的数据，换月那天会出现跳空，别把它当成真实的行情波动。

WebSocket实时推送：一条连接盯多个品种

盘中实时监控用 WebSocket，流程还是那四步：连上 → 等鉴权成功 → 订阅 → 心跳。一条连接最多订500个标的，主流品种一条连接全搞定：

import json
import threading
import time
import websocket

WS_URL = "wss://api.itick.org/future"
TOKEN = "your_token_here"

authenticated = False
heartbeat_alive = True

def on_message(ws, message):
    global authenticated
    payload = json.loads(message)

    if payload.get("resAc") == "auth" and payload.get("code") == 1 and not authenticated:
        authenticated = True
        ws.send(json.dumps({
            "ac": "subscribe",
            "params": "GC$US,SI$US,CL$US,NQ$US,ES$US,IF$CN",
            "types": "quote"
        }))
        threading.Thread(target=heartbeat, args=(ws,), daemon=True).start()
        return

    if payload.get("resAc") == "pong":
        return

    data = payload.get("data") or {}
    if data.get("type") == "quote":
        # 推送字段与REST一致，s/ld/o/h/l/v/tu
        print(data["s"], data["ld"], data["t"])

def heartbeat(ws):
    while heartbeat_alive:
        time.sleep(30)
        ws.send(json.dumps({"ac": "ping", "params": str(int(time.time() * 1000))}))

ws = websocket.WebSocketApp(WS_URL, header=[f"token: {TOKEN}"], on_message=on_message)
ws.run_forever()

订阅格式 GC$US 是"品种代码$市场代码"，多品种逗号分隔。这里有个容易忽略的点：IF 是中金所品种，region 是 CN，不是 US——写订阅参数时别想当然全填 US。

主力合约换月：期货监控绕不开的坑

股票没有"换月"概念，期货有。随着临近交割，主力合约会切换到下一个月份，这时候连续合约的K线和报价会出现一个跳变。做多品种监控要注意：

  • 换月跳空不是行情波动：策略判断涨跌时要排除换月带来的价格跳变，不然会误触发信号
  • 别在代码里写死合约月份：用连续合约代码，让数据源帮你处理主力切换
  • 持仓量变化是换月的信号：成交量、持仓量明显转移时，多半是在换月，关注 ts 交易状态和价格是否异常

组合示例：多品种期货监控面板

把快照和实时拼起来，就是一个可用的监控面板骨架：

import json, threading, time, requests, websocket

API_BASE = "//api.itick.org"
WS_URL = "wss://api.itick.org/future"
TOKEN = "your_token_here"
headers = {"accept": "application/json", "token": TOKEN}

WATCHLIST = "GC,SI,CL,ZS,NQ,ES"

# 启动快照
print("=== 期货快照 ===")
snap = requests.get(f"{API_BASE}/future/quotes",
                    headers=headers,
                    params={"region": "US", "codes": WATCHLIST}).json()["data"]
for symbol, q in snap.items():
    sign = "+" if q["chp"] >= 0 else ""
    print(f"  {symbol}: {q['ld']} ({sign}{q['chp']}%)")

# 实时推送
authenticated = False
def on_message(ws, message):
    global authenticated
    payload = json.loads(message)
    if payload.get("resAc") == "auth" and payload.get("code") == 1 and not authenticated:
        authenticated = True
        params = ",".join(f"{c}$US" for c in WATCHLIST.split(","))
        ws.send(json.dumps({"ac": "subscribe", "params": params, "types": "quote"}))
        threading.Thread(target=heartbeat, args=(ws,), daemon=True).start()
        return
    if payload.get("resAc") == "pong":
        return
    data = payload.get("data") or {}
    if data.get("type") == "quote":
        print(f"[实时] {data['s']}: {data['ld']} ({data['chp']}%)")

def heartbeat(ws):
    while True:
        time.sleep(30)
        ws.send(json.dumps({"ac": "ping", "params": str(int(time.time() * 1000))}))

ws = websocket.WebSocketApp(WS_URL, header=[f"token: {TOKEN}"], on_message=on_message)
ws.run_forever()

这个脚本启动时拉一次全品种快照，盘中WebSocket实时更新，做监控或者告警的底座足够用了。

踩坑清单速查

  • 品种代码 ≠ 合约代码：接口用品种代码（GC/CL/IF），别拿具体月份合约号来调
  • IF 是 CN 市场：中金所品种 region 填 CN，别全填 US
  • 批量接口参数是 codes（复数），返回 dict 以品种代码为 key
  • 期货报价有 p 前收字段，涨跌额 = 最新价 − 前收
  • 连续合约换月有跳空：回测和信号判断时要排除换月跳变
  • 时间戳毫秒，判断夜盘时段先统一时区
  • WS 先等鉴权成功再订阅，心跳 30 秒一次不能省

小结

多品种期货数据管道，核心就三件事：认识合约体系、快照走批量、实时走推送。合约代码认准 region + code，快照用 /future/quotes 一次拿全，实时用 WebSocket 一条连接盯多个品种，换月的坑提前心里有数，这套管道就能稳定跑起来。接口参数和返回字段的完整说明，写代码时对照官方文档最稳妥。

GitHub：https://github.com/itick-org/

评论