第十二章：回测框架搭建

回测框架这东西，说白了就是你的时光机。我刚开始做市商那会儿，总觉得自己策略无敌，结果实盘一跑就被市场按在地上摩擦。后来才明白——不回测就上实盘，等于闭眼开车。

今天咱们聊聊怎么搭一套靠谱的回测框架。我会从历史数据回测讲起，再到事件驱动回测，最后说说模拟撮合引擎的设计。嗯，这三块是递进关系，缺一不可。

12.1 历史数据回测：最基础的验证手段

历史数据回测，就是拿过去的数据跑一遍你的策略。听起来简单，但坑特别多。

我个人习惯把历史数据回测分成三步：

  1. 数据清洗——去掉异常值、填充缺失值、对齐时间戳
  2. 策略执行——按时间顺序逐笔或逐Tick跑信号
  3. 绩效统计——算夏普、最大回撤、胜率这些指标

这里有个关键点：前视偏差。我在项目中遇到过好几次，回测结果漂亮得不像话，结果发现代码里不小心用了未来的数据。比如用当天的收盘价去判断当天的开仓信号——这肯定不行。

⚠️ 避坑指南
我曾经犯过一个低级错误：在回测时用了整个数据集计算均线，而不是滚动计算。结果回测曲线完美向上，实盘直接崩了。记住：回测必须严格模拟当时能获取到的信息。

来看一个简单的历史回测代码框架：

class BacktestEngine:
    def __init__(self, data, initial_capital=100000):
        self.data = data
        self.capital = initial_capital
        self.positions = []
        self.trades = []
    
    def run(self):
        for i in range(len(self.data)):
            # 获取当前时刻的数据（注意：只能用到i时刻之前的数据）
            current_bar = self.data.iloc[:i+1]
            
            # 生成信号
            signal = self.generate_signal(current_bar)
            
            # 执行交易
            if signal != 0:
                self.execute_trade(signal, self.data.iloc[i])
            
            # 更新持仓市值
            self.update_pnl(self.data.iloc[i])
    
    def generate_signal(self, data):
        # 这里写你的策略逻辑
        pass
    
    def execute_trade(self, signal, price):
        # 执行交易逻辑
        pass
    
    def update_pnl(self, current_price):
        # 更新盈亏
        pass

12.2 事件驱动回测：更贴近真实市场

历史数据回测有个硬伤——它假设你能在任意时刻以任意价格成交。真实市场不是这样的。你想想看，你的订单发出去，得经过交易所撮合，可能成交也可能不成交，成交价格也可能滑点。

事件驱动回测就是为了解决这个问题。它模拟了市场事件的流动：

  • Tick事件——每一笔成交或报价变化
  • 订单事件——你发出去的限价单、市价单的状态变化
  • 定时事件——比如每秒检查一次持仓

我建议用事件队列来管理这些事件。说白了，就是一个先进先出的队列，按时间顺序处理每个事件。

💡 核心思路
事件驱动回测的核心是：把市场变化和策略决策都抽象成事件。每个事件触发相应的处理函数，这样代码结构清晰，也容易扩展。

举个例子，一个简单的事件驱动回测框架：

class EventDrivenBacktest:
    def __init__(self):
        self.event_queue = []
        self.current_time = None
    
    def add_event(self, event):
        # 按时间排序插入事件
        self.event_queue.append(event)
        self.event_queue.sort(key=lambda e: e.timestamp)
    
    def run(self):
        while self.event_queue:
            event = self.event_queue.pop(0)
            self.current_time = event.timestamp
            
            if event.type == 'TICK':
                self.on_tick(event.data)
            elif event.type == 'ORDER':
                self.on_order(event.data)
            elif event.type == 'TIMER':
                self.on_timer()
    
    def on_tick(self, tick_data):
        # 处理Tick数据，生成信号
        signal = self.strategy.on_tick(tick_data)
        if signal:
            self.send_order(signal)
    
    def on_order(self, order_data):
        # 处理订单状态变化
        if order_data.status == 'FILLED':
            self.update_position(order_data)
    
    def on_timer(self):
        # 定时任务，比如风控检查
        self.risk_check()

