4、WebSocket实战：使用Python的websocket库连接交易所行情推送

说到高频数据获取，很多人第一反应是REST API。嗯，我刚开始做量化时也这么想。但很快就被现实教育了——轮询太慢，数据延迟高，还容易被限流。

WebSocket才是正经做高频交易该用的东西。它建立的是长连接，交易所一有数据变化就主动推给你，延迟能控制在毫秒级。说白了，REST是你去问交易所「有数据吗？」，WebSocket是交易所告诉你「数据来了！」。

4.1 WebSocket基础概念

WebSocket是一种全双工通信协议。它通过一次HTTP握手建立连接，之后双方可以随时互发消息。我习惯把它想象成一条「数据管道」——交易所那头不断往里灌数据，你这头接上管子就能喝到。

为什么它适合量化交易？

• 低延迟： 建立连接后，数据直接推送，没有HTTP请求的额外开销
• 实时性： 行情变化毫秒级推送，适合做高频策略
• 节省带宽： 不需要反复发送请求头，数据量更小
• 双向通信： 可以同时发送订阅请求和接收数据

我在项目中遇到过一个问题：用REST API轮询1秒一次，结果某次行情剧烈波动时，我拿到的数据比实际慢了3秒。那次之后，我所有实时策略都改用WebSocket了。

4.2 Python的websocket库

Python里做WebSocket，最常用的就是websocket库。它轻量、稳定，而且支持自动重连。我个人习惯用websocket-client这个版本，因为它更接近标准库的用法。

安装很简单：

pip install websocket-client

基本用法是这样的：

import websocket
import json

def on_message(ws, message):
    """收到消息时的回调"""
    data = json.loads(message)
    print(f"收到行情: {data}")

def on_error(ws, error):
    """出错时的回调"""
    print(f"连接出错: {error}")

def on_close(ws, close_status_code, close_msg):
    """连接关闭时的回调"""
    print("连接已关闭")

def on_open(ws):
    """连接建立时的回调"""
    print("连接已建立")
    # 订阅BTC/USDT的深度数据
    subscribe_msg = {
        "method": "SUBSCRIBE",
        "params": ["btcusdt@depth20"],
        "id": 1
    }
    ws.send(json.dumps(subscribe_msg))

# 创建WebSocket连接
ws = websocket.WebSocketApp(
    "wss://stream.binance.com:9443/ws",
    on_open=on_open,
    on_message=on_message,
    on_error=on_error,
    on_close=on_close
)

# 启动连接（阻塞方式）
ws.run_forever()

4.3 实战：连接币安行情

币安是目前最主流的交易所之一，它的WebSocket接口很规范。我们来写一个完整的实战案例。

首先，确定你要订阅什么数据。币安支持多种数据流：

数据流类型	说明	示例
深度数据	订单簿的快照和增量	btcusdt@depth20
K线数据	不同周期的K线	btcusdt@kline_1m
交易数据	实时成交记录	btcusdt@trade
Ticker数据	24小时统计信息	btcusdt@ticker

下面是一个完整的实战代码：

import websocket
import json
import threading
import time

class BinanceWebSocket:
    """币安WebSocket客户端"""
    
    def __init__(self, symbols, stream_type="depth20"):
        self.symbols = symbols
        self.stream_type = stream_type
        self.ws = None
        self.running = False
        
    def on_message(self, ws, message):
        """处理收到的行情数据"""
        try:
            data = json.loads(message)
            # 这里可以加入你的数据处理逻辑
            print(f"[{time.strftime('%H:%M:%S')}] {data}")
        except Exception as e:
            print(f"解析数据出错: {e}")
    
    def on_error(self, ws, error):
        print(f"WebSocket错误: {error}")
    
    def on_close(self, ws, close_status_code, close_msg):
        print("WebSocket连接关闭")
        self.running = False
    
    def on_open(self, ws):
        """连接建立后订阅数据"""
        print("连接成功，开始订阅...")
        for symbol in self.symbols:
            stream_name = f"{symbol.lower()}@{self.stream_type}"
            subscribe_msg = {
                "method": "SUBSCRIBE",
                "params": [stream_name],
                "id": 1
            }
            ws.send(json.dumps(subscribe_msg))
            print(f"已订阅: {stream_name}")
    
    def start(self):
        """启动WebSocket连接"""
        self.running = True
        url = "wss://stream.binance.com:9443/ws"
        
        self.ws = websocket.WebSocketApp(
            url,
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )
        
        # 在独立线程中运行
        wst = threading.Thread(target=self.ws.run_forever)
        wst.daemon = True
        wst.start()
    
    def stop(self):
        """停止连接"""
        self.running = False
        if self.ws:
            self.ws.close()

