3、实时数据流架构：WebSocket连接、行情数据解析、订单簿维护、交易数据流处理

做市交易，说白了就是跟市场赛跑。你比别人慢一毫秒，可能就亏一套房。我做了这么多年，见过太多因为数据流架构没搭好，导致订单簿错位、报价延迟的惨案。今天咱们就聊聊这个核心中的核心——实时数据流架构。

3.1 WebSocket连接：你的数据生命线

传统的HTTP轮询？别想了。做市交易里，毫秒级的延迟都受不了。WebSocket才是王道。它建立一次连接，后续数据全推过来，省去了握手开销。

我个人习惯用 websockets 这个Python库，轻量又好用。但要注意，交易所的WebSocket服务可不是那么稳定。我在项目中遇到过，某次行情暴涨，交易所的WebSocket直接断连了三次。嗯，这里要注意——重连机制必须自己实现。

核心要点：WebSocket连接必须包含心跳检测、自动重连、断线缓存三大机制。

import asyncio
import websockets
import json

class WebSocketClient:
    def __init__(self, url, ping_interval=20):
        self.url = url
        self.ping_interval = ping_interval
        self.ws = None
        self.reconnect_count = 0
        
    async def connect(self):
        try:
            self.ws = await websockets.connect(self.url)
            self.reconnect_count = 0
            print(f"连接成功: {self.url}")
        except Exception as e:
            print(f"连接失败: {e}")
            await self.reconnect()
    
    async def reconnect(self):
        self.reconnect_count += 1
        wait_time = min(2 ** self.reconnect_count, 30)  # 指数退避
        print(f"等待 {wait_time} 秒后重连...")
        await asyncio.sleep(wait_time)
        await self.connect()
    
    async def receive(self):
        while True:
            try:
                data = await self.ws.recv()
                yield data
            except websockets.ConnectionClosed:
                print("连接断开，准备重连")
                await self.reconnect()
                break

避坑指南：我曾经把重连间隔设成固定5秒，结果交易所维护时，我的程序疯狂重连，直接把IP给封了。后来改成指数退避，再也没出过问题。

3.2 行情数据解析：别让数据在手里烂掉

交易所推过来的数据，通常是JSON格式。但直接解析JSON？太慢了。你想想看，每秒几百上千条行情，每条都做json.loads，CPU很快就扛不住了。

我建议用 ujson 或 orjson 替代标准库的json。实测下来，orjson比标准库快3-5倍。另外，数据解析要放在独立的协程或线程里，别阻塞主循环。

解析库 速度（相对标准库） 内存占用
json（标准库） 1x 中等
ujson 2-3x 较低
orjson 3-5x 最低 
import orjson

def parse_ticker(raw_data: bytes) -> dict:
    """解析行情数据，使用orjson加速"""
    try:
        data = orjson.loads(raw_data)
        # 提取关键字段
        ticker = {
            'symbol': data['s'],
            'price': float(data['c']),
            'volume': float(data['v']),
            'timestamp': data['E']
        }
        return ticker
    except Exception as e:
        print(f"解析失败: {e}")
        return None

3.3 订单簿维护：你的市场地图

订单簿是做市交易的核心数据结构。说白了，就是买卖双方的挂单队列。维护好它，你才能知道市场深度在哪，该往哪报价。

我见过有人用Python的dict来存订单簿，每次更新都全量替换。这其实很浪费。正确的做法是增量更新——交易所推什么，你改什么。

订单簿维护三原则：

  • 增量更新：只处理变化的部分，不重建整个订单簿
  • 快照校验：每N次增量后，请求一次全量快照做对比
  • 时间戳对齐：确保订单簿的时间戳与行情时间戳一致
