4、WebSocket实时行情：建立连接、订阅主题、心跳维持、数据解析与重连机制
做量化交易，最怕什么？
行情断了，你还在那儿傻傻地挂单。我见过不止一个团队，因为WebSocket重连没写好，盘中直接亏掉几十万。所以这一章，咱们把实时行情的每个环节都掰开揉碎讲清楚。
4.1 为什么非用WebSocket不可？
你想想看，如果用HTTP轮询，每秒拉一次行情，延迟高不说，服务器还得把你IP封了。WebSocket是全双工通信，建立一次连接，后面全是推送。说白了，它就是为实时行情量身定做的。
我个人习惯，所有交易系统的行情模块，底层通信只用WebSocket。RESTful只用来做查询和下单。
4.2 建立连接——第一步就踩坑
连接代码其实很简单，但坑都在细节里。我直接给出一份我项目里用过的代码：
import asyncio
import websockets
async def connect_market():
uri = "wss://api.yourbroker.com/market"
try:
async with websockets.connect(
uri,
ping_interval=None, # 我们自己控制心跳
close_timeout=5,
max_size=2**20 # 限制消息大小，防止内存爆炸
) as ws:
print(f"连接成功: {uri}")
await handle_messages(ws)
except Exception as e:
print(f"连接失败: {e}")
# 触发重连逻辑
4.3 订阅主题——别一股脑全订阅
行情数据量很大。你如果同时订阅几百个币种的深度和成交，本地带宽和CPU直接拉满。我建议分两步走：
- 先订阅核心品种（比如你持仓的、准备交易的）
- 再按需动态订阅（比如监控列表里的）
订阅消息的格式各家不同，但大同小异。以某头部交易所为例：
# 订阅深度
{
"op": "subscribe",
"args": [
{
"channel": "depth",
"instId": "BTC-USDT",
"depth": 5
}
]
}
# 订阅成交
{
"op": "subscribe",
"args": [
{
"channel": "trade",
"instId": "ETH-USDT"
}
]
}
嗯，这里要注意：订阅成功后，服务器通常会返回一条确认消息。一定要等确认收到后，再开始处理行情数据。否则你可能会收到乱序数据。
4.4 心跳维持——活着才能赚钱
WebSocket连接如果不发心跳，一般几分钟后就会被服务器断开。心跳机制说白了就是定时发一个固定格式的消息，告诉服务器「我还活着」。
我见过最蠢的写法是每30秒发一次心跳，结果服务器要求的是15秒。所以一定要看文档！
async def heartbeat(ws, interval=15):
while True:
try:
await ws.send('{"op": "ping"}')
await asyncio.sleep(interval)
except Exception as e:
print(f"心跳发送失败: {e}")
break
4.5 数据解析——别在回调里做重活
行情数据通常是JSON格式。解析本身不复杂，但有个大坑：不要在消息回调函数里做耗时操作。
为什么？因为WebSocket的消息处理是单线程的。你在回调里做数据库写入、计算指标，后面的消息就得排队等着。延迟就是这么来的。
我个人的做法是：
async def handle_messages(ws):
async for message in ws:
# 1. 快速解析
data = json.loads(message)
# 2. 判断消息类型
if data.get("op") == "ping":
await ws.send('{"op": "pong"}')
continue
# 3. 丢进队列，让其他协程处理
await data_queue.put(data)
解析时还要注意：有些交易所的深度数据是增量更新的，你需要维护一个本地的OrderBook快照。这个逻辑写不好，盘口价格能差好几个点。
4.6 重连机制——最后的防线
网络不可能100%稳定。重连机制是必须的，而且不能写得简单粗暴。
我曾经犯过一个错误：断线后立即重连，结果服务器以为我在攻击，直接封IP了。后来我改成了指数退避：
async def reconnect_with_backoff(uri, max_retries=10):
base_delay = 1 # 初始1秒
for attempt in range(max_retries):
try:
async with websockets.connect(uri) as ws:
print(f"第{attempt+1}次重连成功")
await handle_messages(ws)
break # 正常退出循环
except Exception as e:
delay = base_delay * (2 ** attempt) # 指数增长
delay = min(delay, 60) # 最大60秒
print(f"重连失败，{delay}秒后重试...")
await asyncio.sleep(delay)
4.7 整体架构图
下面这张图，是我自己项目里用的WebSocket行情模块架构。你照着这个结构写，基本不会出大问题：
4.8 避坑指南
最后，我把这些年踩过的坑总结一下：
- 不要信任默认配置——每个交易所的WebSocket实现都有差异，文档必须逐字看
- 数据校验不能省——我遇到过服务器推送了错误的时间戳，导致策略误判
- 日志要打全——连接断开、重连、订阅成功/失败，这些关键事件必须记录
- 本地缓存要加锁——多协程同时读写OrderBook，不加锁数据就乱了
好了，WebSocket实时行情这块，核心就是这些。你把这几个环节写扎实了，行情模块基本就稳了。记住：连接是基础，心跳是保障，解析要高效，重连要智能。
公众号:蓝海资料掘金营，微信deep3321