A 股实时行情 WebSocket 静默断线：心跳保活与重连实战笔记

标签：Python, WebSocket, 量化研究, 行情数据, Tick回测 摘要：做量化研究经常会自己搭行情采集脚本，本地跑一切正常，丢到服务器长期运行就会遇到WebSocket假连接问题。尤其是A股午间休市的空窗期，很容易悄无声息断流，造成Tick数据缺失，直接影响回测质量。本文记录踩坑过程，梳理心跳参数，附上可直接调试使用的Python代码，供做策略、做数据的朋友参考交流。

自己折腾量化策略的时候，不管是做本地Tick库搭建，还是跑模拟实盘、精细化回测，都离不开稳定的实时行情源。我之前写的A股WebSocket采集脚本，在本机调试的时候订阅、解析全部没问题。部署到服务器后台跑，隔一段时间就出现诡异现象：进程还在，没有报错日志，但已经收不到任何行情。往往等到下午开盘复盘数据，才发现中间一大段Tick直接丢了。
这类问题不会在短时间本地测试暴露，只会在长时间服务端运行才显现，一旦发生会破坏时序数据完整性，回测、策略模拟结果也就失去参考意义。下面把排查思路和完整实现整理出来。
什么是WebSocket假存活断线
WebSocket本身是长连接协议，但数据包传输途中，会经过NAT网关、负载均衡、运营商路由等多层网络设备。绝大多数中间网络设备都设置了空闲超时机制：一段时间没有数据包往来，就会清除内部连接映射表。
此时客户端、服务端两边的TCP栈依旧认为连接正常，实际上已经无法收发数据，也就是常说的假连接。
A股11:30‑13:00午间休市有长达90分钟几乎没有行情推送，链路长期处于静默状态，是断线高发窗口。没有保活逻辑，休市阶段连接就可能被回收，下午开盘之后脚本只会原地等待，不会主动报错。对于依赖连续时序的量化工作，数据缺口会直接干扰模型与回测结论。
心跳与自动重连设计要点
核心逻辑：客户端按固定间隔主动发送心跳探测包，等待服务端应答；连续多次收不到回复，则判定链路异常，主动关闭连接并执行重连。

实践提醒：多数行情接口的订阅状态是绑定WebSocket会话的，一旦连接断开，订阅就随之失效。重连完成必须重新提交标的订阅，不然表面看连接成功，却收不到任何行情。

参考配置参数

参数	推荐配置	说明
心跳发送间隔	20‑30秒	周期性维持链路活跃，防止中间设备回收空闲连接
超时判定阈值	连续3次无应答	规避偶然网络抖动造成误判，减少无效重连
重连退避策略	1s、2s、4s递增，最大30s	故障期间避免疯狂重试，减轻接口服务压力
重连后置动作	重传完整标的订阅列表	在新会话恢复行情订阅关系

Python完整示例代码

依赖安装：

1. 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')}")
            # 此处扩展：数据入库、策略信号计算、回测数据源输入
        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出现缺口，后面回测、策略评估都会失真。
几点实际使用建议：

• 重连后务必重新发起标的订阅；
• 重连做好退避，避免短时间大量请求打接口；
• 如果用于长期研究，建议补充日志持久化、异常告警，进一步提升稳定性。

这份实践使用AllTick API提供的标准化WebSocket行情协议，省去不少底层协议适配工作，可以把更多精力放在数据处理、策略逻辑、回测验证上面。

免责声明：本文仅为个人量化技术实践分享，代码仅供学习研究，不构成投资建议。

讨论
当前示例只实现心跳包发送，没有做服务端心跳应答的超时校验。如果要做到更严谨的链路健康检测，还需要补充应答超时逻辑。各位写行情脚本的朋友一般是怎么处理这块，欢迎回帖交流。
要不要我帮你再生成2~3个适配一亩三分地社区的备选标题？