3. 实时数据采集层：交易所WebSocket连接管理、行情数据订阅与解析、订单状态同步、数据去重与缓存策略

做市商系统里，数据采集层是地基。地基不稳，上面再漂亮的策略逻辑都是空中楼阁。我见过太多团队，策略模型跑得飞起，结果一上实盘，WebSocket断连、数据乱序、订单状态不同步……直接崩盘。

这一章，咱们就聊聊怎么把这层地基打扎实。说白了，就是四件事：连得上、收得全、解析对、存得稳。

3.1 交易所WebSocket连接管理

WebSocket这东西，看着简单，用起来全是坑。你以为连上就完事了？我告诉你，真正的挑战在连接之后。

3.1.1 连接的生命周期

一个健康的WebSocket连接，要经历这几个阶段：

  1. 握手阶段：建立TCP连接，发送HTTP Upgrade请求
  2. 认证阶段：发送API Key和签名（私有频道需要）
  3. 订阅阶段：发送订阅消息，指定要收的数据类型
  4. 心跳维持：定期发送Ping，接收Pong
  5. 重连机制：断了怎么办？自动重连，还要带指数退避

我个人习惯，把连接管理封装成一个类。这样每个交易所实例都有自己的连接状态，互不干扰。

class WebSocketManager:
    def __init__(self, exchange_name, config):
        self.exchange = exchange_name
        self.ws_url = config['ws_url']
        self.api_key = config.get('api_key')
        self.secret = config.get('secret')
        self.ws = None
        self.reconnect_attempts = 0
        self.max_attempts = 10
        self.subscriptions = set()
        
    async def connect(self):
        """建立连接，带重试机制"""
        while self.reconnect_attempts < self.max_attempts:
            try:
                self.ws = await websockets.connect(self.ws_url)
                await self._authenticate()
                await self._resubscribe()
                self.reconnect_attempts = 0
                return True
            except Exception as e:
                self.reconnect_attempts += 1
                wait_time = min(2 ** self.reconnect_attempts, 60)
                await asyncio.sleep(wait_time)
        return False
注意：重连的指数退避一定要加随机抖动。我曾经遇到过，系统崩溃后所有实例同时重连，直接把交易所网关打挂了。加个0~1秒的随机偏移，能救命。

3.1.2 多路复用与连接池

做市商通常要同时监控多个交易对。每个交易对开一个WebSocket？别闹。一来浪费资源，二来容易被交易所限流。

我建议的做法是：

  • 公共频道共享连接：行情数据走一个连接，订阅多个交易对
  • 私有频道独立连接：订单更新、账户变动走另一个连接
  • 连接池管理：每个交易所维护2~3个长连接，按频道类型分配

你想想看，如果每个交易对都开一个连接，同时做20个币对，就是20个连接。交易所那边一看，直接给你限流。合并订阅，一个连接搞定所有行情，多清爽。

3.2 行情数据订阅与解析

连上之后，第一件事就是订阅数据。不同交易所的订阅格式千奇百怪，但核心逻辑是一样的。

3.2.1 订阅消息格式

拿币安和OKX举个例子：

交易所 订阅格式 示例
币安 JSON数组，包含方法、参数 {"method":"SUBSCRIBE","params":["btcusdt@depth20"],"id":1}
OKX JSON对象，包含操作、参数 {"op":"subscribe","args":[{"channel":"books","instId":"BTC-USDT"}]}
Bybit JSON对象，包含操作、参数 {"op":"subscribe","args":["orderbook.200.100ms.BTCUSDT"]}

这里有个坑：订阅成功不一定有确认消息。有些交易所会返回一个ack，有些直接开始推数据。我建议统一加一个订阅确认的超时机制，5秒内没收到数据就重试。

3.2.2 数据解析与标准化

行情数据到了，但每个交易所的字段名、格式都不一样。币安叫"bids"，OKX叫"bids"但结构不同，Bybit叫"b"……

我的做法是：统一内部数据模型。不管外面怎么变，进到系统里都是一个标准格式。

class OrderBook:
    def __init__(self, symbol):
        self.symbol = symbol
        self.bids = {}  # price -> size
        self.asks = {}
        self.timestamp = 0
        self.update_id = 0
        
    @classmethod
    def from_binance(cls, data):
        """从币安原始数据解析"""
        book = cls(data['s'])
        for bid in data['b']:
            book.bids[float(bid[0])] = float(bid[1])
        for ask in data['a']:
            book.asks[float(ask[0])] = float(ask[1])
        book.timestamp = data['E']
        book.update_id = data['u']
        return book
        
    @classmethod
    def from_okx(cls, data):
        """从OKX原始数据解析"""
        book = cls(data['arg']['instId'])
        for bid in data['data'][0]['bids']:
            book.bids[float(bid[0])] = float(bid[1])
        for ask in data['data'][0]['asks']:
            book.asks[float(ask[0])] = float(ask[1])
        book.timestamp = data['data'][0]['ts']
        book.update_id = data['data'][0]['seqId']
        return book
经验之谈：解析层一定要做字段校验。我遇到过交易所推送空数组、字段缺失、甚至类型错误的情况。加一层try-except，解析失败就丢弃，别让脏数据污染下游。

3.3 订单状态同步

行情数据丢了可以重拉，订单状态丢了……那就是真金白银的损失了。

订单状态同步，说白了就是：你下的单，交易所到底执行了没有？

