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

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

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

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

正确写法：

import requests

API_BASE = "https://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，完全不知道是什么。翻文档整理了一下：

一个小坑：别自己拿最新价跟开盘价算涨跌幅，那样算出来的是日内振幅。涨跌幅直接用 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 = "https://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资金流向很活跃，做个自己的监控工具比来回切行情软件方便多了。

参考文档：docs.itick.org/websocket/fund

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

本作品采用《CC 协议》，转载必须注明作者和本文链接