回测方法论：如何避免自欺欺人

95% 的回测都是垃圾。它们使用了未来数据、忽略交易成本、在有幸存者偏差的数据上测试，并把参数拟合到资金曲线看起来完美为止。然后交易者拿真钱去冒险，再纳闷策略为什么失效。下面讲如何正确回测——以及为什么多数人做不到。

回测的七宗罪

第一宗：前视偏差（使用未来数据）

它是什么： 使用了交易发生当时根本拿不到的信息。

示例 (wrong):

# WRONG: Using today's close to calculate indicator that triggers today's trade
rsi = calculate_rsi(df['Close'])
df['Signal'] = (rsi < 30).astype(int)  # Buy when RSI < 30

# Problem: Today's RSI includes today's close, but you'd need to decide
# to buy BEFORE the close. You're peeking into the future!

示例 (correct):

# CORRECT: Use yesterday's close to generate today's signal
rsi = calculate_rsi(df['Close']).shift(1)  # Shift forward by 1 day
df['Signal'] = (rsi < 30).astype(int)

# Now today's signal is based on data available yesterday

其他前视偏差的陷阱：

• 用当天数据计算波动率： ATR、标准差都必须使用前一日收盘价
• 使用财报数据： 财报是在收盘后发布的——除非用盘后数据，否则别在当天交易
• 在错误的日期再平衡： 指数再平衡在执行前数周就公布了——按执行日交易，而不是公告日
• 股息： 除息日与派息日的区别（价格在除息日下跌，而非派息日）

第二宗：幸存者偏差（只测试活下来的）

它是什么： 只测试今天仍然存在的股票或资产，忽略了那些破产或退市的。

影响：

为什么重要：

• 2000 年的股票中有 40% 如今已不再交易（破产、被收购或退市）
• 你在"当前标普 500"上做的回测，排除了雷曼、贝尔斯登、安然、世通——全都是巨额亏损
• 你测试的是一个由赢家组成的组合，而不是现实

如何修正：

• 使用时点数据： 包含退市记录的数据库（Norgate、Sharadar、Quandl）
• 对 ETF 和指数： 使用真实的 ETF 价格数据（幸存者问题已被正确处理）
• 如果只能用 Yahoo Finance： 要承认结果每年偏乐观 3% 至 7%

第三宗：数据窥探／数据挖掘（测试上百组参数）

它是什么： 把上百种参数组合都试一遍，直到找出一个有效的，然后宣称自己"发现"了一个好策略。

示例： 你用以下参数测试均线交叉系统：

• 快线：5、10、15、20、25、30……100（20 种）
• 慢线：50、75、100、125……300（12 种）
• 组合总数：20 × 12 = 240 次测试

把 240 种组合全测完后，你发现 MA(17, 143) 带来了 24% 的年化，而最大回撤仅 -8%。找到了！

现实是： 你找到的只是随机噪音。做 240 次测试，纯靠运气就该有 12 组"有效"（5% 的假阳性率）。MA(17, 143) 没有任何说得通的道理——它只是拟合了不会重演的历史怪癖。

如何避免数据窥探：

1. 使用基于理论的参数： 之所以用 50/200 日均线，是因为 50 天约等于两个月、200 天约等于一年。这是有逻辑的，不是优化出来的。
2. 限制优化次数： 最多测试 3 至 5 组参数，而不是 240 组
3. 样本外测试： 在 1990-2010 上优化，在 2011-2024 上测试（见第四宗罪）
4. 稳健性测试： MA(45, 190) 的效果是否与 MA(50, 200) 相近？若是，说明稳健；若只有 MA(50, 200) 有效，那就是曲线拟合。

第四宗：没有做样本外测试

它是什么： 用全部数据构建策略，然后又在同一批数据上测试它。

正确的做法：

所需的纪律：

• 绝不查看测试集数据 ——直到你的策略百分之百定稿
• 如果测试结果不好， 你**不能**回头去改策略（那就是在测试集上做数据窥探）
• 接受结果： 如果测试集显示 8% 年化、训练集是 15%，那是现实在说话

第五宗：忽略交易成本

交易成本不只是佣金。 完整的成本包括：

年度影响示例：

第六宗：忽略市场冲击（可扩展性）

