4、回测框架搭建：用Python实现一个简单的回测引擎

回测引擎，说白了就是你的策略时光机。

它能让你穿越到过去，看看你的套利策略在历史上表现如何。我个人习惯把回测引擎比作「交易实验室」——你在里面做实验，失败了也不会亏真金白银。

4.1 回测引擎的核心架构

先别急着写代码。我们得搞清楚，一个回测引擎到底长什么样？

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

嗯，这张图里最核心的就是那个「主循环」。它像钟表一样，一格一格地往前走。每走一格，就触发一次数据更新、策略判断、交易执行。

4.2 数据模块：喂给引擎的「原料」

回测的第一步，是拿到干净的数据。我在项目中遇到过最坑的事——数据里有未来函数，回测曲线漂亮得像假的一样，实盘直接崩了。

所以，数据清洗是底线。我们至少要做三件事：

• 去重：同一时间戳的重复数据，只保留一条
• 补全：缺失的时间点，用前值填充或插值
• 对齐：不同品种的时间戳必须对齐到同一频率

来看一个简单的数据加载函数：

import pandas as pd
import numpy as np

def load_data(file_path, freq='1min'):
    """
    加载并清洗行情数据
    freq: 重采样频率，默认1分钟
    """
    df = pd.read_csv(file_path, parse_dates=['time'])
    df.set_index('time', inplace=True)
    
    # 去重：按时间戳去重，保留第一条
    df = df[~df.index.duplicated(keep='first')]
    
    # 补全缺失时间点
    full_idx = pd.date_range(start=df.index.min(), 
                             end=df.index.max(), freq=freq)
    df = df.reindex(full_idx, method='ffill')
    
    # 删除开盘前和收盘后的数据（假设9:30-15:00交易）
    df = df.between_time('09:30', '15:00')
    
    return df

这段代码看起来简单，但够用了。你想想看，真实场景下数据量动辄几百万行，性能优化是后话，先把逻辑跑通再说。

4.3 策略模块：把你的想法写成代码

策略模块是回测引擎的灵魂。我建议把策略设计成一个「黑盒」——给它数据，它返回信号。

一个标准的策略接口长这样：

class BaseStrategy:
    """策略基类，所有策略都继承它"""
    
    def __init__(self, params=None):
        self.params = params or {}
        self.position = 0  # 当前持仓
        self.signals = []  # 信号列表
    
    def on_data(self, data):
        """
        每来一条新数据，调用一次
        data: 当前时刻的行情数据（DataFrame的一行）
        返回: 1（买入）, -1（卖出）, 0（持有）
        """
        raise NotImplementedError("子类必须实现on_data方法")
    
    def reset(self):
        """回测开始前重置状态"""
        self.position = 0
        self.signals = []

举个例子，一个简单的均值回归策略：

class MeanReversionStrategy(BaseStrategy):
    """均值回归策略：价格偏离均线太多就反向开仓"""
    
    def __init__(self, params=None):
        super().__init__(params)
        self.window = params.get('window', 20)  # 均线周期
        self.threshold = params.get('threshold', 2.0)  # 标准差倍数
        self.prices = []  # 缓存历史价格
    
    def on_data(self, data):
        price = data['close']
        self.prices.append(price)
        
        if len(self.prices) < self.window:
            return 0  # 数据不够，不交易
        
        # 计算均线和标准差
        ma = np.mean(self.prices[-self.window:])
        std = np.std(self.prices[-self.window:])
        
        if std == 0:
            return 0
        
        z_score = (price - ma) / std
        
        # 信号判断
        if z_score > self.threshold and self.position <= 0:
            return -1  # 卖出
        elif z_score < -self.threshold and self.position >= 0:
            return 1   # 买入
        else:
            return 0   # 持有

4.4 执行模块：模拟真实交易

策略发出了信号，接下来就是「执行」。执行模块要考虑几个现实问题：

• 滑点：你看到的价格和实际成交价之间有差距
• 手续费：每笔交易都要交钱
• 成交限制：大单可能无法全部成交

我习惯用一个简单的「订单簿」来模拟：

