supermind logo
同花顺logo

ETF行情API接入踩坑记：从报错到跑通的七个问题

用户头像Fxdund
2026-09-20 发布

最近ETF资金流向很活跃，不少人开始做ETF相关的工具。我整理了一下自己接ETF行情API时被问到最多、也是踩得最多的七个问题，每个都附了当时的排查过程和最终能用的代码。

Q1：调 /fund/quote 返回404，路径是不是写错了？

排查过程：我第一次写的是 /funds/quote，直接404。翻了文档才发现，这个接口所有产品都是单数路径——股票是 /stock/，期货是 /future/，基金是 /fund/。别想当然地加s。

正确写法：

import requests

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

# 注意是 /fund/ 不是 /funds/
resp = requests.get(
    f"{API_BASE}/fund/quote",
    headers=headers,
    params={"region": "US", "code": "QQQ"}
)
print(resp.json())

Q2：返回的data是空的，参数不对吗？

排查过程：我传了 region=US、code=QQQ，但返回 data: null。一开始以为是token没生效，后来发现是代码格式问题——ETF代码要大写，而且有些ETF的代码跟你想的不一样。

怎么确认有哪些代码可用：先从最主流的开始试，QQQ（纳指100）、SPY（标普500）、IWM（罗素2000）、GLD（黄金ETF），这些肯定有。港股ETF用数字代码，比如 2800（盈富基金），这时候region要传 HK。

# 验证代码是否有效
def check_code(region, code):
    resp = requests.get(
        f"{API_BASE}/fund/quote",
        headers=headers,
        params={"region": region, "code": code}
    )
    data = resp.json().get("data")
    if data:
        print(f"{code}: 最新价 {data['ld']}")
    else:
        print(f"{code}: 无数据")

check_code("US", "QQQ")   # 应该有数据
check_code("US", "SPY")   # 应该有数据
check_code("HK", "2800")  # 港股盈富基金

Q3：返回字段全是缩写，ld/chp/o/h/l到底是什么意思？

排查过程：第一次打印返回数据，看到 ld: 613.7、chp: -1.9，完全不知道是什么。翻文档整理了一下：

字段 含义 说明
s 标的代码 比如 QQQ
ld 最新价 last price
o 开盘价 open
h 最高价 high
l 最低价 low
p 前收盘价 previous close
ch 涨跌额 change
chp 涨跌幅% change percent
v 成交量 volume
tu 成交额 turnover
ts 交易状态 0正常 1停牌 2退市 3熔断

一个小坑：别自己拿最新价跟开盘价算涨跌幅，那样算出来的是日内振幅。涨跌幅直接用 chp 字段，服务端已经算好了。

Q4：WebSocket连上了，但订阅一直报 "cannot be resolved action"

排查过程：我以为是订阅格式错了，调了半天params和types。后来发现根本不是格式问题——是我在连接刚建立的时候就发订阅了，这时候鉴权还没完成。

正确的顺序：连接 → 等 resAc:"auth" 成功消息 → 再发订阅。

import websocket
import json
import time
import threading

WS_URL = "wss://api.itick.org/fund"
TOKEN = "your_token_here"
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
        print("鉴权成功，开始订阅")
        ws.send(json.dumps({
            "ac": "subscribe",
            "params": "QQQ$US,SPY$US",
            "types": "quote"
        }))
        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']}%)")

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

Q5：WebSocket连接一分钟就断了，为什么？

排查过程：跑了没两分钟连接就关了，一开始以为是网络问题。后来才知道这个服务要求30秒发一次心跳，超过1分钟不发就踢人。

加个心跳线程：

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

# 鉴权成功后启动
threading.Thread(target=heartbeat, args=(ws,), daemon=True).start()

Q6：K线数据拉出来时间戳不对，怎么转成日期？

排查过程：返回的 t 字段是毫秒时间戳，比如 1765573199000。直接当秒数处理会得到一个很远的未来时间，必须先除以1000。

from datetime import datetime

# 正确的转换方式
kline_data = [
    {"t": 1765573199000, "c": 613.7},
    {"t": 1765486799000, "c": 615.2},
]

for bar in kline_data:
    # 注意：除以1000转成秒
    dt = datetime.fromtimestamp(bar["t"] / 1000)
    print(f"{dt.strftime('%Y-%m-%d')}: 收盘 {bar['c']}")

另外，K线周期 kType 的编码也容易记混：

  • 1 = 1分钟
  • 2 = 5分钟
  • 5 = 1小时
  • 8 = 日K
  • 9 = 周K
  • 10 = 月K

Q7：想同时监控好几只ETF，要开几个WebSocket？

排查过程：我一开始给每只ETF开一个连接，结果发现完全没必要——单个连接最多支持订阅500个标的，用逗号把代码拼起来就行。

def subscribe_multiple(ws):
    # 多个标的用逗号分隔
    ws.send(json.dumps({
        "ac": "subscribe",
        "params": "SPY$US,QQQ$US,IWM$US,GLD$US",
        "types": "quote"
    }))

订阅格式是 代码$市场，比如 SPY$US。港股ETF就是 2800$HK。

最后整理一个能跑的最小脚本

把上面这些坑都避开，最终的最小可用版本是这样：

import json, time, requests, threading, websocket

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

watchlist = ["SPY", "QQQ", "GLD"]

# 启动时先拉快照
print("=== ETF快照 ===")
for code in watchlist:
    r = requests.get(f"{API_BASE}/fund/quote",
                     headers=headers,
                     params={"region": "US", "code": code}).json()["data"]
    print(f"  {code}: {r['ld']} ({r['chp']}%)")

# WebSocket实时更新
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
        ws.send(json.dumps({
            "ac": "subscribe",
            "params": ",".join(f"{c}$US" for c in watchlist),
            "types": "quote"
        }))
        threading.Thread(target=heartbeat, args=(ws,), daemon=True).start()
        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()

这个脚本避开了上面提到的所有坑：路径用单数 /fund/、WebSocket先等鉴权再订阅、有心跳保活、多个标的用逗号拼在一个订阅里。

最近ETF资金流向很活跃，做个自己的监控工具比来回切行情软件方便多了。

参考文档：https://docs.itick.org/websocket/fund
GitHub：https://github.com/itick-org/

评论