3.3.1 订单生命周期

一个订单从创建到完成，通常经历这些状态：

  • NEW：刚提交，还没成交
  • PARTIALLY_FILLED：部分成交
  • FILLED：完全成交
  • CANCELED：被取消
  • REJECTED：被拒绝（价格不对、余额不足等）
  • EXPIRED：过期（比如IOC订单）

每个状态变化，交易所都会推送一条消息。但问题是：消息可能丢。

3.3.2 状态同步策略

我常用的策略是双通道同步：

  1. WebSocket实时推送：第一时间收到状态变化
  2. REST API定期轮询：每隔几秒拉一次未成交订单列表，做对账

为什么会这样？因为WebSocket虽然快，但可能断连。断连期间的状态变化，全靠REST API补回来。

class OrderSyncManager:
    def __init__(self, ws_manager, rest_client):
        self.ws = ws_manager
        self.rest = rest_client
        self.pending_orders = {}  # order_id -> Order
        self.sync_interval = 5  # 秒
        
    async def on_order_update(self, ws_message):
        """处理WebSocket推送的订单更新"""
        order = self._parse_order(ws_message)
        self.pending_orders[order.id] = order
        # 触发策略层的回调
        await self.strategy.on_order_changed(order)
        
    async def periodic_sync(self):
        """定期用REST API做对账"""
        while True:
            await asyncio.sleep(self.sync_interval)
            open_orders = await self.rest.get_open_orders()
            # 对比本地状态和交易所状态
            for order_id, local_order in self.pending_orders.items():
                if order_id not in open_orders:
                    # 本地有但交易所没有，说明已经成交或取消
                    # 需要补拉成交记录
                    filled = await self.rest.get_order_fills(order_id)
                    self._sync_order_status(local_order, filled)
避坑指南：我曾经遇到过一个情况，WebSocket推送了订单成交，但REST API查询时订单还在。原因是交易所内部状态同步有延迟。这时候要以WebSocket为准，但要做状态机校验，防止重复处理。

3.4 数据去重与缓存策略

数据到了，但你可能收到重复数据。为什么？

  • WebSocket重连后，交易所会重推断连期间的数据
  • 同一个订单状态变化，可能推送多次
  • 行情数据更新频繁，部分交易所会推增量更新

3.4.1 去重策略

去重，核心是唯一标识。不同数据类型，唯一标识不同：

数据类型 唯一标识 去重方式
订单更新 order_id + 状态 + 更新时间 用字典缓存，key为组合标识
行情快照 交易对 + 更新序号 比较序号，只处理更大的序号
成交记录 trade_id 用集合缓存已处理的trade_id

我个人习惯用LRU缓存来做去重。设置一个合理的缓存大小，比如10000条。超过的自动淘汰，防止内存爆炸。

from collections import OrderedDict

class DedupCache:
    def __init__(self, max_size=10000):
        self.cache = OrderedDict()
        self.max_size = max_size
        
    def is_duplicate(self, key):
        """检查是否重复，并更新缓存"""
        if key in self.cache:
            self.cache.move_to_end(key)
            return True
        self.cache[key] = True
        if len(self.cache) > self.max_size:
            self.cache.popitem(last=False)
        return False

3.4.2 缓存策略

缓存不只是为了去重。它还承担着数据缓冲和快速查询的作用。

我常用的缓存分层：

  • L1：内存缓存（毫秒级）—— 最新的行情快照、订单簿、账户余额
  • L2：本地文件/数据库（秒级）—— 历史行情切片、订单日志
  • L3：远程存储（分钟级）—— 归档数据，用于回测和分析

你想想看，策略层要查当前最优买卖价，如果每次都去数据库查，黄花菜都凉了。内存里存一份最新的，查起来就是O(1)。

class MarketDataCache:
    def __init__(self):
        self.orderbooks = {}  # symbol -> OrderBook
        self.tickers = {}     # symbol -> Ticker
        self.trades = {}      # symbol -> deque(maxlen=1000)
        
    def update_orderbook(self, symbol, orderbook):
        """更新订单簿缓存"""
        self.orderbooks[symbol] = orderbook
        
    def get_best_bid_ask(self, symbol):
        """获取最优买卖价"""
        book = self.orderbooks.get(symbol)
        if not book:
            return None, None
        best_bid = max(book.bids.keys()) if book.bids else None
        best_ask = min(book.asks.keys()) if book.asks else None
        return best_bid, best_ask
小技巧：缓存一定要设置过期时间。行情数据超过1秒没更新，说明连接可能断了。这时候要触发告警，而不是让策略用过期数据做决策。

3.5 整体架构图

说了这么多，咱们用一张图把整个数据采集层的架构串起来：

实时数据采集层架构图 交易所数据源 币安 WebSocket OKX WebSocket Bybit WebSocket REST API WebSocket连接管理 连接池 心跳检测 自动重连(指数退避) 数据解析与标准化 行情解析 订单状态解析 统一内部数据模型 数据去重与缓存 LRU去重 L1内存缓存 L2本地存储

这张图把整个数据采集层的流程串起来了。从上到下，数据从交易所流出，经过连接管理、解析标准化、去重缓存，最终变成干净、有序、可用的数据，供策略层和监控层使用。

嗯，数据采集层就聊到这儿。记住一句话：数据质量决定策略上限。采集层做得越扎实，后面的路就越顺。

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

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