11、订单簿模拟器：构建模拟撮合引擎、回测环境、模拟订单生成

做量化交易这些年，我越来越觉得：纸上谈兵是最大的坑。策略写得再漂亮，不上模拟器跑一跑，你永远不知道它会在真实市场里摔得多惨。

这一章，我们就来亲手搭建一个订单簿模拟器。说白了，就是造一个虚拟的交易环境。你可以往里面扔订单，看它怎么撮合，怎么影响价格。我习惯把这个东西叫做「交易沙盘」——你在沙盘里怎么折腾都行，反正不花真金白银。

11.1 模拟撮合引擎：订单簿的心脏

撮合引擎，就是订单簿里最核心的那块逻辑。它的任务很简单：把买单和卖单对上。但实现起来，细节多得让人头疼。

我个人习惯把撮合引擎拆成三个部分：

• 订单簿维护：管理买盘（Bids）和卖盘（Asks）两个价格队列
• 价格优先级：买价高的优先，卖价低的优先
• 时间优先级：同价格下，先来的先成交

先看一个最简化的撮合逻辑：

class MatchingEngine:
    def __init__(self):
        self.bids = []  # 买单队列，按价格降序
        self.asks = []  # 卖单队列，按价格升序

    def add_order(self, order):
        """添加订单并尝试撮合"""
        if order.side == 'buy':
            self._match_buy(order)
        else:
            self._match_sell(order)

    def _match_buy(self, buy_order):
        """买单进来，跟卖单队列撮合"""
        while self.asks and buy_order.quantity > 0:
            best_ask = self.asks[0]
            if buy_order.price < best_ask.price:
                break  # 价格对不上，不成交
            traded_qty = min(buy_order.quantity, best_ask.quantity)
            # 成交逻辑
            buy_order.quantity -= traded_qty
            best_ask.quantity -= traded_qty
            if best_ask.quantity == 0:
                self.asks.pop(0)
        # 如果还有剩余，挂到买单队列
        if buy_order.quantity > 0:
            self.bids.append(buy_order)
            self.bids.sort(key=lambda x: -x.price)  # 按价格降序

嗯，这里要注意：真实场景下，排序操作不能这么粗暴。我在项目中遇到过，每秒几千笔订单进来，每次都全量排序，CPU直接拉满。后来改用heapq或者sortedcontainers，性能才稳住。

11.2 回测环境：让历史数据说话

回测环境，说白了就是让模拟器「倒带」。你把历史行情数据喂进去，它像放电影一样重演一遍。你的策略在哪个时间点该买、该卖，都能看得清清楚楚。

我建议回测环境至少包含这几个组件：

1. 数据加载器：读取历史Tick数据或K线数据
2. 时间推进器：按时间顺序逐笔推送数据
3. 策略接口：让策略能订阅行情、提交订单
4. 绩效统计器：记录每笔交易，计算盈亏

来看一个回测循环的骨架：

class BacktestEngine:
    def __init__(self, data, strategy):
        self.data = data          # 历史数据列表
        self.strategy = strategy  # 策略实例
        self.engine = MatchingEngine()
        self.trades = []

    def run(self):
        for tick in self.data:
            # 1. 更新订单簿
            self.engine.update_orderbook(tick)
            # 2. 通知策略当前行情
            signals = self.strategy.on_tick(tick)
            # 3. 执行策略信号
            for signal in signals:
                order = self.create_order(signal)
                self.engine.add_order(order)
            # 4. 检查成交
            self._process_trades()
        # 5. 输出绩效
        self.report()

你想想看，回测最大的坑是什么？未来函数。我曾经犯过一个低级错误：回测时不小心把当天的收盘价用在了开盘决策里。结果策略在回测里赚翻了，实盘亏到怀疑人生。所以，数据加载器一定要保证时间戳严格递增，不能有任何「偷看」未来的行为。

11.3 模拟订单生成：制造「假想敌」

光有回测还不够。有时候你想测试订单簿在极端行情下的表现，但历史数据里没有这种场景。这时候就需要模拟订单生成器——它能按你设定的规则，源源不断地造出订单来。

我个人常用的生成策略有三种：

来看一个泊松过程的订单生成器：

import random
import time

class OrderGenerator:
    def __init__(self, lambda_rate=10):
        self.lambda_rate = lambda_rate  # 每秒平均订单数
        self.order_id = 0

    def generate_orders(self, current_price, duration_sec=1):
        """生成一段时间内的模拟订单"""
        orders = []
        # 泊松过程：间隔时间服从指数分布
        t = 0
        while t < duration_sec:
            interval = random.expovariate(self.lambda_rate)
            t += interval
            if t > duration_sec:
                break
            # 随机生成订单参数
            side = random.choice(['buy', 'sell'])
            price = current_price * (1 + random.uniform(-0.01, 0.01))
            quantity = random.randint(1, 100)
            order = {
                'id': self.order_id,
                'side': side,
                'price': round(price, 2),
                'quantity': quantity,
                'timestamp': t
            }
            orders.append(order)
            self.order_id += 1
        return orders

为什么用指数分布？因为真实市场的订单到达时间，确实符合这个规律。我在做高频策略回测时，用这个生成器模拟出来的订单流，跟实盘数据的统计特征非常接近。

11.4 把三者串起来：一个完整的模拟器

好了，现在我们有三个模块了：撮合引擎、回测环境、订单生成器。怎么把它们串成一个完整的模拟器？

我画了一张架构图，你看一眼就明白了：

这张图里，三个模块形成了一个闭环。订单生成器造出订单，撮合引擎处理成交，回测环境记录结果，然后根据结果调整生成参数。这样就能模拟出各种市场环境。

我在实际项目中，用这套架构做过一个很有意思的实验：让两个策略互相交易。一个做市商策略负责提供流动性，一个趋势策略负责吃单。结果发现，当两个策略的参数不匹配时，订单簿会出现「假突破」——价格瞬间被打穿，然后又弹回来。这种场景在真实市场里很常见，但如果没有模拟器，你很难复现和分析。

最后说一句：模拟器永远无法100%还原真实市场。但它能帮你过滤掉80%的愚蠢错误。剩下的20%，就得靠实盘小仓位去试了。嗯，这就是我们做模拟器的意义——在犯错成本最低的地方，把能犯的错都犯一遍。