行情模块：数据源的接入与架构设计

做量化交易，行情模块就是系统的眼睛。眼睛要是瞎了，后面策略再牛也没用。今天我就把这块的核心逻辑掰开揉碎了讲给你听。

行情数据源接入

行情数据从哪来？无非两条路：交易所官方API，或者第三方数据商。我个人的习惯是，能直接接交易所就别走第三方——延迟低，数据也干净。

交易所API接入

交易所API通常分两类：REST和WebSocket。REST适合拉历史数据，WebSocket用来收实时行情。举个例子，币安的现货行情接口：

import asyncio
import websockets
import json

class BinanceMarketData:
    def __init__(self, symbol="btcusdt"):
        self.symbol = symbol
        self.ws_url = f"wss://stream.binance.com:9443/ws/{symbol}@depth20@100ms"
    
    async def connect(self):
        async with websockets.connect(self.ws_url) as ws:
            while True:
                data = await ws.recv()
                # 这里解析OrderBook数据
                self.process_orderbook(json.loads(data))
    
    def process_orderbook(self, raw):
        # 数据结构转换
        pass

这里要注意，WebSocket重连机制必须自己写。我在项目中遇到过交易所凌晨维护，连接断了没重连，第二天开盘直接少收半小时数据。嗯，这个坑踩得挺疼。

第三方数据商接入

第三方数据商的好处是省心，一个接口搞定多个交易所。但代价是延迟，通常比直连交易所慢50-200毫秒。做高频交易的话，这个差距就是生与死的区别。

我曾经用某数据商的Level2数据做回测，结果实盘时发现延迟完全对不上。后来查了半天，原来是数据商做了时间戳对齐，把数据"美化"了。说白了，第三方数据适合做分析，不适合做执行。

数据结构设计

行情数据就三种核心结构：OrderBook、Tick、Kline。每种结构的设计都有讲究，我一个个说。

OrderBook（订单簿）

OrderBook就是买卖挂单的快照。设计时要注意：价格精度、数量精度、时间戳精度。我一般用numpy数组来存，查询快，内存占用也小。

from dataclasses import dataclass
import numpy as np

@dataclass
class OrderBook:
    symbol: str
    timestamp: int  # 微秒级时间戳
    bids: np.ndarray  # [[price, qty], ...]
    asks: np.ndarray  # [[price, qty], ...]
    
    def spread(self):
        return self.asks[0][0] - self.bids[0][0]
    
    def mid_price(self):
        return (self.asks[0][0] + self.bids[0][0]) / 2

为什么用numpy？因为OrderBook更新频率极高，每秒可能几百次。用Python原生list的话，内存碎片化严重，GC压力也大。你想想看，一个高频策略跑下来，光GC停顿就能吃掉你10%的性能。

Tick（逐笔成交）

Tick数据是最细粒度的成交记录。每条Tick包含：成交价、成交量、成交方向、成交时间。设计时我建议加上一个trade_id，方便去重。

@dataclass
class Tick:
    symbol: str
    trade_id: int
    price: float
    volume: float
    side: str  # 'buy' or 'sell'
    timestamp: int
    
    @property
    def turnover(self):
        return self.price * self.volume

这里有个坑：不同交易所的trade_id生成规则不一样。有的用全局递增ID，有的用当天递增ID。我曾经因为没注意这个细节，导致去重逻辑写错了，回测结果虚高了不少。

Kline（K线）

K线是Tick数据的聚合。设计时除了OHLCV，我习惯加上成交笔数（count）和成交额（turnover）。这两个字段对判断市场活跃度很有帮助。

@dataclass
class Kline:
    symbol: str
    open_time: int
    close_time: int
    open: float
    high: float
    low: float
    close: float
    volume: float
    count: int
    turnover: float

为什么加成交笔数？举个例子，同样是一万手成交量，分十笔成交和分一百笔成交，市场含义完全不同。前者可能是大单扫货，后者可能是散户跟风。这个细节，很多新手会忽略。

数据缓存与分发机制

行情数据进来之后，不能直接丢给策略模块。中间需要一个缓存层，做三件事：去重、排序、分发。

环形缓冲区设计

我推荐用环形缓冲区（Ring Buffer）来做缓存。原因很简单：内存分配次数少，延迟稳定。

class RingBuffer:
    def __init__(self, capacity=10000):
        self.buffer = [None] * capacity
        self.head = 0
        self.tail = 0
        self.capacity = capacity
    
    def push(self, item):
        self.buffer[self.tail] = item
        self.tail = (self.tail + 1) % self.capacity
        if self.tail == self.head:
            self.head = (self.head + 1) % self.capacity  # 覆盖旧数据
    
    def pop(self):
        if self.head == self.tail:
            return None
        item = self.buffer[self.head]
        self.head = (self.head + 1) % self.capacity
        return item

这个设计的好处是，读写操作都是O(1)复杂度，没有锁竞争（单线程场景下）。我在实盘系统里用了三年，没出过问题。

多路分发机制

行情数据要同时喂给多个策略模块。我用的模式是发布-订阅（Pub-Sub）。每个策略注册一个回调函数，数据到了就广播出去。

class MarketDataDispatcher:
    def __init__(self):
        self.subscribers = {}  # {strategy_id: callback}
    
    def subscribe(self, strategy_id, callback):
        self.subscribers[strategy_id] = callback
    
    def unsubscribe(self, strategy_id):
        self.subscribers.pop(strategy_id, None)
    
    def dispatch(self, data):
        for callback in self.subscribers.values():
            callback(data)

这里要注意：回调函数里不能做耗时操作。我曾经有个策略在回调里写日志，结果行情一快，日志IO把CPU吃满了，其他策略全被拖慢。后来改成异步写日志才解决。

整体架构图

下面这张图展示了行情模块的完整数据流。从数据源接入，到缓存，再到分发，最后到策略模块。

好了，行情模块的核心设计就这些。记住一个原则：数据进来之后，能早处理就早处理，别拖到策略层再做清洗。脏数据进，脏数据出，这个道理在量化里一样适用。

---