10. 回测框架搭建：使用Python构建回测引擎
回测，说白了就是让策略在历史数据上跑一遍。我见过太多人，策略回测时赚得盆满钵满，一上实盘就亏得底掉。为什么？多半是回测框架没搭好，或者绩效评估指标选错了。
这一章，我就带你手搭一个事件驱动的回测引擎。咱们用Python，从零开始。别怕，代码量不大，但逻辑得捋清楚。
10.1 为什么不用现成的框架？
市面上有backtrader、zipline这些成熟框架。我个人习惯是，小策略用backtrader，快速验证想法。但做基差套利这种精细活，我建议自己搭一个轻量引擎。
为什么？
- 基差套利逻辑特殊：需要同时处理两个合约的订单，还要盯价差。通用框架往往要绕很多弯子。
- 数据频率灵活：你可能需要tick级数据，也可能只需要分钟级。自己搭，想怎么切就怎么切。
- 调试方便：出bug了，你能直接定位到自己的代码。用别人的框架，你还得去翻源码。
我的经验：有一次用backtrader做跨期套利，为了处理两个合约的同步下单，我硬是写了三天适配代码。后来自己写了个50行的事件循环，半小时搞定。嗯，有时候「造轮子」反而更快。
10.2 事件驱动回测的核心逻辑
事件驱动，说白了就是「发生了什么事，就执行什么代码」。回测引擎里，事件主要有三种：
- 数据事件：新的行情数据来了
- 订单事件：订单被成交、被拒绝
- 定时事件：比如每天收盘前5分钟检查仓位
你想想看，真实交易中，你也是等行情来了才做决策，对吧？事件驱动就是模拟这个过程。
下面这张图，是我画的事件驱动回测引擎的核心流程：
10.3 手写一个轻量回测引擎
咱们直接上代码。这个引擎只做一件事：按时间顺序处理数据，调用你的策略，然后记录成交。
import pandas as pd
import numpy as np
from collections import deque
class EventDrivenBacktest:
"""事件驱动回测引擎 - 轻量版"""
def __init__(self, data, initial_capital=1000000):
self.data = data # DataFrame，必须包含 datetime 列
self.initial_capital = initial_capital
self.capital = initial_capital
self.positions = {} # 持仓字典
self.trades = [] # 成交记录
self.events = deque() # 事件队列
def run(self, strategy):
"""运行回测"""
# 按时间逐行推进
for idx, row in self.data.iterrows():
# 1. 生成数据事件
self.events.append(('DATA', row))
# 2. 处理所有事件
while self.events:
event_type, event_data = self.events.popleft()
if event_type == 'DATA':
# 调用策略逻辑
signals = strategy.on_data(event_data, self.positions)
# 处理信号，生成订单
for signal in signals:
self._process_order(signal, event_data)
elif event_type == 'ORDER':
# 处理订单成交
self._fill_order(event_data)
# 计算绩效
return self._calculate_performance()
def _process_order(self, signal, market_data):
"""处理交易信号，生成订单"""
# signal: {'direction': 'long'/'short', 'contract': 'IF2401', 'qty': 1}
# 这里可以加入滑点、手续费等
self.events.append(('ORDER', signal))
def _fill_order(self, order):
"""模拟订单成交"""
# 记录成交
self.trades.append(order)
# 更新持仓和资金
# ... 具体逻辑根据策略而定
def _calculate_performance(self):
"""计算绩效指标"""
# 返回收益率序列
pass
注意：上面的代码是骨架，实际使用时需要填充成交逻辑。我曾经犯过一个错——忘了处理「部分成交」，结果回测曲线漂亮得不像话，实盘却一塌糊涂。一定要模拟真实的成交环境。
10.4 绩效评估指标：夏普比率与卡玛比率
回测跑完了，怎么判断策略好不好？光看收益率不行。我见过年化50%的策略，最大回撤80%，你敢用吗？
这里有两个核心指标：
| 指标 | 公式 | 含义 | 我的经验阈值 |
|---|---|---|---|
| 夏普比率 | (策略收益率 - 无风险利率) / 波动率 | 每承担一单位风险，获得多少超额收益 | > 1.5 算不错，> 2.0 很优秀 |
| 卡玛比率 | 年化收益率 / 最大回撤 | 每承受一单位最大回撤，获得多少收益 | > 2.0 算稳健，> 3.0 很出色 |
你想想看，夏普比率看的是「波动」，卡玛比率看的是「最坏情况」。两者结合，才能全面评价策略。
10.5 计算代码实现
直接上代码，我习惯用向量化计算，速度快：
def calculate_sharpe_ratio(returns, risk_free_rate=0.03, periods_per_year=252):
"""
计算夏普比率
returns: 日收益率序列
"""
excess_returns = returns - risk_free_rate / periods_per_year
sharpe = np.sqrt(periods_per_year) * excess_returns.mean() / returns.std()
return sharpe
def calculate_calmar_ratio(returns, periods_per_year=252):
"""
计算卡玛比率
"""
# 年化收益率
total_return = (1 + returns).prod()
years = len(returns) / periods_per_year
annual_return = total_return ** (1 / years) - 1
# 最大回撤
cumulative = (1 + returns).cumprod()
running_max = cumulative.cummax()
drawdown = (cumulative - running_max) / running_max
max_drawdown = drawdown.min()
calmar = annual_return / abs(max_drawdown)
return calmar
# 使用示例
daily_returns = pd.Series([0.001, -0.002, 0.003, ...])
sharpe = calculate_sharpe_ratio(daily_returns)
calmar = calculate_calmar_ratio(daily_returns)
print(f"夏普比率: {sharpe:.2f}")
print(f"卡玛比率: {calmar:.2f}")
避坑指南：我曾经用日收益率算夏普，结果策略看起来很好。后来发现，日收益率波动率被低估了——因为策略只在特定时间交易，其他时间都是0收益。正确的做法是，只计算「有交易的日子」的收益率。嗯，这个坑我踩过。
10.6 回测中的常见陷阱
做回测这么多年，我总结了几条铁律：
- 未来函数：千万别用未来的数据做决策。比如用收盘价计算信号，却在开盘价成交。我见过有人这么干，回测年化50%，实盘亏30%。
- 幸存者偏差：回测时只用了现在还在交易的合约。那些退市的、被摘牌的，你都没算进去。结果就是回测偏乐观。
- 手续费和滑点：基差套利看似利润薄，手续费一扣可能就没了。我建议至少按双边万分之一算，滑点按1个tick算。
我的习惯：每次回测完，我都会做一次「压力测试」——把手续费翻倍，滑点翻倍，看看策略还能不能活。如果还能盈利，我才敢上实盘。
好了，回测框架的核心就这些。记住，回测不是为了证明策略有多牛，而是为了发现策略的弱点。你想想看，一个在回测中漏洞百出的策略，实盘能赚钱吗？