标签：Python、WebSocket、行情接口、量化工具、回测数据 摘要：在量化研究与策略实盘模拟过程中，实时行情数据流的稳定性直接影响Tick级回测、信号生成的可靠性。WebSocket长连接部署服务器长期运行时，容易出现进程存活但链路实际断开的假连接现象，A股午间休市的数据流静默窗口会放大该故障风险。本文从工程实践角度分析故障成因，给出心跳、重连的参数方案，提供可复用Python代码，帮助研究者构建稳定的行情数据采集链路。
在开展量化策略研究时，无论是实盘模拟、Tick数据落库，还是精细化回测，都依赖持续可靠的A股实时行情输入。我在搭建WebSocket行情采集客户端的过程中遇到过一类典型线上问题：本地环境订阅、数据解析全部正常，部署至服务器持续运行后，程序进程没有退出、无报错输出，但已经不再接收行情Tick，往往等到午后开盘才发现数据链路已经中断。
该问题不会在调试阶段暴露，却会直接破坏数据完整性，进而干扰信号校验、回测结果的可信度。下面结合实际工程经验，分析问题根源并给出完整实现方案。
静默假连接产生原理
WebSocket属于全双工长连接协议，但客户端与行情服务端之间会经过NAT网关、负载均衡、运营商路由设备。网络中间件普遍存在空闲超时回收机制：当链路长时间没有报文交互，设备会清除内部连接映射表。
此时TCP两端均判定连接状态正常，但数据已经无法传输，也就是假存活连接。
A股交易时段存在11:30‑13:00的午间休市窗口，90分钟内几乎没有Tick数据推送。如果缺少保活逻辑，该静默期就极易触发中间设备的连接回收，造成下午开盘后无法获取行情。对于依赖连续时序数据的回测与策略模型，数据断档会造成样本缺失，直接影响研究结论。
WebSocket心跳与重连整体设计
核心处理逻辑：客户端按照固定周期发送心跳探测包，等待服务端应答；连续多次未收到响应，则判定链路失效，主动关闭连接并执行重连流程。
工程要点：多数行情接口订阅会话与WebSocket连接绑定，连接销毁后订阅状态同步丢失。完成重连之后，必须重新提交标的订阅请求，否则连接状态显示正常，但无法收到任何行情数据。
关键参数参考
| 参数 | 推荐配置 | 应用说明 |
|---|---|---|
| 心跳发送间隔 | 20‑30秒 | 周期性维持链路活跃度，规避空闲连接被中间设备回收 |
| 超时判定阈值 | 连续3次无应答 | 屏蔽单次网络抖动带来的误判，减少无效重连 |
| 重连退避策略 | 1s、2s、4s递进，最大上限30s | 抑制故障场景下高频重试请求，降低接口侧压力 |
| 重连后置动作 | 重传完整标的订阅列表 | 重建会话的行情订阅关系，保障数据持续输出 |
Python实现：带心跳与自动重连的行情采集客户端
环境依赖：
pip install websocket‑client代码基于AllTick API的A股WebSocket行情接口开发，可作为行情数据采集的基础骨架，采集得到的Tick数据可用于本地存储、策略信号运算以及细粒度回测。
import websocket
import json
import time
import threading
# ========== 配置项，替换为自身token和研究标的列表 ==========
TOKEN = "your_token_here"
WS_URL = f"wss://quote.alltick.co/quote-stock-b-ws-api?token=yourtoken"
# 待采集的A股标的，根据研究需求调整
SYMBOLS = ["600519.SH", "000001.SZ"]
# ========== 回调处理函数 ==========
def on_message(ws, message):
"""处理服务端推送Tick报文及各类响应消息"""
try:
data = json.loads(message)
cmd_id = data.get("cmd_id")
# cmd_id=22998 对应A股实时tick推送协议
if cmd_id == 22998:
tick_payload = data.get("data", {})
print(f"Tick: {tick_payload.get('code')} | "
f"Price: {tick_payload.get('price')} | "
f"Volume: {tick_payload.get('volume')} | "
f"Time: {tick_payload.get('tick_time')}")
# 研究扩展：Tick落库、策略信号计算、回测数据源输入
else:
# 打印订阅确认等响应报文 cmd_id=22005
print("Response:", data)
except json.JSONDecodeError as e:
print("JSON解析异常：", e)
def on_error(ws, error):
print("WebSocket异常：", error)
def on_close(ws, close_status_code, close_msg):
print("WebSocket连接已关闭")
def on_open(ws):
"""连接建立完成，发起订阅，启动后台心跳线程"""
print("WebSocket连接建立，发送订阅请求")
subscribe_msg = {
"cmd_id": 22004,
"seq_id": 1,
"trace": f"trace‑{int(time.time() * 1000)}",
"data": {
"symbol_list": [{"code": symbol} for symbol in SYMBOLS]
}
}
ws.send(json.dumps(subscribe_msg))
print(f"已订阅标的：{SYMBOLS}")
# 后台心跳循环线程
def heartbeat_loop():
while ws.sock and ws.sock.connected:
time.sleep(10)
try:
ws.send("ping")
print("已发送心跳包")
except Exception as e:
print("心跳发送异常：", e)
break
threading.Thread(target=heartbeat_loop, daemon=True).start()
# ========== 主程序，外层循环实现自动重连 ==========
if __name__ == "__main__":
ws = websocket.WebSocketApp(
WS_URL,
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close
)
while True:
try:
ws.run_forever()
print("连接断开，3秒后执行重连……")
time.sleep(3)
except KeyboardInterrupt:
print("程序正常退出")
break
部署后经过午间休市时段实测，心跳逻辑可以稳定维持链路，午后开盘能够接续接收Tick数据，不需要人工重启进程，保障时序行情数据连续性。
研究实践小结
心跳与重连逻辑代码体量不大，但属于量化数据基础设施中容易被忽略的部分。本地测试环境很难复现休市静默断连、网络抖动、服务端维护断连等边界场景，而这些问题会直接造成行情样本缺失，对Tick级回测、策略模拟的结果产生干扰。
在实际研究工作中，仅保证Demo能够运行是不够的，需要重点关注：
- 重连之后必须重新执行标的订阅；
- 重连流程配置退避策略，避免短时间大量请求；
- 面向研究生产场景，可继续补充持久化日志、链路异常告警，进一步提升数据链路的稳定性。
本示例借助AllTick API标准化的WebSocket行情协议，降低底层协议适配成本，让研究者可以把重心放在数据处理、信号逻辑、模型回测等核心研究环节。
免责声明：本文为量化技术研究分享，代码仅用于学习研究，不构成投资建议。
交流思考
当前示例完成了心跳报文发送，但未校验服务端心跳应答报文。如果需要完整的链路存活判定，应当增加心跳响应超时逻辑。各位策略研究者会采用怎样的实现思路，欢迎一起探讨。