它是什么： 1 万美元的回测结果很好，但放大到 100 万美元策略就崩了，因为你的订单会推动市场。

示例： 小盘股动量策略

• 股票： XYZ，市值 2 亿美元，日均成交额 200 万美元
• 你的订单： 10 万美元（占日成交量的 5%）
• 影响： 买卖价差从 0.20% 拉宽到 0.80%，你的成交价比回测差 0.60%
• 年度成本（40 笔）： 40 × 0.60% = 24% 的拖累

市场冲击的经验法则：

• 安全： 订单 < 日均成交量的 0.5%（冲击极小）
• 中等： 占成交量 0.5% 至 2%（增加 0.10% 至 0.30% 滑点）
• 较大： 占成交量 2% 至 10%（增加 0.50% 至 2.00% 滑点）
• 不可行： > 成交量的 10%（你会显著推动市场，策略无法放大）

第七宗：曲线拟合（对噪音的过拟合）

它是什么： 堆砌大量规则和条件，让策略完美贴合历史数据，却毫无预测能力。

过拟合的示例：

# Overfit strategy with 15 conditions
def overfit_strategy(data):
    # Buy if ALL conditions are true:
    if (RSI(14) > 32 and RSI(14) < 47 and
        MACD > signal_line and
        SMA(50) > SMA(200) and
        Volume > SMA_volume(20) * 1.43 and
        Close > Open and
        High - Low < ATR * 1.67 and
        Day_of_week in [1, 3, 5] and  # Monday, Wednesday, Friday
        Month not in [6, 7, 8] and    # Avoid summer
        VIX < 22.4 and VIX > 11.8 and
        Dollar_index < 104.2):
        return 'BUY'

为什么这是垃圾：

• 15 个任意设定的条件（为什么是 RSI > 32，而不是 30 或 35？）
• 精确到离谱的阈值（VIX < 22.4？认真的？）
• 自由度太多（拟合的是噪音，不是信号）
• 样本内表现： 年化 42%（哇！）
• 样本外表现： -2.8%（灾难）

过拟合的迹象：

1. 参数太多： 超过 5 个输入 = 很可能过拟合
2. 阈值过于精确： "RSI 必须在 33.7 到 41.2 之间"（毫无道理）
3. 完美的资金曲线： 20 年里没有亏损年份、最大回撤 <10%（没有后见之明是不可能的）
4. 样本内外差距巨大： 训练集年化 30%，验证集 5%（曲线拟合）
5. 说不出道理： 讲不出这条规则**为什么**有效（只知道回测里有效）

正确的回测框架

第 1 步：由假设驱动的开发

从理论出发，而不是从数据挖掘出发：

好的假设： "动量之所以持续，是因为投资者对新闻反应不足、趋势得以延续。我要买入 12 个月的赢家，持有到动量反转为止。"

糟糕的假设： "我发现在以 'r' 结尾的月份里，周二 RSI 上穿 37.4 时股票会涨。我来测测看。"

有理论支撑的策略：

• 动量： 行为层面（羊群效应、反应不足）
• 均值回归： 对新闻反应过度，向均值回归
• 价值： 市场对基本面定价有误，随时间修正
• 套息： 提供流动性所获得的补偿（利差）

第 2 步：向前推进分析

它是什么： 模拟真实世界的策略开发：不断在过去的数据上训练、在之后的数据上测试，然后向前滚动。

示例（20 年回测，2000-2020）：

好处：

• 模拟真实环境（你在过去的数据上优化，在未来的数据上交易）
• 检验策略在不同市场环境下的稳健性
• 能看出参数是否需要不断重新优化（那是坏兆头）

第 3 步：蒙特卡洛模拟

它是什么： 把交易顺序随机打乱，检验你的结果是稳健的还是只是运气好。

运作方式：

1. 取出你的历史交易（比如 10 年间的 200 笔）
2. 随机打乱顺序 1,000 次
3. 计算每一次打乱后的表现
4. 画出结果的分布

示例 results:

• 你的回测年化： 15.2%
• 蒙特卡洛均值： 12.8%（你的结果高于平均）
• 95% 置信区间： 6.4% 至 18.9%
• 盈利模拟的占比： 87%（稳健性不错）

蒙特卡洛给出的危险信号：

