4、数据采集架构：WebSocket实时行情、REST API补全、多交易所数据对齐、数据缓存策略

做跨交易所套利，说白了就是跟数据赛跑。

你比别人快1毫秒，可能就多赚一笔。你数据没对齐，可能就亏一单。我见过太多团队，策略模型写得漂漂亮亮，结果死在数据采集这个环节上。嗯，今天咱们就把这块硬骨头啃下来。

4.1 为什么数据采集是套利系统的命门？

套利的本质是什么？是发现同一资产在不同交易所的价格差异。如果数据不准、不快、不对齐，你看到的价差就是假的。

我个人习惯把数据采集架构比作「水管系统」：

  • WebSocket 是主水管，负责源源不断输送实时水流
  • REST API 是备用水龙头，主水管断了或者水压不够时，用它补上
  • 数据对齐 是水表校准，确保两边读数在同一时间点
  • 缓存策略 是蓄水池，防止下游用水忽大忽小把水管冲爆

我在项目中遇到过最惨的一次：因为WebSocket断线没及时发现，系统用旧数据跑了3分钟，直接亏了6位数。从那以后，我对数据采集的敬畏心就刻进骨子里了。

4.2 WebSocket实时行情：你的第一道防线

WebSocket是双向通信协议，交易所主动往你这边推数据。延迟低、带宽省，是实时行情的首选。

4.2.1 连接管理

别以为连上就完事了。WebSocket会断线、会重连、会心跳超时。我建议你这样设计：

class WebSocketManager:
    def __init__(self, url, channels):
        self.url = url
        self.channels = channels
        self.ws = None
        self.reconnect_attempts = 0
        self.max_attempts = 10
        
    async def connect(self):
        while self.reconnect_attempts < self.max_attempts:
            try:
                self.ws = await websockets.connect(self.url)
                # 订阅频道
                for ch in self.channels:
                    await self.ws.send(json.dumps({
                        "op": "subscribe",
                        "channel": ch
                    }))
                self.reconnect_attempts = 0
                await self.listen()
            except Exception as e:
                self.reconnect_attempts += 1
                wait_time = min(2 ** self.reconnect_attempts, 60)
                print(f"断线重连，第{self.reconnect_attempts}次，等待{wait_time}秒")
                await asyncio.sleep(wait_time)
⚠️ 我曾经踩过的坑： 重连间隔不要固定！用指数退避策略，第一次等1秒，第二次2秒，第三次4秒……最大不超过60秒。否则交易所会把你当DDoS攻击封IP。

4.2.2 心跳机制

很多交易所要求客户端定期发ping，服务器回pong。如果连续几次没收到pong，就该主动断线重连了。

async def heartbeat(self):
    while True:
        await asyncio.sleep(15)  # 每15秒发一次ping
        try:
            await self.ws.ping()
        except:
            print("心跳失败，触发重连")
            break

4.2.3 数据解析与去重

WebSocket推送的数据，有时候会重复。比如交易所内部做了容错，同一个tick推了两次。你需要在内存里维护一个简单的去重机制：

# 用时间戳+交易对+价格作为唯一键
seen = set()
def dedup(tick):
    key = f"{tick['symbol']}_{tick['price']}_{tick['timestamp']}"
    if key in seen:
        return False
    seen.add(key)
    # 控制seen大小，防止内存泄漏
    if len(seen) > 10000:
        seen.clear()
    return True

4.3 REST API补全：兜底方案

WebSocket再稳，也有断的时候。或者某些历史数据、深度数据，WebSocket不推。这时候REST API就是你的救兵。

4.3.1 什么时候用REST？

  • 启动时：先拉一次全量深度数据，再切到WebSocket增量更新
  • 断线重连后：补全断线期间丢失的数据
  • 定时校验：每5分钟用REST拉一次快照，跟WebSocket数据做交叉验证

4.3.2 限频管理

交易所对REST API都有频率限制。比如币安是1200次/分钟，火币是100次/秒。超了会被封IP。

我建议用令牌桶算法做限流：

import time
from collections import deque

class RateLimiter:
    def __init__(self, max_calls, period=1.0):
        self.max_calls = max_calls
        self.period = period
        self.calls = deque()
    
    def acquire(self):
        now = time.time()
        # 清理过期记录
        while self.calls and self.calls[0] < now - self.period:
            self.calls.popleft()
        if len(self.calls) >= self.max_calls:
            sleep_time = self.calls[0] + self.period - now
            if sleep_time > 0:
                time.sleep(sleep_time)
        self.calls.append(time.time())
