回测过拟合：量化策略死亡的99%原因

import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LinearRegression
from sklearn.preprocessing import PolynomialFeatures
from sklearn.pipeline import make_pipeline
import matplotlib.pyplot as plt

# 生成模拟数据
np.random.seed(42)
X = np.linspace(0, 1, 50).reshape(-1, 1)
y = np.sin(2 * np.pi * X).ravel() + np.random.normal(0, 0.1, 50)

# 不同复杂度的模型
degrees = [1, 3, 15]

plt.figure(figsize=(12, 4))
for i, degree in enumerate(degrees):
    plt.subplot(1, 3, i + 1)
    
    # 训练模型
    model = make_pipeline(PolynomialFeatures(degree), LinearRegression())
    model.fit(X, y)
    
    # 预测
    X_test = np.linspace(0, 1, 100).reshape(-1, 1)
    y_pred = model.predict(X_test)
    
    plt.scatter(X, y, s=10, label='Data')
    plt.plot(X_test, y_pred, 'r-', label=f'Degree {degree}')
    plt.legend()
    plt.title(f'Degree = {degree}')

plt.tight_layout()
plt.savefig('bias_variance_tradeoff.png', dpi=150, bbox_inches='tight')

def walk_forward_analysis(data, strategy, window_size=252):
    """
    滚动窗口分析：模拟实盘逐步获得数据
    """
    results = []
    
    for start in range(0, len(data) - window_size, window_size // 4):
        # 训练窗口
        train = data[start:start + window_size]
        
        # 测试窗口
        test = data[start + window_size:start + window_size * 2]
        
        # 优化参数（只在训练集上）
        params = optimize(strategy, train)
        
        # 测试（只在测试集上）
        performance = backtest(strategy, params, test)
        
        results.append(performance)
    
    return results

# 如果样本外表现明显差于样本内 → 过拟合
is_sharpe = 2.5
oos_sharpe = 1.2  # 明显下降
print(f"IS夏普: {is_sharpe}, OOS夏普: {oos_sharpe}")
print(f"衰减率: {(is_sharpe - oos_sharpe) / is_sharpe:.1%}")  # 52%衰减！

def parameter_stability_test(strategy, data, param_ranges):
    """
    测试参数微小变化对性能的影响
    """
    base_params = optimize(strategy, data)
    base_performance = backtest(strategy, base_params, data)
    
    stability_scores = {}
    
    for param_name, param_range in param_ranges.items():
        scores = []
        for value in param_range:
            test_params = base_params.copy()
            test_params[param_name] = value
            
            perf = backtest(strategy, test_params, data)
            scores.append(perf)
        
        # 计算性能波动
        stability_scores[param_name] = np.std(scores)
    
    return stability_scores

# 如果参数微小变化导致性能大幅波动 → 过拟合
stability = parameter_stability_test(my_strategy, data, {
    'ma_window': range(18, 23),
    'threshold': [0.01, 0.012, 0.014, 0.016, 0.018]
})

print("参数稳定性（标准差）:", stability)
# 输出: {'ma_window': 0.05, 'threshold': 0.8}  ← threshold不稳定！

def white_reality_check(strategy, data, n_bootstrap=1000):
    """
    White's Reality Check: 检测策略是否真的有预测能力
    """
    # 原始策略性能
    actual_performance = backtest(strategy, optimize(strategy, data), data)
    
    # 自助法（Bootstrap）生成零分布
    null_dist = []
    for _ in range(n_bootstrap):
        # 随机打乱标签（破坏时序关系）
        shuffled_data = data.copy()
        shuffled_data['returns'] = np.random.permutation(data['returns'])
        
        # 重新优化（在打乱的数据上）
        shuffled_params = optimize(strategy, shuffled_data)
        shuffled_perf = backtest(strategy, shuffled_params, shuffled_data)
        
        null_dist.append(shuffled_perf)
    
    # p值：零分布中大于等于实际性能的比例
    p_value = np.mean(np.array(null_dist) >= actual_performance)
    
    return p_value

# p值 > 0.05 → 策略可能没有真实预测能力
p_val = white_reality_check(my_strategy, data)
print(f"White Reality Check p-value: {p_val:.4f}")

from sklearn.model_selection import TimeSeriesSplit

def time_series_cv(strategy, data, n_splits=5):
    """
    时间序列交叉验证（保持时序性）
    """
    tscv = TimeSeriesSplit(n_splits=n_splits)
    cv_scores = []
    
    for train_idx, test_idx in tscv.split(data):
        train = data.iloc[train_idx]
        test = data.iloc[test_idx]
        
        params = optimize(strategy, train)
        score = backtest(strategy, params, test)
        
        cv_scores.append(score)
    
    return cv_scores

# 如果CV分数波动大 → 过拟合
cv_scores = time_series_cv(my_strategy, data, n_splits=5)
print(f"CV夏普: {np.mean(cv_scores):.2f} ± {np.std(cv_scores):.2f}")

# ✅ 分割数据
train = data[:'2018-01-01']
test = data['2018-01-01':]

# 在训练集上优化
best_params = None
best_sharpe = -np.inf
for ma_short in range(5, 50):
    for ma_long in range(50, 200):
        sharpe = backtest_ma_strategy(train, ma_short, ma_long)
        if sharpe > best_sharpe:
            best_sharpe = sharpe
            best_params = (ma_short, ma_long)

print(f"训练集最优: {best_params}, 夏普: {best_sharpe}")

# 在测试集上验证
test_sharpe = backtest_ma_strategy(test, *best_params)
print(f"测试集夏普: {test_sharpe}")
print(f"衰减率: {(best_sharpe - test_sharpe) / best_sharpe:.1%}")

# 输出:
# 训练集最优: (12, 67), 夏普: 2.8
# 测试集夏普: 1.1
# 衰减率: 60.7% ← 严重过拟合！