• 你的回测落在模拟结果的前 5%（交易顺序上走了运）
• 置信区间过宽（15.2% ± 12%，运气成分太大）
• 盈利模拟占比 < 70%（不够稳健）

Python 回测框架

下面是一套避开全部七宗罪的完整框架：

"""
Proper Backtesting Framework
Author: Plan My Retire Finance University
Date: 2026-02-22

Features:
- No look-ahead bias (all indicators shifted)
- Transaction costs included
- Walk-forward analysis
- Monte Carlo simulation
- Out-of-sample testing
"""

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import yfinance as yf
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')

class BacktestFramework:
    """
    Rigorous backtesting framework avoiding common pitfalls

    Parameters:
    -----------
    transaction_cost : float
        Cost per trade as percentage (0.001 = 0.1% per trade)
    slippage : float
        Slippage as percentage (0.0005 = 0.05%)
    """

    def __init__(self, transaction_cost=0.001, slippage=0.0005):
        self.transaction_cost = transaction_cost
        self.slippage = slippage
        self.total_cost = transaction_cost + slippage

    def backtest(self, data, signals, initial_capital=100000):
        """
        Run backtest with proper cost modeling

        Parameters:
        -----------
        data : DataFrame
            Price data with OHLC columns
        signals : Series
            Trading signals (1 = long, 0 = flat, -1 = short)
            MUST be shifted to avoid look-ahead bias
        initial_capital : float
            Starting capital

        Returns:
        --------
        results : dict
            Performance metrics
        """
        df = data.copy()
        df['Signal'] = signals
        df['Position'] = df['Signal'].diff()

        # Track portfolio
        cash = initial_capital
        shares = 0
        portfolio_values = []
        trades = []

        for i in range(len(df)):
            row = df.iloc[i]
            date = df.index[i]
            price = row['Close']

            # Calculate portfolio value
            portfolio_value = cash + (shares * price)
            portfolio_values.append({
                'Date': date,
                'Portfolio_Value': portfolio_value,
                'Cash': cash,
                'Shares': shares
            })

            # Execute trades
            if row['Position'] == 1:  # Buy signal
                if cash > 0:
                    # Apply slippage and costs
                    execution_price = price * (1 + self.slippage)
                    shares = int((cash * 0.99) / execution_price)
                    cost = shares * execution_price
                    commission = cost * self.transaction_cost
                    cash -= (cost + commission)

                    trades.append({
                        'Date': date,
                        'Type': 'BUY',
                        'Price': execution_price,
                        'Shares': shares,
                        'Cost': cost + commission
                    })

            elif row['Position'] == -1:  # Sell signal
                if shares > 0:
                    # Apply slippage and costs
                    execution_price = price * (1 - self.slippage)
                    proceeds = shares * execution_price
                    commission = proceeds * self.transaction_cost
                    cash += (proceeds - commission)

                    trades.append({
                        'Date': date,
                        'Type': 'SELL',
                        'Price': execution_price,
                        'Shares': shares,
                        'Proceeds': proceeds - commission
                    })

                    shares = 0

        # Convert to DataFrames
        equity_df = pd.DataFrame(portfolio_values).set_index('Date')
        trades_df = pd.DataFrame(trades)

        # Calculate metrics
        metrics = self.calculate_metrics(equity_df, initial_capital)

        return {
            'metrics': metrics,
            'equity_curve': equity_df,
            'trades': trades_df
        }

    def calculate_metrics(self, equity_df, initial_capital):
        """Calculate comprehensive performance metrics"""
        returns = equity_df['Portfolio_Value'].pct_change().dropna()

        final_value = equity_df['Portfolio_Value'].iloc[-1]
        total_return = (final_value - initial_capital) / initial_capital

        # CAGR
        days = (equity_df.index[-1] - equity_df.index[0]).days
        years = days / 365.25
        cagr = (final_value / initial_capital) ** (1 / years) - 1

        # Risk metrics
        volatility = returns.std() * np.sqrt(252)
        sharpe = (cagr - 0.04) / volatility if volatility > 0 else 0

        # Downside deviation (for Sortino)
        downside = returns[returns < 0].std() * np.sqrt(252)
        sortino = (cagr - 0.04) / downside if downside > 0 else 0

        # Drawdown
        cumulative = (1 + returns).cumprod()
        running_max = cumulative.expanding().max()
        drawdown = (cumulative - running_max) / running_max
        max_drawdown = drawdown.min()

        # Calmar ratio
        calmar = cagr / abs(max_drawdown) if max_drawdown != 0 else 0

        # Win rate
        win_rate = (returns > 0).sum() / len(returns) if len(returns) > 0 else 0

        return {
            'final_value': final_value,
            'total_return': total_return,
            'cagr': cagr,
            'volatility': volatility,
            'sharpe_ratio': sharpe,
            'sortino_ratio': sortino,
            'max_drawdown': max_drawdown,
            'calmar_ratio': calmar,
            'win_rate': win_rate
        }

    def walk_forward_analysis(self, data, strategy_func, train_period=252*5,
                             test_period=252, step=252):
        """
        Walk-forward analysis

        Parameters:
        -----------
        data : DataFrame
            Price data
        strategy_func : function
            Function that takes data and returns signals
        train_period : int
            Training window (days)
        test_period : int
            Testing window (days)
        step : int
            Step size (days) for rolling window

        Returns:
        --------
        results : dict
            Walk-forward results
        """
        results = []
        start = train_period

        while start + test_period < len(data):
            # Split data
            train_data = data.iloc[start - train_period:start]
            test_data = data.iloc[start:start + test_period]

            # Generate signals on training data
            train_signals = strategy_func(train_data)

            # Test on out-of-sample data
            test_signals = strategy_func(test_data)

            # Backtest
            backtest_result = self.backtest(test_data, test_signals)

            results.append({
                'train_start': train_data.index[0],
                'train_end': train_data.index[-1],
                'test_start': test_data.index[0],
                'test_end': test_data.index[-1],
                'cagr': backtest_result['metrics']['cagr'],
                'sharpe': backtest_result['metrics']['sharpe_ratio'],
                'max_dd': backtest_result['metrics']['max_drawdown']
            })

            start += step

        return pd.DataFrame(results)

    def monte_carlo_simulation(self, trades_df, n_simulations=1000, initial_capital=100000):
        """
        Monte Carlo simulation by randomizing trade sequence

        Parameters:
        -----------
        trades_df : DataFrame
            Historical trades with PnL
        n_simulations : int
            Number of simulations
        initial_capital : float
            Starting capital

        Returns:
        --------
        results : dict
            Distribution of outcomes
        """
        # Calculate PnL per trade
        buy_trades = trades_df[trades_df['Type'] == 'BUY'].reset_index(drop=True)
        sell_trades = trades_df[trades_df['Type'] == 'SELL'].reset_index(drop=True)

        if len(buy_trades) != len(sell_trades):
            print("Warning: Unmatched trades, adjusting...")
            min_len = min(len(buy_trades), len(sell_trades))
            buy_trades = buy_trades.iloc[:min_len]
            sell_trades = sell_trades.iloc[:min_len]

        trade_returns = (sell_trades['Price'].values - buy_trades['Price'].values) / buy_trades['Price'].values

        # Run simulations
        final_values = []

        for _ in range(n_simulations):
            # Randomly shuffle trade sequence
            shuffled_returns = np.random.permutation(trade_returns)

            # Calculate cumulative return
            portfolio_value = initial_capital
            for ret in shuffled_returns:
                portfolio_value *= (1 + ret)

            final_values.append(portfolio_value)

        final_values = np.array(final_values)
        cagr_values = (final_values / initial_capital) ** (1 / 10) - 1  # Assuming 10 years

        return {
            'mean_final_value': final_values.mean(),
            'median_final_value': np.median(final_values),
            'std_final_value': final_values.std(),
            'percentile_5': np.percentile(final_values, 5),
            'percentile_95': np.percentile(final_values, 95),
            'mean_cagr': cagr_values.mean(),
            'prob_profitable': (final_values > initial_capital).sum() / n_simulations,
            'distribution': final_values
        }


