3. 订单簿数据结构设计：Python中的OrderBook类、价格队列、订单节点

做市交易的核心，说白了就是跟订单簿打交道。你每天盯着屏幕上跳动的数字，其实背后就是一个个数据结构在支撑。我个人习惯把订单簿想象成一个巨大的书架——每个价格是一层隔板，上面放着等待成交的订单。今天我们就来亲手搭这个书架。

3.1 为什么需要专门的数据结构？

你可能觉得，用个列表存订单不就完了？嗯，我刚开始也这么想。直到有一次回测时，我发现程序处理100万笔订单花了整整40秒——这在高频场景下完全不可接受。

订单簿的核心需求其实就三个：

• 快速插入：新订单进来，要立刻找到它该去的位置
• 快速删除：订单被成交或取消，要能瞬间移除
• 快速查询：想知道当前最优买卖价，不能遍历整个列表

这三个需求，用Python内置的list或dict都很难同时满足。我踩过这个坑，后来才明白——订单簿需要分层设计。

3.2 三层架构：OrderBook → PriceQueue → OrderNode

我习惯把订单簿拆成三个层次，每一层解决一个具体问题：

3.3 OrderNode：最小的积木块

每个订单就是一个节点。我设计OrderNode时，除了基本字段，还特意加了prev和next指针——这是为了后续实现O(1)删除做准备。

class OrderNode:
    """订单节点 - 双向链表的基本单元"""
    
    def __init__(self, order_id: str, price: float, quantity: float, 
                 side: str, timestamp: int):
        self.order_id = order_id      # 订单唯一标识
        self.price = price            # 价格
        self.quantity = quantity      # 剩余数量
        self.side = side              # 'buy' 或 'sell'
        self.timestamp = timestamp    # 时间戳，用于FIFO
        
        # 双向链表指针
        self.prev = None
        self.next = None
    
    def __repr__(self):
        return f"Order({self.order_id}, {self.price}@{self.quantity})"

3.4 PriceQueue：价格层面的队列管理

同一个价格的所有订单，需要按时间顺序排队。这里我用了双向链表 + 字典的组合：链表维护顺序，字典提供O(1)的节点查找。

class PriceQueue:
    """价格队列 - 管理同一价格的所有订单"""
    
    def __init__(self, price: float):
        self.price = price
        self.head = None    # 队列头部（最早到达的订单）
        self.tail = None    # 队列尾部（最新到达的订单）
        self.order_map = {} # order_id -> OrderNode 的映射
        self.volume = 0     # 该价格的总挂单量
        self.count = 0      # 订单数量
    
    def append(self, order: OrderNode):
        """将订单添加到队列尾部"""
        self.order_map[order.order_id] = order
        self.volume += order.quantity
        self.count += 1
        
        if self.tail is None:
            self.head = self.tail = order
        else:
            self.tail.next = order
            order.prev = self.tail
            self.tail = order
    
    def remove(self, order_id: str) -> OrderNode:
        """O(1)删除指定订单"""
        order = self.order_map.pop(order_id, None)
        if order is None:
            return None
        
        self.volume -= order.quantity
        self.count -= 1
        
        # 更新前后节点的指针
        if order.prev:
            order.prev.next = order.next
        if order.next:
            order.next.prev = order.prev
        if order == self.head:
            self.head = order.next
        if order == self.tail:
            self.tail = order.prev
        
        order.prev = order.next = None
        return order
    
    def __len__(self):
        return self.count

3.5 OrderBook：整合一切的主类

现在我们把所有零件组装起来。OrderBook类负责管理所有价格层级，并提供买卖盘口的查询接口。

class OrderBook:
    """订单簿主类 - 管理所有价格层级"""
    
    def __init__(self):
        # 买盘：价格降序排列（最高价优先）
        self.bids = {}  # price -> PriceQueue
        self.bid_prices = []  # 维护价格排序
        
        # 卖盘：价格升序排列（最低价优先）
        self.asks = {}  # price -> PriceQueue
        self.ask_prices = []  # 维护价格排序
        
        # 全局订单索引
        self.orders = {}  # order_id -> (price, side)
    
    def add_order(self, order_id: str, price: float, 
                  quantity: float, side: str, timestamp: int):
        """添加新订单"""
        order = OrderNode(order_id, price, quantity, side, timestamp)
        
        # 选择买卖盘
        book = self.bids if side == 'buy' else self.asks
        prices = self.bid_prices if side == 'buy' else self.ask_prices
        
        # 如果该价格不存在，创建新的价格队列
        if price not in book:
            book[price] = PriceQueue(price)
            # 插入排序维护价格列表
            self._insert_price(prices, price, side)
        
        # 添加到价格队列
        book[price].append(order)
        self.orders[order_id] = (price, side)
    
    def cancel_order(self, order_id: str):
        """取消订单"""
        if order_id not in self.orders:
            return
        
        price, side = self.orders.pop(order_id)
        book = self.bids if side == 'buy' else self.asks
        
        queue = book[price]
        queue.remove(order_id)
        
        # 如果价格队列为空，清理掉
        if len(queue) == 0:
            del book[price]
            prices = self.bid_prices if side == 'buy' else self.ask_prices
            prices.remove(price)
    
    def get_best_bid(self) -> tuple:
        """获取最优买价"""
        if not self.bid_prices:
            return None
        best_price = self.bid_prices[0]
        return best_price, self.bids[best_price]
    
    def get_best_ask(self) -> tuple:
        """获取最优卖价"""
        if not self.ask_prices:
            return None
        best_price = self.ask_prices[0]
        return best_price, self.asks[best_price]
    
    def _insert_price(self, prices: list, price: float, side: str):
        """维护价格排序（买盘降序，卖盘升序）"""
        if side == 'buy':
            # 降序插入
            for i, p in enumerate(prices):
                if price > p:
                    prices.insert(i, price)
                    return
            prices.append(price)
        else:
            # 升序插入
            for i, p in enumerate(prices):
                if price < p:
                    prices.insert(i, price)
                    return
            prices.append(price)

3.6 实战中的性能考量

这个数据结构在实际运行中表现如何？我拿真实市场数据做过测试：

3.7 一些实用的小改进

在实际做市系统中，我还会加几个小功能：

• 快照缓存：每次查询深度时都重新计算太慢，我习惯维护一个缓存，只在订单簿变化时更新
• 价格精度处理：不同交易所的价格精度不一样，我建议用整数存储（比如把价格乘以10000），避免浮点数比较的坑
• 事件回调：每次订单簿变化时触发回调，方便做策略的实时响应

嗯，这套数据结构我用了好几年，从加密货币到股票市场都验证过。你想想看，核心其实就三个字——分层。把复杂问题拆成小块，每个小块只做好一件事。这就是做市系统的地基，地基稳了，上面的策略才能跑得安心。

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