第1章：WebSocket入门——从轮询到长连接的蜕变

做量化交易的朋友应该都有体会：行情数据晚一秒，可能就是真金白银的差距。我刚开始做交易系统时，用的还是最原始的HTTP轮询——每隔几秒发一次请求，问服务器"有新的行情吗？"。

嗯，现在想想，那会儿真是又笨又费流量。

1.1 长连接与短连接：到底差在哪？

先说说短连接。HTTP短连接就像你去便利店买东西——每次进门、挑选、结账、走人，下次再来还得重新开门。每次请求都要建立TCP连接，三次握手四次挥手，光握手就浪费不少时间。

长连接呢？就像你租了个房子，钥匙拿在手里，随时进出。WebSocket就是这种模式——建立一次连接，后续数据直接推送。

对比项	短连接（HTTP轮询）	长连接（WebSocket）
连接建立	每次请求都要建连	一次建连，持续复用
延迟	高（含握手开销）	低（数据实时推送）
服务器压力	高（大量无效请求）	低（有数据才推送）
实时性	取决于轮询间隔	毫秒级
典型场景	查询账户余额	实时行情、订单推送

1.2 WebSocket协议原理：它到底怎么工作的？

WebSocket的原理其实不复杂。说白了，就是先通过HTTP进行一次"握手"，然后协议升级为WebSocket，后续数据直接在TCP通道上双向传输。

握手过程大致是这样的：

1. 客户端发一个HTTP请求，带上 Upgrade: websocket 头
2. 服务器看到这个头，知道你想升级协议
3. 服务器返回101状态码，表示"同意切换"
4. 连接建立完成，后续走WebSocket协议

为什么会这样设计？因为要兼容现有的HTTP基础设施——防火墙、代理服务器都能识别HTTP请求，直接裸TCP容易被拦截。

1.3 使用websocket库订阅实时行情

Python里最常用的WebSocket库就是 websocket（也叫 websocket-client）。我个人习惯用这个库，因为它轻量、稳定，而且支持自动重连。

先安装：

pip install websocket-client

下面是一个订阅币安实时行情的例子：

import json
import websocket

def on_message(ws, message):
    """收到行情数据时的回调"""
    data = json.loads(message)
    print(f"最新价格: {data['p']}")

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

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

def on_open(ws):
    """连接建立后，订阅BTC/USDT的实时行情"""
    subscribe_msg = {
        "method": "SUBSCRIBE",
        "params": ["btcusdt@ticker"],
        "id": 1
    }
    ws.send(json.dumps(subscribe_msg))
    print("已订阅BTC/USDT实时行情")

# 创建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()

1.4 心跳机制与重连策略

做量化交易，最怕的就是连接断了还不知道。你想想看，行情停了10分钟，策略还在那傻等——等发现的时候，亏损已经造成了。

所以，心跳机制和重连策略是必须的。

心跳机制

WebSocket本身有Ping/Pong帧，但很多交易所会自己定义心跳。比如币安要求客户端每3分钟发一次Ping，否则断开连接。

我的做法是：

• 客户端每30秒发一次Ping
• 服务器必须在5秒内回复Pong
• 如果连续3次Ping没有Pong，判定连接断开

import time
import threading

class HeartbeatManager:
    def __init__(self, ws, interval=30, timeout=5):
        self.ws = ws
        self.interval = interval  # 心跳间隔（秒）
        self.timeout = timeout    # 超时时间（秒）
        self.last_pong = time.time()
        self.running = False
    
    def start(self):
        self.running = True
        thread = threading.Thread(target=self._heartbeat_loop)
        thread.daemon = True
        thread.start()
    
    def _heartbeat_loop(self):
        while self.running:
            time.sleep(self.interval)
            # 检查上次Pong是否超时
            if time.time() - self.last_pong > self.timeout * 3:
                print("心跳超时，准备重连")
                self.ws.close()
                break
            # 发送Ping
            self.ws.ping()
    
    def on_pong(self):
        self.last_pong = time.time()

重连策略

重连不是简单地"断了就重连"。我总结了一套策略：

1. 指数退避：第一次重连等1秒，第二次等2秒，第三次等4秒...最多等60秒
2. 最大重试次数：连续重连10次失败，就报警人工介入
3. 状态恢复：重连后重新订阅之前的行情，恢复所有状态

import time
import random

class ReconnectManager:
    def __init__(self, max_retries=10, base_delay=1):
        self.max_retries = max_retries
        self.base_delay = base_delay
        self.retry_count = 0
    
    def reconnect(self, ws_url, subscribe_func):
        while self.retry_count < self.max_retries:
            try:
                # 计算等待时间（指数退避 + 随机抖动）
                delay = min(
                    self.base_delay * (2 ** self.retry_count) + random.uniform(0, 1),
                    60
                )
                print(f"等待 {delay:.1f} 秒后重连...")
                time.sleep(delay)
                
                # 尝试重连
                ws = websocket.WebSocketApp(ws_url)
                subscribe_func(ws)
                ws.run_forever()
                
                # 如果连接成功，重置计数
                self.retry_count = 0
                return ws
                
            except Exception as e:
                self.retry_count += 1
                print(f"重连失败 ({self.retry_count}/{self.max_retries}): {e}")
        
        # 重连次数耗尽，报警
        self._alert_admin()
    
    def _alert_admin(self):
        print("⚠️ 严重告警：WebSocket重连失败，需要人工介入！")
        # 这里可以发邮件、钉钉通知等

1.5 完整示例：一个带心跳和重连的行情客户端

把上面的代码整合一下，就是一个可用的行情客户端了：

class RealTimeMarketClient:
    def __init__(self, url, symbols):
        self.url = url
        self.symbols = symbols
        self.ws = None
        self.heartbeat = None
        self.reconnector = ReconnectManager()
    
    def start(self):
        self._connect()
    
    def _connect(self):
        self.ws = websocket.WebSocketApp(
            self.url,
            on_open=self._on_open,
            on_message=self._on_message,
            on_error=self._on_error,
            on_close=self._on_close
        )
        
        # 启动心跳
        self.heartbeat = HeartbeatManager(self.ws)
        self.heartbeat.start()
        
        # 启动连接
        self.ws.run_forever(ping_interval=30, ping_timeout=5)
    
    def _on_open(self, ws):
        print("连接已建立")
        # 订阅所有品种
        for symbol in self.symbols:
            sub_msg = {
                "method": "SUBSCRIBE",
                "params": [f"{symbol}@ticker"],
                "id": 1
            }
            ws.send(json.dumps(sub_msg))
    
    def _on_message(self, ws, message):
        # 更新心跳时间
        if self.heartbeat:
            self.heartbeat.on_pong()
        # 处理行情数据
        data = json.loads(message)
        print(f"{data['s']}: {data['c']}")
    
    def _on_error(self, ws, error):
        print(f"错误: {error}")
    
    def _on_close(self, ws, close_status_code, close_msg):
        print("连接关闭，尝试重连...")
        self.reconnector.reconnect(self.url, self._subscribe_all)

# 使用示例
client = RealTimeMarketClient(
    url="wss://stream.binance.com:9443/ws",
    symbols=["btcusdt", "ethusdt"]
)
client.start()

---

公众号:蓝海数据掘金营，微信deep3321