class OrderBook:
    def __init__(self, symbol: str):
        self.symbol = symbol
        self.bids = {}  # 价格 -> 数量
        self.asks = {}
        self.last_update_id = 0
    
    def apply_snapshot(self, snapshot: dict):
        """应用全量快照"""
        self.bids = {float(p): float(q) for p, q in snapshot['bids']}
        self.asks = {float(p): float(q) for p, q in snapshot['asks']}
        self.last_update_id = snapshot['lastUpdateId']
    
    def apply_update(self, update: dict):
        """应用增量更新"""
        # 检查更新ID是否连续
        if update['U'] != self.last_update_id + 1:
            print("更新ID不连续，需要重新获取快照")
            return False
        
        # 处理买单更新
        for price, qty in update['b']:
            price = float(price)
            qty = float(qty)
            if qty == 0:
                self.bids.pop(price, None)  # 删除该价位
            else:
                self.bids[price] = qty
        
        # 处理卖单更新
        for price, qty in update['a']:
            price = float(price)
            qty = float(qty)
            if qty == 0:
                self.asks.pop(price, None)
            else:
                self.asks[price] = qty
        
        self.last_update_id = update['u']
        return True

注意：订单簿的更新ID必须严格递增。我曾经因为网络抖动，跳过了几个更新ID，结果订单簿里的价格全乱了，报价直接报到了市场中间价之外。那笔亏损，够我记一辈子。

3.4 交易数据流处理：从行情到决策

行情数据进来了，订单簿维护好了，接下来就是交易数据流处理。这一步，是把原始数据变成交易信号的关键环节。

我习惯用管道模式来处理数据流。每个处理环节是一个独立的组件，数据像流水一样流过。这样做的好处是，每个环节都可以独立测试、独立优化。

WebSocket 数据解析 订单簿维护 信号生成 数据流方向 → 异常处理 快照校验

每个环节之间用asyncio的Queue来传递数据。这样即使某个环节处理慢了，也不会阻塞其他环节。说白了，就是解耦。

import asyncio
from dataclasses import dataclass

@dataclass
class MarketData:
    symbol: str
    price: float
    volume: float
    timestamp: int

class DataPipeline:
    def __init__(self):
        self.raw_queue = asyncio.Queue(maxsize=1000)
        self.parsed_queue = asyncio.Queue(maxsize=1000)
        self.signal_queue = asyncio.Queue(maxsize=100)
    
    async def raw_data_handler(self, ws_client):
        """原始数据接收环节"""
        async for raw_data in ws_client.receive():
            await self.raw_queue.put(raw_data)
    
    async def parser_handler(self):
        """数据解析环节"""
        while True:
            raw_data = await self.raw_queue.get()
            parsed = parse_ticker(raw_data)
            if parsed:
                await self.parsed_queue.put(parsed)
    
    async def orderbook_handler(self, orderbook):
        """订单簿维护环节"""
        while True:
            data = await self.parsed_queue.get()
            # 更新订单簿
            orderbook.apply_update(data)
            # 生成信号
            signal = self.generate_signal(orderbook)
            if signal:
                await self.signal_queue.put(signal)
    
    def generate_signal(self, orderbook):
        """生成交易信号"""
        # 这里只是示例，实际策略会更复杂
        best_bid = max(orderbook.bids.keys())
        best_ask = min(orderbook.asks.keys())
        spread = best_ask - best_bid
        
        if spread > 0.01:  # 价差足够大
            return {
                'action': 'quote',
                'bid': best_bid + 0.001,
                'ask': best_ask - 0.001
            }
        return None

经验之谈：每个Queue的maxsize一定要设置。我曾经没设上限，结果行情爆发时，内存直接爆了。设个合理的上限，满了就丢弃旧数据，比系统崩溃强一万倍。

整个数据流架构，说白了就是三个字：快、稳、准。快——数据解析要快，订单簿更新要快；稳——连接要稳，重连要稳；准——数据要准，时间戳要对齐。这三样做好了，你的做市系统就成功了一半。

我记得刚入行时，总觉得这些基础设施不重要，把精力全花在策略上。结果呢？策略再牛，数据跟不上，全是白搭。后来我花了整整三个月，把数据流架构重写了一遍。从那以后，系统再也没出过因为数据问题导致的亏损。

专注资料整理

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