# Example: Dual Momentum Strategy (properly implemented)
def dual_momentum_strategy(data):
    """
    Dual momentum with NO look-ahead bias

    Returns signals shifted to avoid peeking into future
    """
    df = data.copy()

    # Calculate 12-month return (252 trading days)
    df['Return_12M'] = df['Close'].pct_change(252)

    # Signal: 1 if 12-month return positive, 0 otherwise
    # CRITICAL: Shift by 1 to avoid look-ahead bias
    df['Signal'] = (df['Return_12M'] > 0).astype(int).shift(1)

    return df['Signal'].fillna(0)


# Example usage
if __name__ == "__main__":
    # Download data
    print("Downloading SPY data...")
    data = yf.download('SPY', start='2000-01-01', end='2024-12-31', progress=False)

    # Generate signals (properly shifted)
    signals = dual_momentum_strategy(data)

    # Initialize framework
    framework = BacktestFramework(transaction_cost=0.001, slippage=0.0005)

    # Run backtest
    print("\nRunning backtest...")
    results = framework.backtest(data, signals, initial_capital=100000)

    # Print results
    print("\n" + "="*60)
    print("BACKTEST RESULTS (Dual Momentum)")
    print("="*60)
    metrics = results['metrics']
    print(f"CAGR:              {metrics['cagr']:.2%}")
    print(f"Volatility:        {metrics['volatility']:.2%}")
    print(f"Sharpe Ratio:      {metrics['sharpe_ratio']:.2f}")
    print(f"Sortino Ratio:     {metrics['sortino_ratio']:.2f}")
    print(f"Max Drawdown:      {metrics['max_drawdown']:.2%}")
    print(f"Calmar Ratio:      {metrics['calmar_ratio']:.2f}")
    print(f"Win Rate:          {metrics['win_rate']:.2%}")
    print(f"Number of Trades:  {len(results['trades'])}")
    print("="*60)

    # Walk-forward analysis
    print("\nRunning walk-forward analysis...")
    wf_results = framework.walk_forward_analysis(
        data, dual_momentum_strategy,
        train_period=252*5,  # 5 years training
        test_period=252,     # 1 year testing
        step=252             # Roll forward 1 year
    )
    print("\nWalk-Forward Results:")
    print(f"Average CAGR:      {wf_results['cagr'].mean():.2%}")
    print(f"Std Dev CAGR:      {wf_results['cagr'].std():.2%}")
    print(f"Best Year:         {wf_results['cagr'].max():.2%}")
    print(f"Worst Year:        {wf_results['cagr'].min():.2%}")

    # Monte Carlo simulation
    if len(results['trades']) > 10:
        print("\nRunning Monte Carlo simulation...")
        mc_results = framework.monte_carlo_simulation(
            results['trades'],
            n_simulations=1000,
            initial_capital=100000
        )
        print("\nMonte Carlo Results (1,000 simulations):")
        print(f"Mean Final Value:  ${mc_results['mean_final_value']:,.0f}")
        print(f"Median Final Value: ${mc_results['median_final_value']:,.0f}")
        print(f"5th Percentile:    ${mc_results['percentile_5']:,.0f}")
        print(f"95th Percentile:   ${mc_results['percentile_95']:,.0f}")
        print(f"Prob. Profitable:  {mc_results['prob_profitable']:.1%}")

