第十五章：回测框架搭建

做市商策略写好了，模型也训练完了。然后呢？

直接上实盘？别闹。我见过太多人，策略回测都没跑明白，就敢拿真金白银去试。结果呢？一周亏掉三个月的工资。

回测，说白了就是给策略做「模拟考试」。但这里面的坑，比你想象的多得多。

15.1 历史数据回测：最基础，也最容易犯错

历史数据回测，就是把过去的数据喂给策略，看它表现如何。听起来简单吧？

嗯，我刚开始做的时候也这么想。直到有一次，我发现策略在2015年股灾期间收益特别好——后来才意识到，我用的数据里包含了「未来信息」。

正确的做法是什么？

• 严格按时间顺序处理数据：每条数据只能用到它之前的信息
• 使用「滚动窗口」：训练集和测试集要严格分离
• 注意数据对齐：比如用tick数据时，要确保买卖价的时间戳对齐

# 一个简单但正确的时间序列回测框架
def backtest_simple(strategy, data):
    """
    严格按时间顺序回测
    我习惯用这种方式，虽然慢，但不会出前瞻偏差
    """
    positions = []
    for i in range(len(data)):
        # 只用当前时刻之前的数据
        current_data = data.iloc[:i+1]
        signal = strategy.generate_signal(current_data)
        positions.append(signal)
    return positions

15.2 事件驱动回测：更接近真实市场

历史数据回测有个问题：它假设市场是「静止」的。但真实市场是动态的，订单簿在变，流动性在变，对手方也在变。

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

• 订单到达事件：新的买单或卖单进入市场
• 成交事件：订单被匹配
• 撤单事件：订单被取消
• 行情更新事件：价格、成交量变化

我个人习惯用「事件队列」来管理这些。每个事件都带时间戳，按时间顺序处理。

class EventDrivenBacktest:
    def __init__(self):
        self.event_queue = []  # 事件队列
        self.current_time = None
        
    def run(self, market_data):
        for event in market_data:
            self.current_time = event.timestamp
            self.process_event(event)
            
    def process_event(self, event):
        # 根据事件类型处理
        if event.type == 'ORDER':
            self.handle_order(event)
        elif event.type == 'TRADE':
            self.handle_trade(event)

15.3 模拟撮合引擎：核心中的核心

做市商策略的核心是「撮合」。你的订单能不能成交？以什么价格成交？这些都需要模拟撮合引擎来处理。

一个完整的模拟撮合引擎应该包含：

1. 订单簿管理：维护买卖双方的订单队列
2. 价格优先、时间优先：这是交易所的基本规则
3. 部分成交处理：大单可能被拆成多笔成交
4. 滑点模拟：实际成交价可能和预期不同

class MatchingEngine:
    def __init__(self):
        self.buy_orders = []  # 买单队列
        self.sell_orders = []  # 卖单队列
        
    def match(self, order):
        """
        模拟撮合逻辑
        嗯，这里要注意：真实交易所的撮合比这复杂得多
        """
        if order.side == 'BUY':
            # 检查是否有卖单可以匹配
            while self.sell_orders and order.quantity > 0:
                best_sell = self.sell_orders[0]
                if best_sell.price <= order.price:
                    # 成交
                    trade_price = best_sell.price
                    trade_qty = min(order.quantity, best_sell.quantity)
                    # 更新订单状态
                    order.quantity -= trade_qty
                    best_sell.quantity -= trade_qty
                    if best_sell.quantity == 0:
                        self.sell_orders.pop(0)
                else:
                    break
        # 类似处理卖单...

15.4 避免过拟合：回测最大的敌人

你想想看，一个策略在历史数据上表现特别好，是不是就代表未来也好？

不一定。过拟合就是这个问题——策略「记住」了历史数据的噪声，而不是真正的规律。

我踩过最大的坑就是这个。有一次，我优化了一个参数组合，回测年化收益50%，最大回撤只有5%。我当时兴奋坏了。结果实盘一周，亏了15%。

怎么避免？

• 交叉验证：把数据分成多段，轮流做训练和测试
• 参数稳定性测试：稍微改变参数，看结果是否稳定
• 样本外测试：留出一段数据完全不参与优化
• 蒙特卡洛模拟：随机打乱数据，看策略是否还能赚钱

def cross_validate(strategy, data, k=5):
    """
    K折交叉验证
    我一般用5折，数据量大的时候用10折
    """
    fold_size = len(data) // k
    results = []
    
    for i in range(k):
        # 留出一段作为测试集
        test_start = i * fold_size
        test_end = (i + 1) * fold_size
        
        train_data = data[:test_start] + data[test_end:]
        test_data = data[test_start:test_end]
        
        # 训练和测试
        strategy.fit(train_data)
        result = strategy.backtest(test_data)
        results.append(result)
    
    return results

15.5 前瞻偏差：回测的隐形杀手

前瞻偏差比过拟合更隐蔽。它发生在你「无意中」用到了未来信息。

举个例子：你在计算某个指标时，用了当天的收盘价。但你的策略是在盘中交易的，收盘价还没出来呢。这就是前瞻偏差。

常见的几种前瞻偏差：

类型	描述	如何避免
数据泄露	用到了未来数据	严格按时间顺序处理
幸存者偏差	只用了存活下来的股票	包含退市股票的数据
回填偏差	用到了后来修正的数据	使用原始数据，不要用修正版
选择偏差	只选了表现好的时间段	使用完整的时间序列

15.6 回测框架的整体架构

说了这么多，我们来画一张图，看看整个回测框架长什么样。

这张图展示了回测框架的核心流程：数据从底层流入，经过事件引擎处理，再到撮合引擎模拟交易，最后交给策略执行和评估。

每个环节都有它的坑。数据层要注意前瞻偏差，事件引擎要注意时间顺序，撮合引擎要注意精度，策略层要注意过拟合。

回测不是目的，目的是找到真正能在市场上赚钱的策略。所以，别在回测上花太多时间「美化」结果。真实一点，残酷一点，对你有好处。

---

无相订单流研究社 微信Lucian808555