class ExecutionEngine:
    def __init__(self, slippage=0.0001, commission=0.0003):
        self.slippage = slippage  # 滑点比例
        self.commission = commission  # 手续费比例
        self.trades = []  # 成交记录
    
    def execute(self, signal, price, volume):
        """
        执行交易
        signal: 1买入, -1卖出
        price: 当前价格
        volume: 交易数量
        返回: 实际成交价格和数量
        """
        # 模拟滑点：买入价格上浮，卖出价格下浮
        if signal == 1:
            exec_price = price * (1 + self.slippage)
        else:
            exec_price = price * (1 - self.slippage)
        
        # 计算手续费
        fee = exec_price * volume * self.commission
        
        # 记录交易
        self.trades.append({
            'price': exec_price,
            'volume': volume,
            'fee': fee,
            'signal': signal
        })
        
        return exec_price, volume, fee

4.5 主循环：让一切转起来

好了，各个模块都准备好了。现在把它们组装起来，跑一个完整的回测：

def run_backtest(data, strategy, execution, initial_capital=100000):
    """
    运行回测
    data: 历史行情数据（DataFrame）
    strategy: 策略实例
    execution: 执行引擎实例
    initial_capital: 初始资金
    """
    capital = initial_capital
    position = 0
    equity_curve = []  # 权益曲线
    
    # 重置策略状态
    strategy.reset()
    
    for idx, row in data.iterrows():
        # 1. 获取信号
        signal = strategy.on_data(row)
        
        # 2. 执行交易
        if signal != 0:
            # 简单仓位管理：每次交易固定数量
            volume = 100  # 假设每次交易100股
            price = row['close']
            
            exec_price, exec_volume, fee = execution.execute(
                signal, price, volume
            )
            
            # 更新持仓和资金
            if signal == 1:  # 买入
                cost = exec_price * exec_volume + fee
                if capital >= cost:
                    capital -= cost
                    position += exec_volume
            else:  # 卖出
                revenue = exec_price * exec_volume - fee
                capital += revenue
                position -= exec_volume
        
        # 3. 记录当前权益
        current_equity = capital + position * row['close']
        equity_curve.append(current_equity)
    
    # 计算绩效指标
    returns = pd.Series(equity_curve).pct_change().dropna()
    total_return = (equity_curve[-1] - initial_capital) / initial_capital
    sharpe = np.sqrt(252) * returns.mean() / returns.std() if returns.std() > 0 else 0
    
    return {
        'equity_curve': equity_curve,
        'total_return': total_return,
        'sharpe_ratio': sharpe,
        'trades': execution.trades
    }

这段代码虽然简单，但已经包含了回测引擎的核心逻辑。你想想看，实际生产环境里，我们还会加入多线程、内存管理、异常处理等等。但核心骨架，就是上面这些。

4.6 绩效评估：回测结果怎么看？

回测跑完了，一堆数字摆在你面前。怎么看？我一般关注这几个指标：

指标	含义	合格线
总收益率	回测期间的总收益比例	至少跑赢基准
夏普比率	每单位风险获得的超额收益	> 1.0 算不错
最大回撤	账户从峰值跌到谷底的最大幅度	< 20% 可接受
胜率	盈利交易占总交易的比例	> 40% 算正常
盈亏比	平均盈利 / 平均亏损	> 1.5 较好

4.7 避坑指南

最后，分享几个我踩过的坑：

• 幸存者偏差：只用现在还在交易的品种做回测，退市的都被忽略了。结果看起来很好，但实际可能很差。
• 过拟合：参数调得太多，策略只适合历史数据。换一个时间段就失效。我建议参数不超过3个。
• 未来信息泄露：比如用当天的收盘价计算当天的信号，这在实盘中根本做不到。一定要确保信号只基于历史数据。
• 忽略交易成本：高频套利策略，手续费和滑点可能是利润的杀手。我曾经有一个策略，算上手续费后直接从盈利变成亏损。

嗯，回测引擎就讲到这里。代码虽然简单，但足够你开始做实验了。记住，回测只是起点，不是终点。真正的考验，永远在实盘里。

---

公众号:蓝海资料掘金营，微信deep3321