💡 小技巧： 把限流器做成装饰器，一行代码就能给任意API加上限流功能。我项目中就是这么干的，清爽得很。

4.4 多交易所数据对齐：最容易被忽视的环节

你想想看，交易所A的订单簿更新了，交易所B的订单簿也更新了。但这两个更新可能差了200毫秒。如果你直接用这两个快照算价差，那就是在算假账。

4.4.1 时间戳对齐

每个交易所的时间戳精度不一样。有的精确到毫秒，有的精确到微秒。我建议统一转成UTC毫秒时间戳：

def align_timestamp(ts, exchange):
    if exchange == 'binance':
        # 币安返回的是毫秒时间戳
        return ts
    elif exchange == 'huobi':
        # 火币返回的是秒时间戳
        return ts * 1000
    elif exchange == 'okx':
        # OKX返回的是字符串格式
        return int(datetime.fromisoformat(ts).timestamp() * 1000)

4.4.2 快照对齐策略

我个人习惯的做法是：

  1. 维护一个「对齐窗口」，比如100毫秒
  2. 收到交易所A的快照，不立即使用，等100毫秒
  3. 在这100毫秒内，收集交易所B的快照
  4. 取时间戳最接近的两个快照，组成一对
核心原则： 宁可慢一点，也要用对齐的数据。用不对齐的数据做交易，比没有数据更危险。

4.5 数据缓存策略：别让CPU闲着等网络

数据采集是IO密集型操作。如果每次计算都去拉网络数据，CPU大部分时间都在空转。缓存就是解决这个问题的。

4.5.1 本地缓存层级

层级 存储介质 容量 延迟 用途
L1 内存字典 几百条 纳秒级 最新tick、订单簿快照
L2 Redis 几万条 微秒级 分钟级K线、历史深度
L3 本地文件/数据库 海量 毫秒级 回测数据、日志归档

4.5.2 缓存更新策略

我推荐用「写穿透」模式：数据写入缓存的同时，也写入后端存储。这样即使缓存挂了，数据也不会丢。

class CacheLayer:
    def __init__(self, redis_client):
        self.cache = {}  # L1内存缓存
        self.redis = redis_client  # L2 Redis缓存
    
    def set(self, key, value, ttl=5):
        # 同时写入L1和L2
        self.cache[key] = value
        self.redis.setex(key, ttl, json.dumps(value))
    
    def get(self, key):
        # 先从L1读，没有再从L2读
        if key in self.cache:
            return self.cache[key]
        data = self.redis.get(key)
        if data:
            self.cache[key] = json.loads(data)
            return self.cache[key]
        return None
⚠️ 注意： 缓存一定要设置过期时间（TTL）。行情数据是时效性极强的，5秒前的深度数据可能已经没用了。我见过有人把TTL设成1小时，结果用半小时前的数据做交易，那画面太美不敢看。

4.6 整体架构图

下面这张图，是我做套利系统时画的第一张架构图。它帮你把上面讲的所有东西串起来：

数据采集架构总览 交易所A WebSocket + REST 交易所B WebSocket + REST 交易所C WebSocket + REST 数据采集层 WebSocket连接管理 REST API补全 心跳/重连/去重 数据对齐层 时间戳统一 | 快照对齐 | 交叉校验 数据缓存层 L1: 内存字典 L2: Redis L3: 本地存储 策略引擎（消费数据）

4.7 避坑指南

最后，分享几个我用真金白银换来的教训：

  • 不要迷信WebSocket的稳定性。我曾经以为币安的WebSocket稳如老狗，结果某天凌晨3点断了45分钟，监控报警都没响（因为监控也是走WebSocket）。现在我的做法是：WebSocket和REST双通道互相监控。
  • 数据对齐的误差要量化。别只说「尽量对齐」，要明确写出「允许最大时间差50毫秒」。超过这个阈值的快照对，直接丢弃。
  • 缓存不是越大越好。行情数据过期如废纸。我见过有人把Redis的maxmemory设成10GB，结果存了3天的tick数据，查询慢得像蜗牛。合理设置TTL和淘汰策略，比堆硬件更重要。
  • 日志要打全。每条数据的来源、时间戳、处理耗时，都记下来。出问题的时候，这些日志就是你的救命稻草。

数据采集这块，说白了就是「稳、准、快」三个字。稳是不断线，准是不失真，快是低延迟。把这三点做到位，你的套利系统就成功了一半。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →