订单簿数据结构设计:用Python字典和列表搭建核心骨架

做量化交易的朋友都知道,订单簿是整个交易系统的「心脏」。它实时记录着所有买单和卖单的排队情况。说白了,就是一张动态的「价格-数量」映射表。

我个人习惯用Python的字典和列表来构建这个核心数据结构。为什么?因为字典的键值对天然适合做价格-数量映射,而列表的队列特性正好用来管理订单的先进先出。我在项目中试过用数据库、用Redis,但说实话,对于仿真模拟来说,纯Python的数据结构反而最灵活、最可控。

1. 价格-数量映射:字典的妙用

先看一个最简单的例子。假设我们有一个卖单队列,价格从低到高排列:

# 卖单队列:价格 -> 数量
ask_book = {
    100.5: 200,   # 价格100.5元,有200股要卖
    101.0: 150,
    102.3: 300
}

# 买单队列:价格 -> 数量
bid_book = {
    99.8: 500,
    99.5: 300,
    99.0: 100
}

嗯,这里要注意。字典的键是价格,值是该价格上的总挂单数量。为什么用字典?因为查询某个价格上的挂单量是O(1)的,速度极快。我在做高频仿真时,每秒要处理几万笔订单,这个性能优势就体现出来了。

核心要点:字典的键必须是浮点数或整数,不能是字符串。价格精度要统一,建议用Decimal或保留两位小数。

2. 订单队列管理:列表的先进先出

光有价格-数量映射还不够。同一个价格上,可能有多个订单。比如100.5元这个价位,张三挂了100股,李四挂了50股,王五挂了50股。这时候就需要订单队列了。

我建议用列表来模拟队列。每个订单是一个字典,包含订单ID、数量、时间戳等信息:

# 每个价格对应的订单队列
order_queue = {
    100.5: [
        {'order_id': 'A001', 'quantity': 100, 'timestamp': 1001},
        {'order_id': 'A002', 'quantity': 50,  'timestamp': 1002},
        {'order_id': 'A003', 'quantity': 50,  'timestamp': 1003}
    ],
    101.0: [
        {'order_id': 'B001', 'quantity': 150, 'timestamp': 1004}
    ]
}

为什么会这样设计?因为订单簿的核心规则就是「价格优先、时间优先」。同一个价格上,先来的订单先成交。列表的append操作在尾部添加新订单,pop(0)从头部取出订单,完美模拟了队列行为。

我的经验:曾经我直接用list.pop(0)来出队,结果在百万级订单量时性能崩了。后来改用collections.deque,出队操作从O(n)降到了O(1)。如果你做高频仿真,记得用deque。

3. 完整的订单簿数据结构

把上面两个结构组合起来,就得到了一个完整的订单簿:

class OrderBook:
    def __init__(self):
        # 买单:价格降序排列(最高价优先)
        self.bids = {}      # 价格->数量映射
        self.bid_queues = {} # 价格->订单队列
        
        # 卖单:价格升序排列(最低价优先)
        self.asks = {}      # 价格->数量映射
        self.ask_queues = {} # 价格->订单队列
    
    def add_order(self, side, price, quantity, order_id):
        """添加订单到订单簿"""
        if side == 'buy':
            book = self.bids
            queues = self.bid_queues
        else:
            book = self.asks
            queues = self.ask_queues
        
        # 更新价格-数量映射
        book[price] = book.get(price, 0) + quantity
        
        # 添加到订单队列
        if price not in queues:
            queues[price] = []
        queues[price].append({
            'order_id': order_id,
            'quantity': quantity,
            'timestamp': time.time()
        })
    
    def remove_order(self, side, price, order_id):
        """移除订单"""
        queues = self.bid_queues if side == 'buy' else self.ask_queues
        book = self.bids if side == 'buy' else self.asks
        
        if price in queues:
            queue = queues[price]
            # 找到并移除指定订单
            for i, order in enumerate(queue):
                if order['order_id'] == order_id:
                    removed_qty = order['quantity']
                    queue.pop(i)
                    # 更新价格-数量映射
                    book[price] -= removed_qty
                    if book[price] <= 0:
                        del book[price]
                        del queues[price]
                    break

你看,核心逻辑其实就这几行。但我在实际项目中踩过不少坑。比如价格精度问题——浮点数比较时,100.5和100.5000001可能被认为是不同的键。我建议统一用整数表示价格,比如把100.5元存为10050(乘以100),这样既避免了精度问题,又提高了字典查询速度。

4. 价格排序与最佳买卖价

订单簿最重要的功能之一,就是快速获取最佳买卖价(即买一价和卖一价)。

对于买单,价格越高越优先,所以我们需要降序排列。对于卖单,价格越低越优先,所以需要升序排列。

def get_best_bid(self):
    """获取最佳买价(最高买单价格)"""
    if not self.bids:
        return None
    return max(self.bids.keys())

def get_best_ask(self):
    """获取最佳卖价(最低卖单价格)"""
    if not self.asks:
        return None
    return min(self.asks.keys())

def get_market_depth(self, levels=5):
    """获取市场深度(前N档价格)"""
    bid_prices = sorted(self.bids.keys(), reverse=True)[:levels]
    ask_prices = sorted(self.asks.keys())[:levels]
    
    depth = {
        'bids': [(p, self.bids[p]) for p in bid_prices],
        'asks': [(p, self.asks[p]) for p in ask_prices]
    }
    return depth
避坑指南:我曾经在获取最佳买卖价时直接用了max()和min(),在订单量大的时候性能很差。后来我改用heapq维护一个价格堆,每次插入和删除都维护堆结构,获取最佳价格变成了O(1)操作。如果你的仿真需要频繁查询最佳价格,建议用堆优化。

5. 知识体系总览

下面这张图展示了订单簿数据结构的核心逻辑。你想想看,从最底层的订单队列,到价格-数量映射,再到完整的订单簿,其实就是一个层层封装的过程。

订单簿数据结构设计 订单 (Order) order_id, price, quantity, timestamp 订单队列 (Order Queue) List[Order] 或 deque[Order] 价格-数量映射 (Price-Quantity Map) Dict[price, total_quantity] 订单簿 (Order Book) bids: Dict[price, quantity] + bid_queues: Dict[price, List] asks: Dict[price, quantity] + ask_queues: Dict[price, List] 先进先出 O(1)查询 价格优先 关键操作 • add_order() • remove_order() • get_best_bid() • get_best_ask() • get_market_depth() • match_order() • cancel_order() • update_order()

6. 实战中的几个坑

最后分享几个我在项目中踩过的坑,希望能帮你少走弯路:

  • 价格精度问题:浮点数作为字典键时,0.1+0.2不等于0.3。我建议统一用整数表示,比如价格乘以10000。
  • 队列性能问题:用list.pop(0)出队是O(n)操作,百万级订单时卡成狗。改用collections.deque的popleft()。
  • 并发安全问题:多线程仿真时,字典操作不是线程安全的。我一般用threading.Lock包裹关键操作。
  • 内存泄漏问题:撤单后如果不清除空队列,字典会越来越大。记得在remove_order里删除空队列。
我的建议:刚开始做仿真时,先用简单的字典+列表实现。等跑通了核心逻辑,再考虑性能优化。过早优化是万恶之源,这句话在量化交易里同样适用。

好了,订单簿的数据结构就讲到这里。这套设计我在多个仿真项目中都用过,稳定性和性能都还不错。你可以在自己的代码里直接复用,也可以根据需求做调整。记住,数据结构是骨架,算法是血肉,两者配合好了,订单簿才能跑得又快又稳。


专注资料整理