2. 行情数据源接入：WebSocket协议基础，连接交易所行情网关，心跳机制与重连策略
好，咱们进入实战的第二站。上一章我们把整体架构搭起来了，现在要解决一个核心问题：怎么把交易所的实时行情数据，稳定地拉进来。
你想想看，行情数据是典型的“高吞吐、低延迟”场景。用HTTP轮询？那肯定不行，延迟高、带宽浪费也大。业内标准做法就是WebSocket。说白了，WebSocket就是一条全双工的长连接，交易所往你这边推数据，你只管收就行了。
2.1 WebSocket协议基础
WebSocket本质上是一个应用层协议，它基于TCP。我刚开始接触时也困惑过：它和Socket有什么区别？其实很简单：WebSocket是浏览器和服务器都能用的标准化协议，而Socket是操作系统层面的抽象。
一个典型的WebSocket连接建立过程是这样的：
- 客户端发HTTP请求，带上
Upgrade: websocket头 - 服务器返回101状态码，切换协议
- 之后双方就可以互相发数据帧了
嗯，这里要注意：WebSocket的帧格式是二进制的，不是纯文本。虽然我们平时看到的数据往往是JSON字符串，但那只是应用层封装。底层传输时，数据会被分片、掩码处理。
核心要点：WebSocket的帧类型主要有三种——文本帧、二进制帧、控制帧（Ping/Pong/Close）。行情数据通常用文本帧传输JSON，但有些交易所为了性能，会用二进制帧传压缩后的数据。
2.2 连接交易所行情网关
不同的交易所，WebSocket的接入方式大同小异。我以币安和OKX为例，给你看看实际代码怎么写。
先看一个最简单的连接示例：
import asyncio
import websockets
async def connect_binance():
uri = "wss://stream.binance.com:9443/ws"
async with websockets.connect(uri) as ws:
# 订阅BTC/USDT的深度数据
subscribe_msg = {
"method": "SUBSCRIBE",
"params": ["btcusdt@depth20"],
"id": 1
}
await ws.send(json.dumps(subscribe_msg))
# 持续接收数据
async for message in ws:
data = json.loads(message)
# 处理行情数据...
print(f"收到深度数据: {data}")
这段代码看起来简单，但实际生产环境里，坑可不少。我曾经在对接某二线交易所时，发现它的WebSocket网关每隔30分钟就会主动断开连接，而且没有任何错误提示。查了半天文档，才发现它要求客户端必须定期发送Ping帧。
经验之谈：连接交易所网关时，建议先做三件事：
- 检查交易所的WebSocket文档，确认心跳机制
- 确认数据格式是JSON还是Protocol Buffers
- 测试连接稳定性，特别是网络抖动时的表现
2.3 心跳机制
为什么要心跳？说白了就是保活。TCP连接虽然理论上不会断，但中间的路由器、防火墙、NAT设备都可能把空闲连接给掐掉。交易所网关通常要求客户端定期发送心跳包，否则就会主动断开。
心跳机制一般有两种实现方式：
| 方式 | 说明 | 适用场景 |
|---|---|---|
| WebSocket Ping/Pong | 协议层自带，发送控制帧 | 大多数交易所支持 |
| 应用层心跳 | 发送自定义的JSON消息 | 部分交易所自定义协议 |
我个人习惯用应用层心跳，因为更可控。比如币安要求每隔3分钟发送一个Pong帧，而OKX则要求每隔30秒发送一个Ping消息。看代码：
async def heartbeat(ws, interval=30):
"""发送心跳包"""
while True:
try:
# 应用层心跳
ping_msg = {"ping": int(time.time() * 1000)}
await ws.send(json.dumps(ping_msg))
await asyncio.sleep(interval)
except Exception as e:
print(f"心跳发送失败: {e}")
break
async def main():
async with websockets.connect(uri) as ws:
# 启动心跳任务
asyncio.create_task(heartbeat(ws, interval=30))
# 主循环接收数据
async for message in ws:
# 处理行情数据...
pass
避坑指南：我曾经遇到过一个问题——心跳线程和接收线程同时操作同一个WebSocket连接，导致数据错乱。后来我改用asyncio.create_task来管理心跳，确保所有操作都在同一个事件循环里。记住：WebSocket连接不是线程安全的。
2.4 重连策略
连接断开了怎么办？这是最现实的问题。网络波动、交易所维护、服务器重启，任何原因都可能导致连接中断。一个好的重连策略，是行情系统的生命线。
我总结了一个实用的重连策略：
- 指数退避：第一次重连等1秒，第二次等2秒，第三次等4秒...最大间隔60秒
- 随机抖动：在退避时间上加上随机偏移，防止多个客户端同时重连
- 最大重试次数：比如重试10次后，进入人工介入流程
- 状态恢复：重连后重新订阅之前的行情
看一个完整的重连实现：
async def connect_with_reconnect(uri, max_retries=10):
"""带重连机制的WebSocket连接"""
retry_count = 0
base_delay = 1
while retry_count < max_retries:
try:
async with websockets.connect(uri) as ws:
print("连接成功")
retry_count = 0 # 重置重试计数
# 重新订阅
await subscribe(ws)
# 启动心跳
asyncio.create_task(heartbeat(ws))
# 接收数据
async for message in ws:
await process_message(message)
except websockets.ConnectionClosed:
retry_count += 1
delay = min(base_delay * (2 ** retry_count), 60)
# 加入随机抖动
delay += random.uniform(0, 0.5 * delay)
print(f"连接断开，{delay:.1f}秒后重试 (第{retry_count}次)")
await asyncio.sleep(delay)
except Exception as e:
print(f"未知错误: {e}")
break
print("重试次数已达上限，请检查网络或联系交易所")
关键设计：重连时一定要恢复订阅状态。我见过有人重连后忘了重新订阅，结果行情数据一直为空，策略还在跑，最后亏了不少。建议把订阅信息保存在一个全局变量里，重连时自动恢复。
2.5 本章知识体系
为了让你更直观地理解整个流程，我画了一张图：
这张图把整个流程串起来了。从交易所网关开始，经过WebSocket连接，到数据接收模块。同时心跳模块和重连策略在后台运行，订阅状态管理确保重连后能恢复。
我的建议：刚开始做行情接入时，别急着优化性能。先把连接稳定性做好，再考虑数据解析速度。我见过太多人一开始就追求微秒级延迟，结果连接三天两头断，得不偿失。
好了，这一章的内容就到这里。WebSocket接入是行情系统的地基，地基打牢了，后面的数据处理、存储、分发才能稳。下一章我们会深入数据解析，看看怎么高效处理这些海量的行情数据。