第七章：回测框架搭建

做量化交易，回测框架就是你的实验室。没有它，你的策略就是空中楼阁。我个人习惯把回测框架分成两大流派：向量化回测和事件驱动回测。今天咱们就把这两套东西掰开揉碎了讲清楚。

7.1 向量化回测 vs 事件驱动回测

先说说向量化回测。说白了，就是用pandas的向量化操作，一次性算完所有信号和持仓。速度快，代码短，适合快速验证想法。

import pandas as pd
import numpy as np

def vectorized_backtest(data, signal_col='signal'):
    """
    向量化回测核心逻辑
    data: DataFrame，包含价格和信号
    """
    # 计算每日收益率
    data['returns'] = data['close'].pct_change()
    
    # 信号平移一天，避免未来信息
    data['position'] = data[signal_col].shift(1)
    
    # 策略收益率 = 持仓 * 每日收益率
    data['strategy_returns'] = data['position'] * data['returns']
    
    # 累计净值
    data['equity'] = (1 + data['strategy_returns']).cumprod()
    
    return data

但向量化有个大坑——它假设你可以在收盘价瞬间成交。我在实盘中吃过这个亏，回测曲线漂亮得很，一上实盘就崩。为什么？因为滑点和流动性问题被忽略了。

事件驱动回测就不一样了。它模拟真实交易流程：收到行情→判断信号→发单→成交确认。每一步都有时间戳，能精确模拟订单簿变化。

class EventDrivenBacktest:
    """
    事件驱动回测引擎（简化版）
    """
    def __init__(self, initial_capital=1000000):
        self.capital = initial_capital
        self.positions = {}  # 持仓字典
        self.orders = []     # 订单队列
        self.trades = []     # 成交记录
        
    def on_bar(self, bar_data):
        """处理每个K线事件"""
        # 1. 检查是否有待成交订单
        self.process_orders(bar_data)
        
        # 2. 生成新信号
        signal = self.generate_signal(bar_data)
        
        # 3. 如果信号变化，生成新订单
        if signal != self.current_signal:
            self.place_order(signal, bar_data)
    
    def process_orders(self, bar_data):
        """模拟订单成交（考虑滑点）"""
        for order in self.orders:
            # 模拟成交价 = 开盘价 + 滑点
            fill_price = bar_data['open'] * (1 + self.slippage)
            # 更新持仓和资金
            self.fill_order(order, fill_price)

7.2 手续费与滑点模拟

手续费和滑点是回测里最容易造假的地方。我见过太多人把手续费设成0，回测年化50%，实盘直接亏到姥姥家。

咱们得把成本算清楚：

成本类型	期货市场	股票市场
手续费	按手数固定（如3元/手）或按成交额比例	佣金万2.5 + 印花税千1（卖出）
滑点	1-2个tick（看流动性）	0.01-0.05元/股
冲击成本	大单时额外加0.5-1个tick	按成交额比例估算

def calculate_trade_cost(price, volume, market='futures'):
    """
    计算交易成本
    """
    if market == 'futures':
        # 期货：按手数收费
        commission_per_lot = 3  # 每手3元
        lots = volume / 10  # 假设每手10吨
        commission = lots * commission_per_lot
        
        # 滑点：1个tick
        tick_size = 1  # 假设1元/tick
        slippage = tick_size * volume
        
    elif market == 'stock':
        # 股票：按成交额比例
        turnover = price * volume
        commission = turnover * 0.00025  # 万2.5
        stamp_tax = turnover * 0.001 if 'sell' else 0  # 卖出印花税
        
        # 滑点：0.02元/股
        slippage = 0.02 * volume
    
    return commission + slippage

7.3 资金管理模块

资金管理是量化交易的命门。没有好的资金管理，再牛的策略也会爆仓。我习惯用凯利公式做仓位参考，但实际执行时会打折扣。

class RiskManager:
    """
    资金管理模块
    """
    def __init__(self, initial_capital, max_risk_per_trade=0.02):
        self.capital = initial_capital
        self.max_risk = max_risk_per_trade  # 单笔最大风险2%
        
    def kelly_position(self, win_rate, avg_win, avg_loss):
        """
        凯利公式计算最优仓位
        """
        # 凯利公式：f* = (p*b - q) / b
        # p = 胜率, q = 1-p, b = 盈亏比
        b = avg_win / avg_loss
        p = win_rate
        q = 1 - p
        
        kelly_f = (p * b - q) / b
        
        # 实际使用半凯利，降低风险
        return min(kelly_f * 0.5, 0.25)  # 不超过25%仓位
    
    def position_sizing(self, entry_price, stop_loss, capital_pct=None):
        """
        基于止损计算仓位
        """
        if capital_pct is None:
            capital_pct = self.max_risk
            
        risk_per_share = entry_price - stop_loss
        total_risk = self.capital * capital_pct
        
        shares = total_risk / risk_per_share
        
        return int(shares)

7.4 绩效指标计算

回测做完，得算算策略到底行不行。我一般看这几个指标：

def calculate_performance_metrics(equity_curve, risk_free_rate=0.03):
    """
    计算核心绩效指标
    """
    # 日收益率序列
    daily_returns = equity_curve.pct_change().dropna()
    
    # 年化收益率
    total_return = equity_curve.iloc[-1] / equity_curve.iloc[0] - 1
    n_years = len(daily_returns) / 252
    annual_return = (1 + total_return) ** (1/n_years) - 1
    
    # 年化波动率
    annual_vol = daily_returns.std() * np.sqrt(252)
    
    # 夏普比率
    sharpe = (annual_return - risk_free_rate) / annual_vol
    
    # 最大回撤
    rolling_max = equity_curve.expanding().max()
    drawdown = (equity_curve - rolling_max) / rolling_max
    max_drawdown = drawdown.min()
    
    # 卡玛比率
    calmar = annual_return / abs(max_drawdown)
    
    # 胜率
    win_rate = len(daily_returns[daily_returns > 0]) / len(daily_returns)
    
    return {
        'annual_return': annual_return,
        'annual_vol': annual_vol,
        'sharpe_ratio': sharpe,
        'max_drawdown': max_drawdown,
        'calmar_ratio': calmar,
        'win_rate': win_rate
    }

这里有个细节：夏普比率的分母是波动率。波动率越低，夏普越高。但低波动不代表低风险，你想想看，一个策略天天不赚钱，波动率当然低，夏普可能还挺好看。所以不能只看夏普。

7.5 回测框架整体架构

最后，我把整个回测框架的结构画出来。你照着这个搭，基本不会漏东西。

嗯，这个架构图基本覆盖了回测需要的所有模块。你从数据层开始，一层层往下搭，每层都做单元测试。我当年就是偷懒没测数据层，结果用了一周才发现行情数据有缺失，白白浪费了时间。

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