4、数据采集引擎：WebSocket实时订阅Tick数据、心跳机制、断线重连策略

好，咱们进入第四章。这一章讲的是数据采集引擎，说白了就是怎么把Tick数据从交易所实时拉下来。

我个人习惯把数据采集比作「水管工程」——你得把水（数据）从源头（交易所）引到你家（系统），中间不能断流，不能漏水。WebSocket就是这根水管，心跳机制是检查水管有没有堵，断线重连是水管破了之后自动修好。

4.1 为什么选WebSocket？

你可能要问：为什么不用HTTP轮询？

嗯，我刚开始做量化的时候也这么想过。后来发现，Tick数据是毫秒级的，HTTP每次请求都要建立连接、发送头信息、等待响应，一来一回几十毫秒就没了。你想想看，高频交易里几十毫秒能差出多少个价位？

WebSocket不一样。它建立一次连接，之后数据就像流水一样推过来。延迟低，开销小，适合实时场景。

4.2 订阅Tick数据的基本流程

流程其实不复杂，我画了张图帮你理清思路：

4.3 代码实现：WebSocket客户端

我用Python的websockets库来实现。这个库轻量、稳定，我在生产环境里用过两年多，没出过什么大问题。

import asyncio
import websockets
import json
from datetime import datetime

class TickDataClient:
    def __init__(self, url, symbols):
        self.url = url
        self.symbols = symbols
        self.ws = None
        self.reconnect_attempts = 0
        self.max_reconnect = 10
        
    async def connect(self):
        """建立WebSocket连接"""
        self.ws = await websockets.connect(self.url)
        print(f"[{datetime.now()}] 连接成功: {self.url}")
        # 发送订阅消息
        subscribe_msg = {
            "action": "subscribe",
            "symbols": self.symbols
        }
        await self.ws.send(json.dumps(subscribe_msg))
        
    async def receive_tick(self):
        """接收Tick数据"""
        async for message in self.ws:
            data = json.loads(message)
            # 这里可以加解析逻辑
            print(f"收到Tick: {data['symbol']} 价格:{data['price']} 时间:{data['timestamp']}")
            
    async def run(self):
        """主循环"""
        while True:
            try:
                await self.connect()
                await self.receive_tick()
            except websockets.ConnectionClosed:
                print("连接断开，准备重连...")
                await self.reconnect()
                
    async def reconnect(self):
        """断线重连"""
        while self.reconnect_attempts < self.max_reconnect:
            wait_time = 2 ** self.reconnect_attempts  # 指数退避
            print(f"等待 {wait_time} 秒后重连...")
            await asyncio.sleep(wait_time)
            try:
                await self.connect()
                self.reconnect_attempts = 0
                return
            except:
                self.reconnect_attempts += 1
        raise Exception("重连失败，已达最大尝试次数")

4.4 心跳机制：保持连接活跃

为什么要心跳？因为网络环境不是完美的。有时候连接看起来还在，实际上已经断了。你不主动检查，就永远不知道。

我曾经遇到过一个坑：某次行情波动特别大，WebSocket连接其实已经断了，但客户端没发现，一直以为还在接收数据。结果策略跑了一整天，全是空数据。从那以后，我强制要求所有采集模块必须加心跳。

心跳机制一般分两种：

类型	原理	适用场景
Ping/Pong	客户端发Ping，服务端回Pong	标准WebSocket协议自带
应用层心跳	自定义消息，比如发个{"type":"heartbeat"}	需要业务层确认连接正常

我一般两种都用。底层用Ping/Pong保活，上层用应用层心跳确认数据链路正常。

async def heartbeat(self):
    """心跳检测协程"""
    while True:
        try:
            # 发送Ping帧
            await self.ws.ping()
            print(f"[{datetime.now()}] 发送Ping")
            await asyncio.sleep(30)  # 每30秒发一次
        except Exception as e:
            print(f"心跳异常: {e}")
            break

4.5 断线重连策略：指数退避

断线重连不是简单的「断了就重连」。如果网络波动频繁，你疯狂重连只会让情况更糟。

我推荐用指数退避策略。说白了就是：第一次断了等1秒，第二次等2秒，第三次等4秒，第四次等8秒……这样既不会把服务器打爆，也能在网络恢复后尽快连上。

代码里我已经实现了这个逻辑，再补充一个带随机抖动的版本：

import random

async def reconnect_with_jitter(self):
    """带随机抖动的指数退避重连"""
    base_delay = 1
    max_delay = 60
    
    for attempt in range(self.max_reconnect):
        # 指数退避 + 随机抖动
        delay = min(base_delay * (2 ** attempt), max_delay)
        jitter = random.uniform(0, delay * 0.5)
        total_delay = delay + jitter
        
        print(f"第 {attempt+1} 次重连，等待 {total_delay:.2f} 秒")
        await asyncio.sleep(total_delay)
        
        try:
            await self.connect()
            print("重连成功！")
            return
        except:
            continue
    
    print("重连失败，请检查网络或交易所状态")

4.6 完整的数据采集引擎

把上面这些整合起来，就是一个完整的数据采集引擎了。我习惯把它封装成一个类，方便复用：

class DataCollector:
    def __init__(self, config):
        self.url = config['ws_url']
        self.symbols = config['symbols']
        self.client = TickDataClient(self.url, self.symbols)
        
    async def start(self):
        """启动采集引擎"""
        # 同时运行接收数据和心跳
        await asyncio.gather(
            self.client.run(),
            self.client.heartbeat()
        )
        
# 使用示例
if __name__ == "__main__":
    config = {
        'ws_url': 'wss://api.exchange.com/ws',
        'symbols': ['BTC/USDT', 'ETH/USDT']
    }
    collector = DataCollector(config)
    asyncio.run(collector.start())

嗯，到这里数据采集引擎的核心就讲完了。WebSocket连接、心跳保活、断线重连，这三板斧用好了，你的Tick数据就能稳定地流进系统里。

我个人觉得，数据采集是整个量化系统里最容易出问题的一环。网络波动、交易所限流、协议变更……各种幺蛾子都可能遇到。但只要把基础打牢，大部分问题都能自动恢复。

---