news 2026/10/3 9:17:07

港股实时行情API接入实战:用WebSocket订阅实时报价,避开延时行情坑

作者头像

张小明

前端开发工程师

1.2k 24
文章封面图
港股实时行情API接入实战:用WebSocket订阅实时报价,避开延时行情坑

先说结论:我最近要跑一个港股行情小助手,盘中得盯实时报价,不能接受十五分钟延迟。网上搜了一圈,能直接用的资源少得可怜,免费的要么接口不稳定,要么数据字段缺东少西,找券商官方接口又怕流程麻烦。折腾下来,最后锁定的方案其实不复杂:用港股实时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 则是建立一条长连接,服务端有新的报价就往下推,延迟低、资源占用小,是行情订阅的主流方式。

整个链路很清晰:

  1. 注册应用,拿到App Key和App Secret。
  2. 调用认证接口,获取Access Token。
  3. 用 Token 建立 WebSocket 连接。
  4. 发送订阅消息,指定想要的股票代码。
  5. 之后持续接收服务端推送的行情消息。

下面我按这个链路给出可复现的代码逻辑。注意各家 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 股,甚至加密货币行情上,逻辑也完全通用。

版权声明: 本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若内容造成侵权/违法违规/事实不符,请联系邮箱:809451989@qq.com进行投诉反馈,一经查实,立即删除!
网站建设 2026/10/3 9:16:47

Java校园旧物交易系统开发实战:从环境配置到论文答辩

每年毕业季,图书馆和宿舍楼下总能看到一堆堆被遗弃的教材、台灯、自行车。其实这些东西里很多都还有使用价值,但缺乏一个靠谱的流转渠道。我前两年带学生做课设时,恰好有小组选了这个题目——Java实现的校园旧物交易系统,当时跟着…

作者头像 李华
网站建设 2026/10/3 9:15:52

DevEco Studio模拟器白屏怎么办?从原理到解决的五步排查指南

1. 问题现象与影响范围 先说一个我最近被问到最多的问题:DevEco Studio里的模拟器打开之后,整个窗口一片白,既没有桌面图标,也没有启动器界面,偶尔底部会有一根进度条,刷完就没了下文。更折腾人的是&#x…

作者头像 李华
网站建设 2026/10/3 9:15:17

RV1106 ISP与MIPI/LVDS配置实战:设备树调优与画质问题定位

做 RV1106 方案的第一个晚上,我盯着排线陷入沉思:sensor 供电正常、复位也拉完了,dmesg里死活不报 sensor 挂载,MIPI 时钟测出来却又是波形。后来翻了一整晚的资料,猜了无数种可能,最后发现根因既不在硬件上…

作者头像 李华
网站建设 2026/10/3 9:14:26

HTTP/3落地指南:从QUIC原理到部署避坑全解析

HTTP/3:旧问题的终结者,新问题的制造者我从19年开始跟HTTP/3的草案,当时还在叫QUIC,RFC 9000一发布我就在生产环境试水。说实话,踩过的坑比收获的惊喜还多。这玩意儿不是简单地把TCP换成UDP,而是把整个传输…

作者头像 李华
网站建设 2026/10/3 9:12:56

Linux网桥搭建与iperf性能验证实战:从原理到排错

做网络设备测试和服务器性能评估这些年,我发现自己最常用的两个工具其实特别朴素:一个是把多张网卡“拧”成一个整体、让设备在局域网里拥有统一身份的网桥,另一个是到处给链路做“体检”的 iperf。这两个东西单独拿出来都有大量文档&#xf…

作者头像 李华
网站建设 2026/10/3 9:12:52

手写BP神经网络逼近二元函数:MATLAB实现与避坑指南

简介:这份压缩包内含一份基于MATLAB手写BP神经网络逼近二元函数的完整源码,面向希望深入理解反向传播原理、不依赖任何工具箱的机器学习初学者与研究生。包内仅含1个m脚本文件,整体大小约1KB,代码精简却完整覆盖了网络结构设计、随…

作者头像 李华