在搭建行情采集、盘口因子、仿真回测工具的过程中，很多策略研究者会通过 Python 股票 API 的 WebSocket 长连接获取实时 Tick 数据流。长连接的心跳机制属于容易被忽视的底层环节，但它直接决定实盘环境下行情数据的连续性，进而影响因子计算、信号生成的有效性。
早期开发原型时，我直接对 WebSocket 配置固定心跳周期。本地测试环境网络稳定，整套链路运行正常。但部署到实盘采集环境后，公网网络时延存在动态波动，出现了一类典型现象：WebSocket 产生僵尸连接，数据流已经中断，但程序无法识别连接异常，继续向模型、回测模块输送过期数据，造成策略信号失真。
固定心跳机制在量化工程中的固有短板
不少量化 Demo、示例代码均采用固定心跳配置，例如设置 10s、30s、60s 定时发送 ping 报文。该方案实现简单，适合本地调试与离线回测回放场景。
迁移至线上实时行情采集场景，网络条件动态变化，固定参数会暴露出两类工程缺陷：
- 网络低时延状态下，心跳间隔设置过短，产生大量无效请求，消耗接口配额与网络带宽资源；
- 网络抖动、往返时延抬升时，心跳周期过长，故障检测滞后，僵尸连接持续存在，干扰上层策略逻辑。
静态参数无法适配多变的公网环境，想要保障实盘数据链路可靠，需要心跳间隔可以跟随链路实际网络质量自适应变化。
解决方案：基于往返时延 RTT 构建自适应心跳逻辑
动态心跳的核心原理，是持续采样 WebSocket 链路的往返网络时延。
发送 ping 心跳报文时记录时间戳，收到服务端 pong 应答后记录响应时间，两者差值即为本次往返时延 RTT。累积多组时延样本，评估链路健康状态，以此动态切换心跳发送周期。
本次实践采用的判定规则：
- 平均时延 100‑500ms：心跳间隔 30 秒
- 平均时延大于 500ms：心跳间隔缩短至 10 秒，提高异常检测频次
- 平均时延小于 100ms：心跳间隔拉长至 60 秒，降低通信开销
链路时延较低，降低心跳发送频率，节约资源；网络条件恶化，缩短检测周期，快速识别连接故障。对比固定配置，该机制更适配量化工具实时行情采集的运行需求。
方案验证阶段，订阅 WebSocket 行情流，接收 Tick 数据的同时集成心跳检测逻辑。
import websocket
import json
import time
def on_open(ws):
sub_req = {
"action": "subscribe",
"source": "alltick",
"symbol": "600000",
"type": "trade"
}
ws.send(json.dumps(sub_req))
def heartbeat_check(ws):
start = time.time()
ws.send(json.dumps({"action": "ping"}))
rtt = (time.time() - start) * 1000
if rtt < 100:
return 60
elif rtt < 500:
return 30
else:
return 10
if __name__ == "__main__":
ws_app = websocket.WebSocketApp("wss://api.alltick.co/stock/websocket", on_open=on_open)
ws_app.run_forever()
⚠️工程提示：以上为最简演示代码。用于策略实盘采集时，建议引入时延滑动平均算法，过滤瞬时网络毛刺带来的误判；心跳逻辑运行在独立线程，避免海量 Tick 消息阻塞心跳检测流程。
量化工程落地的关键注意事项
心跳并不是发送越频繁，连接可靠性就越高。心跳报文过于密集会增加通信负载；间隔设置过大，会拉长故障发现的时间窗口。
实践中的处理原则：不依据单次时延突变就变更心跳周期，需要连续多轮采样确认网络状态持续变化之后，再调整心跳间隔，降低误切换概率。
自适应心跳只是链路维护的其中一环，必须配套断线自动重连机制。连接断开后需要自动重建会话，恢复原有行情订阅，否则网络恢复之后，数据采集依旧无法正常工作。
做量化研究时，大家更多聚焦因子逻辑、回测结果，往往会忽略底层长连接的稳定性。动态自适应心跳的开发成本较低，但可以有效降低线上僵尸连接的发生概率。只有底层行情数据流稳定可靠，上层的因子计算、信号生成、实盘仿真才具备可信的数据基础。
交流探讨
各位策略研究者在使用 Python 股票 API 搭建 WebSocket 行情采集链路时，是否遇到心跳配置不合理、僵尸连接、断线感知滞后等问题？欢迎分享你的工程处理思路与踩坑经验。