先说结论我最近要跑一个港股行情小助手盘中得盯实时报价不能接受十五分钟延迟。网上搜了一圈能直接用的资源少得可怜免费的要么接口不稳定要么数据字段缺东少西找券商官方接口又怕流程麻烦。折腾下来最后锁定的方案其实不复杂用港股实时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. 手把手接入港股实时行情API3.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: fBearer {TOKEN}} async with websockets.connect(WS_URL, extra_headersheaders) 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_second2) 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 股甚至加密货币行情上逻辑也完全通用。