第3章数据采集层

数据采集层：WebSocket连接池管理、REST API降级策略、心跳保活机制

数据采集层，说白了就是整个量化系统的「眼睛」。眼睛要是瞎了，后面策略再牛也没用。我做了这么多年交易系统，见过太多因为数据层崩掉导致爆仓的案例。今天咱们就聊聊怎么把这双「眼睛」练得又亮又稳。

WebSocket连接池管理

先说说WebSocket。这东西比REST快多了，数据是「推」过来的，不用你一遍遍去问。但问题来了——你同时连十几个交易所，每个交易所可能还要连多个市场，连接数一多，管理就成了噩梦。

我个人的习惯是搞一个连接池，而不是每个市场单独开连接。连接池的好处很明显：复用连接、控制并发、统一管理生命周期。

核心思路：每个交易所维护一个连接池，池子里放3-5个WebSocket连接。订阅市场数据时，从池子里取一个空闲连接，用完后归还。如果池子满了，就排队等待或者新建临时连接（但要设上限）。

来看个简单的连接池实现：

class WebSocketPool:
    def __init__(self, exchange, max_connections=5):
        self.exchange = exchange
        self.max_connections = max_connections
        self.connections = []
        self.in_use = set()
        self.queue = deque()
    
    async def acquire(self):
        # 先找空闲连接
        for conn in self.connections:
            if conn not in self.in_use and conn.is_connected():
                self.in_use.add(conn)
                return conn
        
        # 没空闲的，看能不能新建
        if len(self.connections) < self.max_connections:
            conn = await self._create_connection()
            self.connections.append(conn)
            self.in_use.add(conn)
            return conn
        
        # 实在不行，排队等
        future = asyncio.Future()
        self.queue.append(future)
        return await future
    
    async def release(self, conn):
        self.in_use.discard(conn)
        if self.queue:
            # 唤醒一个等待的
            waiter = self.queue.popleft()
            waiter.set_result(conn)

这里有个坑——连接断开后要自动重连。我在项目中遇到过，某个交易所半夜升级，连接全断了，池子里全是死连接。后来我加了个健康检查，每次取连接前先ping一下，不健康的直接扔掉重建。

小技巧：连接池的容量不要设太大。我一般每个交易所3-5个就够用了。太多反而增加心跳开销，而且交易所那边也有连接数限制。

REST API降级策略

WebSocket再稳，也有掉链子的时候。比如网络波动、交易所限流、或者干脆就是WebSocket服务挂了。这时候怎么办？

降级到REST API。说白了就是「有鲜肉吃鲜肉，没鲜肉啃馒头」。REST虽然慢一点，但胜在稳定，HTTP协议几乎不会断。

我设计了一套三级降级策略：

级别	触发条件	行为	恢复条件
L0（正常）	WebSocket正常	全量数据走WebSocket	-
L1（轻度降级）	WebSocket延迟 > 500ms	深度数据走REST，成交数据走WebSocket	延迟 < 200ms 持续30秒
L2（重度降级）	WebSocket断开 > 3秒	全部走REST，轮询间隔500ms	WebSocket重连成功
L3（紧急降级）	REST也频繁超时	暂停该交易所数据，使用本地缓存	人工介入或自动恢复

我曾经在L2降级时踩过一个坑——REST轮询频率太高，直接被交易所封了IP。后来我加了个自适应节流：如果连续3次请求都超时，就把轮询间隔翻倍，从500ms慢慢退到2秒、5秒。等恢复后再慢慢降回来。

注意：降级不是无脑切。要记录降级日志，包括触发时间、原因、当前级别。方便事后复盘。我见过有人降级了三天都没发现，因为日志被淹没了。

心跳保活机制

WebSocket连接是长连接，但网络环境复杂，中间可能有防火墙、NAT设备，它们会定期清理空闲连接。所以你得主动「刷存在感」。

心跳机制其实就两件事：

发送心跳包：每隔一段时间（比如30秒）发一个ping帧
检测超时：如果超过N秒没收到pong，就认为连接断了

但这里有个细节——心跳间隔怎么设？

我个人的经验是：

国内交易所：30-45秒（网络相对稳定）
海外交易所：15-20秒（跨境网络波动大）
如果走代理/VPN：10秒（代理节点容易超时）

来看个心跳管理的实现：

class HeartbeatManager:
    def __init__(self, ws, interval=30, timeout=60):
        self.ws = ws
        self.interval = interval
        self.timeout = timeout
        self.last_pong = time.time()
        self._running = False
    
    async def start(self):
        self._running = True
        asyncio.create_task(self._ping_loop())
        asyncio.create_task(self._check_loop())
    
    async def _ping_loop(self):
        while self._running:
            await asyncio.sleep(self.interval)
            try:
                await self.ws.ping()
            except:
                # ping失败，触发重连
                await self._reconnect()
    
    async def _check_loop(self):
        while self._running:
            await asyncio.sleep(5)  # 每5秒检查一次
            if time.time() - self.last_pong > self.timeout:
                # 超时了，强制重连
                await self._reconnect()
    
    async def on_pong(self):
        self.last_pong = time.time()

避坑指南：我曾经遇到过一种情况——心跳包正常，pong也正常，但数据就是收不到。后来发现是交易所的WebSocket网关有bug，只响应ping但不推送数据。所以除了心跳，还要加一个「数据新鲜度检查」：如果超过N秒没收到新数据，也触发重连。

整体架构图

说了这么多，咱们用一张图把这三块串起来：

这张图把整个数据采集层的流程串起来了。从上到下：交易所数据源 → WebSocket连接池 → 降级策略 + 心跳保活 → 最终输出统一数据流。每个环节都是环环相扣的。

总结一下：

连接池解决的是「怎么高效管理多个WebSocket连接」
降级策略解决的是「WebSocket挂了怎么办」
心跳保活解决的是「怎么让连接一直活着」

这三块缺一不可。我见过太多人只关注策略逻辑，忽略了数据层的稳定性。结果行情一波动，数据先崩了，策略再牛也白搭。

嗯，数据采集层就聊到这儿。记住一句话：稳定的数据流，是量化交易的基石。下一层咱们会讲数据清洗和预处理，到时候再细聊。

📚 课程目录

数据采集层：WebSocket连接池管理、REST API降级策略、心跳保活机制

WebSocket连接池管理

REST API降级策略

心跳保活机制

整体架构图