iTick API 实时行情接入实战：用免费档搭一个多市场股票预警机器人

#tutorial #api #python

前阵子朋友让我帮他盯几只港美股，说是开盘时间他总在开会，错过了好几次加减仓的窗口。我嘴上答应，心里其实有点犯怵——总不能每五分钟刷一次行情软件吧，我也有自己的事。

于是花了一个周末，用 iTick 的免费档搭了个简单的行情预警机器人，WebSocket 长连接收实时推送，价格一碰到我设的阈值就推消息到手机上。这篇文章把整套接入过程写下来，代码都是能直接跑的，给想自己折腾实时行情接入的朋友当个参考。

为什么用 WebSocket 而不是轮询

一开始我也想过用 REST 接口定时去拉，写起来简单嘛。但算了一下账就放弃了：iTick 免费档 REST 接口是 5 次/分钟，如果盯 20 只股票，每只每 5 分钟拉一次就得 4 次/分钟，一个预警机器人就快把配额吃光了，更别说以后加标的。

WebSocket 是另一套逻辑，建一条长连接，服务端把行情主动推过来，你只需要订阅感兴趣的标的，不占 REST 的调用次数。免费档虽然 WebSocket 只给 1 条连接、3 个订阅，但对个人盯盘来说，3 个订阅刚好够我盯 3 只重点股票，REST 的 5 次/分钟配额还能留着做别的查询。

第一步：拉一把历史数据做参考

预警逻辑需要先有个基准价，比如昨收价或者最近 N 根 K 线的均价。我先用 REST 接口把腾讯控股（700）的日线拉出来，算一下过去 20 天的简单移动平均，拿这个当判断趋势的参考线。

iTick 的鉴权很简单，注册后在控制台生成一个 token，请求时放 header 里就行：

import requests

url = "https://api.itick.org/stock/klines?region=HK&codes=700&kType=8&limit=20"
headers = {
    "accept": "application/json",
    "token": "你的token"
}

resp = requests.get(url, headers=headers).json()
print(resp["code"], resp["msg"])

# 取出20根日线收盘价
closes = [k["c"] for k in resp["data"]["700"]]
sma20 = sum(closes) / len(closes)
print(f"700 最近20日收盘均价: {sma20:.2f}")

返回的每根 K 线是标准的 OHLCV 结构，o/h/l/c 是开高低收，v 是成交量，t 是毫秒时间戳。kType=8 是日线，想拉分钟线就改 1（1分钟）、2（5分钟）这些，映射关系文档里写得很清楚。

第二步：WebSocket 订阅实时行情

历史参考线有了，接下来接实时推送。iTick 官方有 Python SDK，装一下：

pip install itick-sdk

然后建连接、注册回调、订阅标的：

from itick.sdk import Client
import time

token = "你的token"
client = Client(token)

def on_message(message):
    # 收到的每一条实时行情都走这里
    print(f"[{time.strftime('%H:%M:%S')}] {message}")

def on_error(error):
    print(f"连接异常: {error}")

client.set_message_handler(on_message)
client.set_error_handler(on_error)

# 连接股票 WebSocket
client.connect_stock_websocket()

# 订阅腾讯控股和阿里巴巴的报价
client.send_websocket_message(
    '{"action": "subscribe", "region": "HK", "codes": ["700", "9988"], "types": ["quote"]}'
)

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    client.close_websocket()

SDK 内部把心跳和重连都处理好了：默认 30 秒发一次心跳，网络断开后按 5 秒间隔自动重连，最多重试 10 次，重连成功后订阅关系会自动恢复。这一点省了我很多事，自己手写 WebSocket 的话，心跳、断线重连、订阅恢复这三块至少要折腾大半天。

推送回来的 quote 消息里有最新价、涨跌幅、买一卖一这些字段，够做基础的盘中判断了。

第三步：加阈值预警逻辑

只打印消息没意义，得让它按我的规则提醒我。我在回调里加了简单的判断：最新价相对参考线（刚才算的 SMA20）的偏离超过 ±3% 就告警，并且用状态去重，同一只股票在告警解除前不重复提醒。

