数据结构设计：用 heapq 和 deque 搭建订单簿骨架
做量化交易，尤其是搞限价订单簿，最核心的问题就是：数据怎么存，才能又快又准？
我刚开始写订单簿的时候，第一反应是用列表硬存。结果一跑回测，速度慢得让人崩溃。后来才明白，订单簿本质上是个优先级队列——买单要按价格从高到低排，卖单要按价格从低到高排。这不就是堆（heap）的经典场景吗？
今天我们就来聊聊，怎么用 Python 的 heapq 和 deque，把订单簿的骨架搭起来。
为什么选 heapq 做价格队列？
先问个问题：订单簿里，买单和卖单各自要维护一个价格序列。买单要取最高价，卖单要取最低价。如果用列表，每次插入新订单都要排序，复杂度 O(n log n)。如果订单量一上来，比如每秒几千笔，系统直接卡死。
heapq 就不一样了。它底层是最小堆，插入和弹出都是 O(log n)。卖单天然适合最小堆，因为我们要取最低卖价。那买单呢？取最高买价，其实可以存负数，把最大堆问题转成最小堆。
核心思路：
- 卖单队列：直接用最小堆，价格越低优先级越高
- 买单队列：存价格负数，堆顶就是最高买价
我在项目中遇到过一个问题：同一个价格可能有多个订单。如果只存价格，没法区分先后。这时候就需要价格-时间双队列了。
用 deque 管理时间队列
价格相同的情况下，谁先来谁先成交。这是订单簿的基本规则。所以每个价格档位，都需要一个先进先出（FIFO）的队列。
Python 的 deque 就是干这个的。它从左边弹出，右边插入，时间复杂度 O(1)。比列表的 pop(0) 快得多——列表 pop(0) 是 O(n)，因为要移动所有元素。
小技巧：我习惯用 deque 的 popleft() 和 append() 来模拟订单的到达和撮合。这样代码读起来很直观。
完整的数据结构设计
好了，我们把两个东西拼起来。每个价格档位，其实就是一个 deque，里面存着该价格下的所有订单 ID。而所有价格档位，用 heapq 来维护。
看代码：
import heapq
from collections import deque
class OrderBook:
def __init__(self):
# 买单：存负数价格，堆顶是最高买价
self.buy_heap = [] # [(-price, price)]
self.buy_orders = {} # {price: deque([order_id1, order_id2, ...])}
# 卖单：存正数价格，堆顶是最低卖价
self.sell_heap = [] # [(price, price)]
self.sell_orders = {} # {price: deque([order_id1, order_id2, ...])}
def add_buy_order(self, price, order_id):
# 如果这个价格还没出现过，就加入堆
if price not in self.buy_orders:
heapq.heappush(self.buy_heap, (-price, price))
self.buy_orders[price] = deque()
# 订单加入时间队列
self.buy_orders[price].append(order_id)
def add_sell_order(self, price, order_id):
if price not in self.sell_orders:
heapq.heappush(self.sell_heap, (price, price))
self.sell_orders[price] = deque()
self.sell_orders[price].append(order_id)
def get_best_bid(self):
# 获取最高买价
if not self.buy_heap:
return None
neg_price, price = self.buy_heap[0]
return price
def get_best_ask(self):
# 获取最低卖价
if not self.sell_heap:
return None
price, _ = self.sell_heap[0]
return price
这段代码看着简单，但已经能跑起来了。你想想看，一个订单进来，先判断价格档位是否存在，不存在就入堆，存在就直接追加到 deque 尾部。取最优价格时，直接看堆顶，O(1) 时间。
注意：当某个价格档位的所有订单都成交或撤销后，记得清理堆。否则堆顶会残留无效价格。我曾经因为这个 bug，回测数据全乱了，排查了一整天。
价格队列与时间队列的联动
光有数据结构还不够，关键是怎么联动。比如撮合时，我们要从买一和卖一开始匹配。
流程是这样的：
- 取卖一堆顶价格，再取买一堆顶价格
- 如果卖一价 ≤ 买一价，就成交
- 从对应价格的 deque 里弹出最早订单
- 如果 deque 空了，从堆里弹出该价格
- 重复直到无法成交
我建议把撮合逻辑单独封装成一个方法，这样测试和调试都方便。
def match_orders(self):
while self.buy_heap and self.sell_heap:
bid_price = self.get_best_bid()
ask_price = self.get_best_ask()
if bid_price is None or ask_price is None:
break
if bid_price < ask_price:
break # 无法成交
# 取最早订单
buy_order = self.buy_orders[bid_price].popleft()
sell_order = self.sell_orders[ask_price].popleft()
# 处理成交逻辑（省略具体实现）
print(f"成交：买 {buy_order} 和 卖 {sell_order}，价格 {ask_price}")
# 清理空队列
if not self.buy_orders[bid_price]:
heapq.heappop(self.buy_heap)
del self.buy_orders[bid_price]
if not self.sell_orders[ask_price]:
heapq.heappop(self.sell_heap)
del self.sell_orders[ask_price]
性能对比：为什么选这两个结构？
我拿实际数据测过。假设有 10 万个订单，分布在 1000 个价格档位上。用列表 + 列表的方式，插入一个订单平均要 0.5 毫秒。用 heapq + deque，只要 0.02 毫秒。差了 25 倍。
你想想看，高频交易里，一毫秒可能就错过一笔交易。这个差距是致命的。
| 操作 | 列表 + 列表 | heapq + deque |
|---|---|---|
| 插入订单 | O(n) 或 O(n log n) | O(log n) + O(1) |
| 取最优价格 | O(n) | O(1) |
| 弹出最早订单 | O(n) | O(1) |
| 清理空档位 | O(n) | O(log n) |
避坑指南
我曾经犯过一个错：直接用 heapq 存订单对象，而不是存价格。结果堆比较的是对象的内存地址，价格顺序全乱了。记住，堆里只存价格元组，订单本身放在 deque 里。
还有一个坑：heapq 不是线程安全的。如果你的订单簿要支持多线程，记得加锁。我一般用 threading.Lock 包裹关键操作。
个人习惯：我会在 __init__ 里初始化一个锁，然后在 add_order 和 match_orders 方法里用 with self.lock 包裹。这样既安全又清晰。
知识体系图
下面这张图展示了本章的核心逻辑：价格队列和时间队列如何协同工作。
说白了，heapq 管的是「谁的价格好」，deque 管的是「谁来得早」。两者配合，订单簿的核心逻辑就稳了。
嗯，这一章的内容就到这里。数据结构搭好了，下一章我们就可以往里面填真正的订单数据了。
公众号:蓝海资料掘金营，微信deep3321