要点回顾

最后的思考：50% 法则

假定你在现实中的回报只有回测年化的一半。

为什么？

• 隐性成本： 你低估了滑点、价差和市场冲击
• 执行失误： 漏掉信号、情绪化推翻决定、平台故障
• 环境切换： 市场在演化，策略在衰减
• 未知的偏差： 你的回测里还有你没发现的错误

示例：

• 回测显示年化 16%
• 现实中就按 8% 来预期
• 若你做到 10% 至 12%，那是超出预期
• 若只有 4% 至 6%，是跑输了，但仍属正常衰减范围

下一步

现在你已经有了正确回测的工具。在动用真金白银之前：

1. 用这套框架回测你的策略
2. 做向前推进分析 （它在不同市场环境下都成立吗？）
3. 做蒙特卡洛检验 （你只是运气好，还是它真的稳健？）
4. 模拟交易 6 至 12 个月 （实时记录每一个信号）
5. 从小规模做起 （第一年只用计划仓位的 25%）
6. 接受衰减 （回测 14%、实盘拿到 8%，那已经算成功）

推荐阅读：

• Evidence-Based Technical Analysis 作者：David Aronson（正确检验方法的圣经）
• Advances in Financial Machine Learning 作者：Marcos López de Prado
• Quantitative 交易 作者：Ernest Chan