# 使用示例
if __name__ == "__main__":
    # 订阅BTC和ETH的深度数据
    client = BinanceWebSocket(
        symbols=["btcusdt", "ethusdt"],
        stream_type="depth20"
    )
    client.start()
    
    # 运行30秒后停止
    time.sleep(30)
    client.stop()

4.4 数据解析与处理

拿到原始数据后，你得解析成自己能用的格式。币安推送的深度数据长这样：

{
  "e": "depthUpdate",      // 事件类型
  "E": 123456789,          // 事件时间
  "s": "BTCUSDT",          // 交易对
  "U": 157,                // 第一更新ID
  "u": 160,                // 最后更新ID
  "b": [                   // 买单深度
    ["0.0024", "10"],      // [价格, 数量]
    ["0.0023", "5"]
  ],
  "a": [                   // 卖单深度
    ["0.0026", "8"],
    ["0.0027", "3"]
  ]
}

我习惯把它转成Pandas DataFrame，方便后续分析：

import pandas as pd

def parse_depth(data):
    """解析深度数据为DataFrame"""
    bids = pd.DataFrame(data['b'], columns=['price', 'quantity'])
    asks = pd.DataFrame(data['a'], columns=['price', 'quantity'])
    
    # 转为数值类型
    bids['price'] = bids['price'].astype(float)
    bids['quantity'] = bids['quantity'].astype(float)
    asks['price'] = asks['price'].astype(float)
    asks['quantity'] = asks['quantity'].astype(float)
    
    return {
        'symbol': data['s'],
        'time': data['E'],
        'bids': bids,
        'asks': asks
    }

4.5 断线重连机制

WebSocket连接不是100%稳定的。网络抖动、交易所维护、防火墙超时都可能导致断开。没有重连机制的WebSocket，说白了就是个定时炸弹。

我写了一个带自动重连的版本：

import time
import random

class ReconnectingWebSocket:
    """带自动重连的WebSocket客户端"""
    
    def __init__(self, url, max_retries=5):
        self.url = url
        self.max_retries = max_retries
        self.retry_count = 0
        
    def connect_with_retry(self):
        """带重试的连接"""
        while self.retry_count < self.max_retries:
            try:
                ws = websocket.WebSocketApp(
                    self.url,
                    on_message=self.on_message,
                    on_error=self.on_error,
                    on_close=self.on_close,
                    on_open=self.on_open
                )
                ws.run_forever()
                
            except Exception as e:
                self.retry_count += 1
                wait_time = min(30, 2 ** self.retry_count)
                print(f"连接失败，{wait_time}秒后重试...")
                time.sleep(wait_time)
                
        print("重试次数已达上限，停止连接")

4.6 性能优化建议

高频场景下，每一毫秒都很关键。这里分享几个我踩过的坑：

• 减少JSON解析次数： 如果数据格式固定，可以用正则或字符串分割代替json.loads
• 使用内存队列： 收到数据后先放队列，另起线程处理，避免阻塞WebSocket接收
• 批量写入： 不要每条数据都写数据库，攒一批再写，IO效率高很多
• 关闭日志： 生产环境关掉print，日志写入也会拖慢速度

嗯，这里要注意一点：优化要适度。我见过有人为了省几微秒，把代码写得完全不可维护。结果后来需求一变，改代码改到崩溃。

4.7 本章知识体系

下面这张图展示了WebSocket实战的核心逻辑：

WebSocket实战的核心就三件事：连得上、收得到、断得重。把这三点做好，你的实时数据管道就稳了。