3、Python WebSocket实战：使用websockets库连接交易所，订阅深度流和交易流
好，咱们直接进入正题。
上一章我们聊了WebSocket协议的原理，说白了就是一条全双工的长连接通道。但光懂原理没用，得能写代码跑起来才行。这一章，我就带你手把手用Python的websockets库，连上真实的交易所，把深度数据和交易数据实时拉下来。
3.1 为什么选websockets库？
Python里做WebSocket的库有好几个，比如websocket-client、aiohttp自带的ws支持。但我个人习惯用websockets这个库。原因很简单：
- 它原生支持
async/await，跟asyncio配合得天衣无缝 - API设计干净，没有多余的花哨功能
- 我在项目中遇到过几次连接断开的情况，它的重连机制写起来很顺手
安装就更简单了：
pip install websockets嗯，一行搞定。
3.2 连接交易所的基本框架
先别急着写具体业务逻辑。我们搭一个通用的WebSocket客户端骨架。你想想看，不管连哪个交易所，流程其实都差不多：
- 建立连接
- 发送订阅消息
- 循环接收数据
- 处理心跳（ping/pong）
- 异常断开后重连
下面这个代码，是我在多个项目中反复用过的模板：
import asyncio
import websockets
import json
async def connect_exchange(url, subscribe_msg, handler):
"""
通用WebSocket连接器
:param url: 交易所ws地址
:param subscribe_msg: 订阅消息（dict格式）
:param handler: 数据处理回调函数
"""
async for websocket in websockets.connect(url, ping_interval=20):
try:
# 发送订阅
await websocket.send(json.dumps(subscribe_msg))
print(f"[订阅成功] {subscribe_msg}")
# 持续接收
async for message in websocket:
data = json.loads(message)
await handler(data)
except websockets.ConnectionClosed:
print("[连接断开] 正在重连...")
continue
except Exception as e:
print(f"[异常] {e}")
continue这里有个细节：ping_interval=20。为什么要设这个？
我曾经踩过一个坑——某交易所的网关如果30秒内没收到任何消息，就会主动断开连接。而我们的行情数据有时候几秒才来一条，中间的空窗期就触发了超时。加上这个参数后，库会自动帮我们发心跳包，省心很多。
3.3 订阅深度流（Order Book）
深度数据，说白了就是买卖盘口的挂单情况。做高频交易或者盘口分析的人，最离不开的就是这个。
以币安为例，深度流的地址是：
wss://stream.binance.com:9443/ws/btcusdt@depth20@100ms这个地址的意思是：订阅BTC/USDT交易对，深度20档，每100毫秒推送一次。
完整代码长这样：
async def handle_depth(data):
"""处理深度数据"""
bids = data.get('b', []) # 买单
asks = data.get('a', []) # 卖单
print(f"买一价: {bids[0][0] if bids else 'N/A'}, 量: {bids[0][1] if bids else 'N/A'}")
print(f"卖一价: {asks[0][0] if asks else 'N/A'}, 量: {asks[0][1] if asks else 'N/A'}")
async def main():
url = "wss://stream.binance.com:9443/ws/btcusdt@depth20@100ms"
# 币安的深度流不需要发送订阅消息，连接后自动推送
await connect_exchange(url, {}, handle_depth)
if __name__ == "__main__":
asyncio.run(main())3.4 订阅交易流（Trade Stream）
交易流，就是每一笔成交的实时记录。它比K线更原始，粒度更细。
币安的交易流地址：
wss://stream.binance.com:9443/ws/btcusdt@trade代码示例：
async def handle_trade(data):
"""处理成交数据"""
trade = {
'price': data['p'],
'quantity': data['q'],
'time': data['T'],
'is_buyer_maker': data['m'] # True表示卖方主动，False表示买方主动
}
side = "卖" if trade['is_buyer_maker'] else "买"
print(f"[{side}] 价格: {trade['price']}, 数量: {trade['quantity']}")
async def main_trade():
url = "wss://stream.binance.com:9443/ws/btcusdt@trade"
await connect_exchange(url, {}, handle_trade)这里有个判断买卖方向的技巧：is_buyer_maker字段。如果为True，说明是卖方主动吃单，市场偏空；反之则是买方主动，市场偏多。我在做盘口异动监控时，就靠这个字段来判断资金流向。
3.5 同时订阅多个流
实际生产中，你不会只盯一个交易对。我经常需要同时监控十几个币种的深度和交易。
币安支持组合流，用/stream?streams=的方式：
wss://stream.binance.com:9443/stream?streams=btcusdt@depth20@100ms/ethusdt@depth20@100ms/btcusdt@trade代码里这样处理：
async def handle_multi_stream(data):
"""处理多路数据"""
stream_name = data.get('stream', 'unknown')
inner_data = data.get('data', {})
print(f"[{stream_name}] 收到数据")
async def main_multi():
url = ("wss://stream.binance.com:9443/stream?"
"streams=btcusdt@depth20@100ms/ethusdt@depth20@100ms/btcusdt@trade")
await connect_exchange(url, {}, handle_multi_stream){'stream': 'xxx', 'data': {...}}，解析时别忘了剥开这层壳。我刚开始写的时候没注意，debug了半小时才发现。
3.6 知识体系总览
下面这张图，帮你把本章的核心逻辑串起来：
3.7 避坑指南
最后，分享几个我实战中遇到的坑：
- 心跳超时：不同交易所的心跳间隔不一样。币安是20秒，OKX是30秒。我曾经用默认值连OKX，结果频繁断连。解决办法就是根据交易所文档手动设置
ping_interval。 - 数据积压：如果handler处理太慢，数据会堆积在内存里。我一般会在handler里加一个
asyncio.Queue做缓冲，把数据丢给另一个协程去处理。 - 重连风暴：当交易所网关出问题时，所有客户端同时重连，可能把网关打挂。我习惯在重连前加一个随机延迟，比如
await asyncio.sleep(random.uniform(1, 5))。
好了，这一章的内容就到这里。代码不多，但都是能直接跑起来的东西。你拿着这些代码，换个交易所地址和订阅参数，就能连上任何支持WebSocket的行情源。