from itick.sdk import Client
import time
import json

token = "你的token"
client = Client(token)

# 参考价：从REST拿的SMA20
REF_PRICE = {"700": 535.0, "9988": 140.0}
THRESHOLD = 0.03  # 3%
alerted = {}  # 去重：记录每只股票是否在告警中

def on_message(message):
    try:
        data = json.loads(message)
    except Exception:
        return

    # 只处理实时报价
    if data.get("type") != "quote":
        return

    code = data.get("code")
    last = data.get("last", data.get("ld"))
    if not code or last is None:
        return

    ref = REF_PRICE.get(code)
    if not ref:
        return

    pct = (last - ref) / ref
    if abs(pct) >= THRESHOLD and not alerted.get(code):
        alerted[code] = True
        direction = "突破" if pct > 0 else "跌破"
        print(f"⚠️ {code} {direction}参考线 {abs(pct)*100:.1f}%，最新价 {last}")
        # 这里可以接推送服务，比如 Server酱、企业微信机器人

    if abs(pct) < THRESHOLD * 0.5:
        alerted[code] = False  # 回到安全区间，解除告警

client.set_message_handler(on_message)

client.connect_stock_websocket()
client.send_websocket_message(
    '{"action": "subscribe", "region": "HK", "codes": ["700", "9988"], "types": ["quote"]}'
)

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    client.close_websocket()

告警触发后怎么通知自己，看个人习惯：想收到手机推送，可以在那个 print 的位置换成 Server酱或者企业微信群机器人的 webhook；想落库做后续分析，就写进 SQLite。逻辑都一样，改一行的事。

第四步：多市场怎么扩展

盯完港股，想顺手把美股也加上，不用另起一套代码。iTick 的接口对多市场是统一的，region 参数切换市场：US 是美股、HK 是港股、SH/SZ 是沪深。订阅消息里把 region 换掉就行：

# 订阅美股
client.send_websocket_message(
    '{"action": "subscribe", "region": "US", "codes": ["AAPL", "TSLA"], "types": ["quote"]}'
)

免费档限制是 1 条连接、3 个订阅，所以同一时间只能盯一个市场的 3 个标的。想同时盯港股和美股，要么升级套餐加连接数，要么按时间段切换——比如 A 股收盘后再切到美股。个人用的话，我觉得按时间段切换其实够用。

几个踩过的坑

整个接入不算复杂，但有几个地方当时卡了我一阵，写出来免得大家重复踩：

region 代码别记混：美股是 US，港股是 HK，一开始我写了个 HKG，返回空数据，排查了半天才发现是市场代码的问题。
订阅参数格式要对：不同接口版本对订阅消息的字段名要求不完全一样，我用的 SDK 版本里 codes 要传数组，传成字符串会静默失败，没有任何报错，容易误以为连上了。
告警要去重：实时行情一秒推好几条，不做状态去重的话，价格一过阈值手机能给你震到没电。
免费档 REST 有频率限制：5 次/分钟，写历史数据拉取的循环时记得加 time.sleep(12)，不然会被限流返回错误。
token 别写死在代码里：放环境变量，尤其是要把代码推到 GitHub 的时候，血的教训。

小结

这套预警机器人从零到跑通，大概花了我一个周末，其中大半时间花在研究订阅格式和调试告警去重上，纯接入的部分其实很快。WebSocket 收实时行情 + REST 拉历史参考线，两条路配合，个人盯盘、辅助决策完全够用。

如果你是刚开始接触实时行情接入，或者想给自家产品加个行情监控功能，建议先拿免费档把 WebSocket 这条链路跑通，重点看数据延迟和推送稳定性是否符合预期，再决定要不要上更高级的规格。详细的接口字段和订阅格式，可以翻 iTick 的官方文档，Python/Java/Go 的 SDK 都有现成的示例。

有问题欢迎在评论区交流，我踩过的坑，希望能帮你少走点弯路。