4. 行情数据接入：WebSocket实时行情接入、行情数据清洗与标准化、行情数据缓存策略

做市商系统里，行情数据就是你的眼睛。眼睛要是瞎了，策略再牛也白搭。这一章，我跟你聊聊怎么把这双「眼睛」擦亮。

4.1 WebSocket实时行情接入

先说说接入方式。RESTful API 轮询？别闹了。做市商系统对延迟的要求，轮询根本扛不住。WebSocket 才是正解。

我个人习惯用 websocket-client 这个库。轻量、稳定，踩过的坑少。下面是我常用的接入模板：

import websocket
import json
import threading

class MarketDataFeed:
    def __init__(self, url, symbols):
        self.url = url
        self.symbols = symbols
        self.ws = None
        self.running = False

    def on_message(self, ws, message):
        # 收到原始数据，先别急着用
        data = json.loads(message)
        self.raw_data_queue.put(data)

    def on_error(self, ws, error):
        # 我曾经在这栽过跟头
        # 网络闪断时，错误回调会疯狂触发
        # 一定要加重连逻辑
        print(f"WebSocket error: {error}")
        self.reconnect()

    def on_close(self, ws, close_status_code, close_msg):
        print("WebSocket closed, reconnecting...")
        self.reconnect()

    def on_open(self, ws):
        # 订阅行情
        subscribe_msg = {
            "op": "subscribe",
            "args": self.symbols
        }
        ws.send(json.dumps(subscribe_msg))

    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.running = True
        # 用独立线程跑，别阻塞主流程
        wst = threading.Thread(target=self.ws.run_forever)
        wst.daemon = True
        wst.start()

    def reconnect(self):
        if self.running:
            self.connect()

4.2 行情数据清洗与标准化

数据接进来了，但能用吗？不一定。不同交易所的行情格式五花八门。有的用毫秒时间戳，有的用微秒。有的叫 bid，有的叫 buy。不统一，就没法用。

我建议你做一个统一的行情模型：

from dataclasses import dataclass
from datetime import datetime

@dataclass
class StandardizedTick:
    exchange: str        # 交易所标识
    symbol: str          # 统一后的交易对名称
    bid: float           # 买一价
    ask: float           # 卖一价
    bid_size: float      # 买一量
    ask_size: float      # 卖一量
    last_price: float    # 最新成交价
    volume: float        # 24h成交量
    timestamp: int       # 统一为毫秒时间戳
    local_time: int      # 本地接收时间

class DataCleaner:
    def __init__(self):
        self.symbol_map = {
            "BTCUSDT": "BTC-USDT",
            "btcusdt": "BTC-USDT",
            "BTC/USDT": "BTC-USDT"
        }

    def clean(self, raw_data, exchange):
        # 清洗逻辑
        # 1. 过滤异常数据
        if not self._validate(raw_data):
            return None

        # 2. 标准化交易对名称
        symbol = self._normalize_symbol(raw_data.get("symbol"))

        # 3. 统一时间戳
        ts = self._unify_timestamp(raw_data.get("timestamp"))

        # 4. 处理缺失字段
        bid = float(raw_data.get("bid", 0))
        ask = float(raw_data.get("ask", 0))

        # 5. 检查价差合理性
        if ask <= bid:
            # 我曾经遇到过交易所返回的卖价低于买价
            # 这种数据直接丢弃
            return None

        return StandardizedTick(
            exchange=exchange,
            symbol=symbol,
            bid=bid,
            ask=ask,
            bid_size=float(raw_data.get("bid_size", 0)),
            ask_size=float(raw_data.get("ask_size", 0)),
            last_price=float(raw_data.get("last_price", 0)),
            volume=float(raw_data.get("volume", 0)),
            timestamp=ts,
            local_time=int(time.time() * 1000)
        )

    def _validate(self, data):
        # 必填字段检查
        required = ["symbol", "bid", "ask", "timestamp"]
        for field in required:
            if field not in data:
                return False
        return True

4.3 行情数据缓存策略

数据清洗完了，存哪？直接写数据库？太慢了。行情数据每秒几百上千条，数据库根本扛不住。

我推荐用环形缓冲区（Ring Buffer）。说白了就是一个固定大小的数组，新数据覆盖旧数据。优点是零分配、无锁、速度快。

import array
from collections import deque

class TickRingBuffer:
    def __init__(self, capacity=1024):
        self.capacity = capacity
        self.buffer = [None] * capacity
        self.head = 0
        self.tail = 0
        self.size = 0

    def push(self, tick):
        self.buffer[self.head] = tick
        self.head = (self.head + 1) % self.capacity
        if self.size < self.capacity:
            self.size += 1
        else:
            # 覆盖旧数据时，移动tail
            self.tail = (self.tail + 1) % self.capacity

    def get_latest(self, n=1):
        """获取最近n条数据"""
        if n > self.size:
            n = self.size
        result = []
        idx = (self.head - n) % self.capacity
        for _ in range(n):
            result.append(self.buffer[idx])
            idx = (idx + 1) % self.capacity
        return result

    def get_snapshot(self):
        """获取当前快照（最新一条）"""
        if self.size == 0:
            return None
        return self.buffer[(self.head - 1) % self.capacity]

你想想看，为什么不用队列？队列虽然方便，但每次 push 都会分配内存。在高频场景下，GC 会成为性能杀手。环形缓冲区从一开始就把内存分配好了，运行期间零分配。

4.4 整体架构图

下面这张图，是我做这套系统时的核心设计思路。你看一眼就明白了：

整个流程其实就三步：接进来、洗干净、存起来。但每一步都有坑。嗯，这里要注意，缓存层是承上启下的关键。它既要扛住上游的洪峰流量，又要保证下游能拿到最新的数据。

4.5 实战中的几个坑

最后，分享几个我踩过的坑：

• 心跳检测：WebSocket 连接久了会静默断开。一定要实现心跳机制，每隔几秒发个 ping。
• 数据乱序：同一个交易对的数据，可能后发的先到。我建议用时间戳排序，或者干脆丢弃旧数据。
• 内存泄漏：环形缓冲区虽然快，但如果消费者跟不上，数据会被覆盖。要监控缓存命中率。
• 交易所限流：有些交易所对 WebSocket 连接数有限制。我曾经同时开了 20 个连接，直接被封 IP。