12.3 模拟撮合引擎设计：回测的灵魂

模拟撮合引擎，这是回测框架里最难也最重要的部分。它决定了你的订单能不能成交、以什么价格成交。

我见过很多团队，回测时直接用收盘价成交，结果实盘滑点吃掉所有利润。所以，撮合引擎的精度直接决定了回测的可信度。

一个合格的模拟撮合引擎至少要考虑：

因素 说明 实现方式
订单簿深度 当前买一卖一的价格和数量 用Level2数据或模拟深度
滑点模型 大单成交时的价格偏移 线性滑点或基于深度的滑点
成交概率 限价单能否成交 根据价格位置和流动性估算
延迟模拟 订单从发出到成交的时间差 固定延迟或随机延迟

我个人习惯用订单簿快照来模拟撮合。每次Tick到来时，更新订单簿，然后检查你的挂单是否被吃掉。这样虽然计算量大一点，但结果更真实。

🔧 实用技巧
如果你没有Level2数据，可以用Tick数据模拟一个简化的订单簿。比如：假设买一卖一价差固定，深度按成交量比例分配。虽然粗糙，但比直接用收盘价强多了。

来看一个简化版的撮合引擎：

class MatchingEngine:
    def __init__(self):
        self.bids = []  # 买单队列
        self.asks = []  # 卖单队列
        self.order_book = {'bid': {}, 'ask': {}}
    
    def update_order_book(self, tick):
        # 更新订单簿
        self.order_book['bid'] = tick['bid_prices']
        self.order_book['ask'] = tick['ask_prices']
    
    def match_order(self, order):
        if order.side == 'BUY':
            # 市价买单：按卖一价成交
            if order.type == 'MARKET':
                best_ask = min(self.order_book['ask'].keys())
                fill_price = best_ask
                fill_qty = min(order.qty, self.order_book['ask'][best_ask])
                return fill_price, fill_qty
            # 限价买单：价格高于卖一才能成交
            else:
                if order.price >= min(self.order_book['ask'].keys()):
                    fill_price = order.price
                    fill_qty = order.qty
                    return fill_price, fill_qty
                else:
                    return None, 0  # 未成交
        else:
            # 卖单逻辑类似
            pass
    
    def simulate_slippage(self, order, fill_price):
        # 模拟滑点：大单加价
        if order.qty > 100:
            slippage = order.qty * 0.0001  # 每手加0.01%
            return fill_price * (1 + slippage)
        return fill_price

12.4 回测框架的整体架构

把上面三块拼起来，就是一个完整的回测框架。我画了张图，帮你理清关系：

回测框架整体架构 数据层 历史Tick数据 | Level2订单簿 | 清洗对齐 | 时间戳管理 事件引擎层 Tick事件 | 订单事件 | 定时事件 | 事件队列管理 撮合引擎层 订单簿维护 | 成交判定 | 滑点模拟 | 延迟模拟 策略层 输入 调度 执行 输出

从这张图你能看到，数据从底层往上流，策略信号从上往下执行。每一层各司其职，互不干扰。这也是我推荐的分层设计思路——每一层都可以单独替换或升级。

🎯 总结一下
回测框架搭建，核心就三件事：
1. 历史数据回测——验证策略在历史数据上的表现
2. 事件驱动回测——模拟真实市场的事件流
3. 模拟撮合引擎——决定订单能否成交、以什么价格成交

这三块做好了，你的回测结果才有参考价值。否则，回测再漂亮也是自欺欺人。

嗯，今天就聊到这儿。回测框架这东西，光看理论没用，你得动手搭一遍。我建议你先从最简单的历史数据回测开始，慢慢加上事件驱动和撮合引擎。每一步踩过的坑，都是你成长的阶梯。

交易系统化学习资料 微信Strategy888888

← 上一章 📖 返回目录 下一章 →