先说结论:我最近要跑一个港股行情小助手,盘中得盯实时报价,不能接受十五分钟延迟。网上搜了一圈,能直接用的资源少得可怜,免费的要么接口不稳定,要么数据字段缺东少西,找券商官方接口又怕流程麻烦。折腾下来,最后锁定的方案其实不复杂:用港股实时API服务,走 WebSocket 订阅行情。从建连到收到第一条报价数据,前后大概半小时。这篇就把我怎么找接口、怎么选协议、怎么解析数据,还有过程里踩的坑,一次性讲清楚。
1. 为什么要用港股实时API:延时行情真的很坑
1.1 延时行情和实时行情的差别
先说一个很容易被新手忽略的问题:港股普通行情默认是延时十五分钟的。什么意思?你在一些免费行情软件上看到的现价,实际是十五分钟前的价格。对于做短线、盯异动、跑量化策略的人来说,这种“看录像回放”式的行情根本没法用。一个简单类比:看实时行情像看直播,球进了你马上知道;看延时行情像看重播,等你看到的时候,盘面可能已经变了。
更重要的是,程序化交易对数据的“时间一致性”有硬性要求。如果你根据一个已经过时的价格去下单,可能买在最高点、卖在最低点。所以想做一个靠谱的行情工具,第一件事就是必须拿到交易所认可的实时数据源,而不是从某个网页接口里碰运气。
1.2 行情数据的常见来源,各自适合什么场景
我把市面上的港股行情获取方式做了个分类,方便你判断自己该走哪条路:
| 来源 | 数据延迟 | 获取难度 | 适合场景 |
|---|---|---|---|
| 券商自带客户端 | 实时 | 需要开户且有资金门槛 | 人工盯盘,不适合二次开发 |
| 开放平台 OpenAPI | 实时 | 注册开发者账号、创建应用、审核 | 个人程序化、量化小工具 |
| 财经网站公开接口 | 几秒到几十秒 | 免费但无授权,随时可能失效 | 学习、验证想法,不建议生产用 |
| 专业数据服务商 | 实时/深度 | 收费高,按年订阅 | 机构、高频交易 |
我自己的需求是长期跑一个小程序,需要稳定、实时、能编程调用,所以直接排除了免费公开接口。最终选择的是正规开放平台提供的行情 API。虽然需要注册和审核,但实际做下来发现没有想象中复杂,而且文档、示例都齐全,比去猜那些不公开的接口靠谱太多。
2. 快速找到合适API的思路与我的选型
2.1 行情API怎么选?我重点看这5个指标
找 API 不是看哪家有就选哪家,我用了一个筛选清单,先对所有候选做减法:
- 数据覆盖范围:是否包含港股主板、创业板、ETF 以及涡轮牛熊证。注意港股代码前缀,有的平台用
00700.HK,有的用HK.00700,这直接影响订阅格式。 - 实时性指标:官方宣称的延迟是多少,实际推送能不能到毫秒或秒级。这一点不能只看宣传,最好用测试账号量一下从 tick 到本地的时间差。
- 协议友好度:支持 REST 还是 WebSocket,有没有 Python SDK,示例代码是否完整。没有 SDK 的,后续所有协议细节都要自己造轮子,成本高不少。
- 鉴权复杂度:是 OAuth 拿 Token,还是简单的 AK/SK 签名。Token 有效期多长,过期了能不能自动刷新,这些都要提前确认。
- 免费额度与价格:个人开发者有没有免费试用,订阅股票数量上限是多少。我见过一些平台免费额度只给 20 只自选股,但对个人完全够用。
按这套标准,我最后能打的候选就不多了。市面上名头响的,要么不支持港股,要么价格离谱,要么文档写着“请联系销售”。真正适合个人开发者的就那么一两家。
2.2 两条路线:免费临时验证 vs 正规实时订阅
如果你的目的纯粹是先跑通一个 Demo,那可以走“免费接口快验路线”:找一个财经网站提供行情地址,拼接股票代码,发 HTTP 请求拿到 JSON,最多再加个定时轮询。优点是零成本、结果直观,缺点也明显:请求频率受限、连接不稳定、字段不一定全,而且别人接口一改,你的程序立刻歇菜。
更稳的是“正规订阅路线”:去开放平台注册应用,拿到 App Key 和 Secret,认证换取 Access Token,然后连 WebSocket 长连接,服务端主动推送。这条路前期需要多花一点时间看文档,但一旦跑通,稳定性和实时性都有保障。我因为要长期运行,直接选了后者,后面所有代码也基于这个模式。
2.3 我最后用的方案:基于WebSocket的行情推送
为什么不用 REST 轮询?因为行情是高频变化的,REST 需要你每隔几秒主动请求一次,延迟高不说,还会遭到限流;WebSocket 则是建立一条长连接,服务端有新的报价就往下推,延迟低、资源占用小,是行情订阅的主流方式。
整个链路很清晰:
- 注册应用,拿到
App Key和App Secret。 - 调用认证接口,获取
Access Token。 - 用 Token 建立 WebSocket 连接。
- 发送订阅消息,指定想要的股票代码。
- 之后持续接收服务端推送的行情消息。
下面我按这个链路给出可复现的代码逻辑。注意各家 API 的具体地址和报文格式有差异,你需要以自己的服务商文档为准,我会用占位符说明。
3. 手把手接入港股实时行情API
3.1 准备工作:申请接口权限与获取连接参数
在写代码前,先把三样东西准备好:
- 已创建的应用信息:App Key、App Secret。
- 行情服务地址:类似
wss://push-api.example.com/hk。 - 正确的股票代码格式:这个最容易出错,比如腾讯控股在平台里可能写成
HK.00700,而不是0700.HK。
获取访问令牌一般就是一个 POST 请求。参考逻辑:
import requests app_key = "your_app_key" app_secret = "your_app_secret" resp = requests.post("https://api.example.com/auth/token", json={ "app_key": app_key, "app_secret": app_secret }) token_data = resp.json() access_token = token_data["access_token"] print("Access Token:", access_token)拿到 Token 后,先别急着写主流程。我建议你打开平台自带的 WebSocket 调试工具,手动订阅一只股票,确认能收到数据后,再回到代码里,这样能把“接口问题”和“代码问题”隔离开。
3.2 核心代码:建立连接、订阅、接收推送
这里我用 Python 的websockets库实现,代码不长,但把核心流程都包含进去了:
import asyncio import json import websockets WS_URL = "wss://push-api.example.com/hk" TOKEN = "your_access_token" SYMBOLS = ["HK.00700", "HK.09988", "HK.03690"] async def receive_quote(): headers = {"Authorization": f"Bearer {TOKEN}"} async with websockets.connect(WS_URL, extra_headers=headers) as ws: subscribe_msg = { "action": "subscribe", "symbols": SYMBOLS } await ws.send(json.dumps(subscribe_msg)) print("订阅已发送:", SYMBOLS) while True: message = await ws.recv() print(message) if __name__ == "__main__": asyncio.run(receive_quote())运行这个脚本后,你会看到类似 JSON 的推送不断打出来。这里有几个点要提醒:
extra_headers不是所有平台都支持,有的平台要求把 Token 放在 URL query 参数里,所以先看文档。- 订阅消息的字段名可能叫
action,也可能叫op,甚至可能是二进制协议,必须按文档来。 while True只是演示,真实项目里要对asyncio.TimeoutError、连接断开等情况做处理,后面我会讲。
3.3 行情数据解析:从裸报文到可读行情
我假设推送过来的 JSON 长这样:
{ "type": "quote", "symbol": "HK.00700", "timestamp": 1716890000, "last_price": 388.20, "prev_close": 381.40, "open": 384.50, "high": 390.10, "low": 383.20, "bid": 388.10, "ask": 388.30, "bid_size": 500, "ask_size": 800, "volume": 20134500, "amount": 7823450000.0 }解析函数很简单:
def parse_quote(payload): return { "symbol": payload["symbol"], "last_price": payload["last_price"], "change": round(payload["last_price"] - payload["prev_close"], 3), "change_pct": round( (payload["last_price"] / payload["prev_close"] - 1) * 100, 2 ), "volume": payload["volume"], "amount": payload["amount"], }这里面有两个小细节:
- 涨跌幅要用
last_price和prev_close算。如果你拿open算涨跌,早盘容易出错。 - 港股的每手股数、最小报价单位,不同股票不一样,如果后面要做交易,需要额外维护一张合约表,这里的行情接口通常也会返回。
另外,有些平台推送的是压缩后的二进制格式,直接用官方 SDK 里封装好的Quote对象最方便,不要自己硬啃字节流,容易掉坑。
3.4 补充小工具:多标的批量监控
行情解析出来了,下一步就是展示。我用rich库在终端里做一个自动刷新的表格,效果直观,还不用启动任何前端。
from rich.live import Live from rich.table import Table quotes = {} def update_quotes(payload): data = parse_quote(payload) quotes[data["symbol"]] = data def render_table(): table = Table(title="港股实时行情") table.add_column("标的") table.add_column("现价") table.add_column("涨跌幅") table.add_column("成交额(亿)") for symbol, data in quotes.items(): table.add_row( symbol, f"{data['last_price']:.2f}", f"{data['change_pct']:.2f}%", f"{data['amount'] / 100000000:.2f}", ) return table def display(): with Live(render_table(), refresh_per_second=2) as live: # 这里假设是在接收循环里,收到一条就更新一次 live.update(render_table())实际工程里,推荐用一个小队列把“网络接收线程”和“UI 渲染线程”解耦,避免阻塞网络。我这里只做演示,核心思想是维护一个最新的行情字典,拿到就更新,渲染函数只管读字典。
4. 实战中会踩的坑和排错经验
4.1 常见问题排查表
我把实际遇到的问题整理成一张表,你调试时直接对着查:
| 问题 | 可能原因 | 解决方式 |
|---|---|---|
| 连接一直断开 | 没做心跳,或 Token 过期 | 加应用层心跳,检测 Token 有效期并提前刷新 |
| 订阅后收不到推送 | 股票代码格式错误 | 切换HK.00700/00700.HK后再试试 |
| 偶尔丢数据 | WebSocket 有延迟或断线 | 记录最后一条时间戳,重连后补拉一次快照 |
| 数据延迟超过几秒 | 网络不好,或连接到了非就近节点 | 检查本地网络,在文档里看是否有多个接入点 |
| 解析时报 KeyError | 推送字段为空或接口返回错误 | 先打印原始报文,对照文档确认字段名 |
| 免费额度超限 | 订阅数量太多 | 减少标的数量,或用订阅+分时拉取的策略 |
4.2 关于消息频率和连接保活的实战细节
这一节是我最想分享的部分,因为官方文档很少写清楚。
第一,开盘时段消息密度非常大。如果你同时订阅几十只股票,一秒钟可能收到十几条消息。这时别傻乎乎地每条都写日志,磁盘很快会被撑爆。正确做法是:在内存里维护每个标的的最新行情,只把需要落库的数据(比如每分钟的收盘价)写盘。
第二,应用层心跳非常重要。WebSocket 协议自带 Ping/Pong,但很多网络环境会自动断开空闲连接。可靠的方案是自己定时发一条业务心跳消息,比如每 30 秒发一个{"action": "ping"},收到pong就继续,否则尝试重连。部分平台的心跳消息里还要带时间戳,用来计算链路延迟。
第三,Token 过期时间一般是一小时到一天不等。你不能让程序跑一个月不管它,所以最好写一个自动刷新逻辑。伪代码如下:
async def ensure_token(): if now > token_expire_time - 60: token = await refresh_token() return token第四,断线重连要加退避。一断就连、一断就连,容易被平台临时封禁。我是这样处理的:第一次断开等 1 秒,第二次等 2 秒,最多等 30 秒,重连成功后重置间隔。
5. 一些不太容易被提到的实操心得
第一次搞实时行情的时候,千万别贪多。我一开始是想把几十只股票都订阅进来,结果消息刷得眼花缭乱,调试根本无从下手。后来改成先订阅 3 只,把鉴权、心跳、重连、解析整条链路跑通,再慢慢加标的,效率高很多。
还有一个很多人忽略的点:如果只是盘中盯自选股,不一定非要深度行情(Lv2),基础实时报价已经能解决大部分需求。深度行情要银子,数据量也大,基础报价的字段已经包含现价、涨跌、买卖五档、成交量和成交额,足够做告警和趋势判断。
如果你想把数据存下来做后续分析,建议只持久化 1 分钟 K 线,而不是把每条 tick 都入库。一条 tick 推过来如果用 JSON 打印,平均几百字节,一天下来几十万条,数据库压力不小。先做一个内存聚合器,每分钟生成一个 OHLC 记录,写到 SQLite 或 CSV 里,就够个人复盘用了。
我自己在跑的过程中,最庆幸的一件事是没有用轮询去怼免费接口,而是提前花半小时把 WebSocket 链路理清了。后面再做价格报警、盘中异动检测,都是在同一根连接上扩展,基础已经稳了。如果你也在折腾港股实时行情,建议从最小订阅集合开始,先把鉴权、心跳、断线重连这三件套做扎实。这套思路,放到美股、A 股,甚至加密货币行情上,逻辑也完全通用。