05-实验分析方法

实验分析方法

1. 参数敏感性分析

1.1 单参数敏感性

def single_parameter_sensitivity(strategy, param_name, param_values, 
                                   data, backtest_func):
    """
    单参数敏感性分析
    
    参数:
        strategy: 策略函数
        param_name: 参数名
        param_values: 参数值列表
        data: 回测数据
        backtest_func: 回测函数
    
    返回:
        results: 分析结果
    """
    results = []
    
    for value in param_values:
        # 设置参数
        params = {param_name: value}
        
        # 回测
        performance = backtest_func(strategy, data, params)
        
        results.append({
            param_name: value,
            'return': performance['return'],
            'sharpe': performance['sharpe'],
            'max_drawdown': performance['max_drawdown']
        })
    
    return pd.DataFrame(results)
 
# 示例：分析K值对Top-K策略的影响
k_values = [10, 20, 30, 40, 50]
results = single_parameter_sensitivity(
    topk_strategy, 'k', k_values, 
    returns_data, backtest
)

1.2 多参数敏感性

def multi_parameter_sensitivity(strategy, param_grid, data, backtest_func):
    """
    多参数敏感性分析
    
    参数:
        strategy: 策略函数
        param_grid: 参数网格
        data: 回测数据
        backtest_func: 回测函数
    
    返回:
        results: 分析结果
    """
    from itertools import product
    
    # 生成所有参数组合
    param_combinations = list(product(*param_grid.values()))
    
    results = []
    
    for combination in param_combinations:
        # 设置参数
        params = dict(zip(param_grid.keys(), combination))
        
        # 回测
        performance = backtest_func(strategy, data, params)
        
        results.append({
            **params,
            'sharpe': performance['sharpe'],
            'return': performance['return']
        })
    
    return pd.DataFrame(results)
 
# 示例
param_grid = {
    'k': [20, 30, 40],
    'risk_aversion': [0.5, 1.0, 2.0],
    'max_weight': [0.05, 0.1, 0.15]
}
 
results = multi_parameter_sensitivity(mv_strategy, param_grid, returns_data, backtest)

1.3 敏感性热力图

import seaborn as sns
import matplotlib.pyplot as plt
 
def plot_sensitivity_heatmap(results, pivot_col, pivot_row, value_col):
    """
    绘制敏感性热力图
    
    参数:
        results: 分析结果
        pivot_col: 列参数
        pivot_row: 行参数
        value_col: 值参数
    
    返回:
        fig: 图形对象
    """
    # 创建透视表
    pivot_table = results.pivot(
        index=pivot_row,
        columns=pivot_col,
        values=value_col
    )
    
    # 绘制热力图
    fig, ax = plt.subplots(figsize=(10, 8))
    sns.heatmap(pivot_table, annot=True, fmt='.3f', 
                cmap='RdYlGn', center=0, ax=ax)
    
    ax.set_title(f'{value_col} Sensitivity')
    plt.tight_layout()
    
    return fig
 
# 示例
fig = plot_sensitivity_heatmap(results, 'risk_aversion', 'k', 'sharpe')
plt.show()

2. 持仓周期研究

2.1 调仓频率类型

日频调仓：每天调整持仓

• 优点：及时反映市场变化
• 缺点：交易成本高

周频调仓：每周调整持仓

• 平衡成本和及时性
• 常用频率

月频调仓：每月调整持仓

• 优点：成本低
• 缺点：反应滞后

2.2 持仓周期分析

def holding_period_study(predictions, holding_periods=[1, 5, 20]):
    """
    持仓周期研究
    
    参数:
        predictions: 预测分数
        holding_periods: 持仓周期列表（天数）
    
    返回:
        results: 分析结果
    """
    results = []
    
    for period in holding_periods:
        # 生成持仓信号（每period天调仓一次）
        signals = generate_signals(predictions, rebalance_freq=period)
        
        # 回测
        performance = backtest(signals)
        
        results.append({
            'holding_period': period,
            'return': performance['return'],
            'sharpe': performance['sharpe'],
            'turnover': performance['turnover']
        })
    
    return pd.DataFrame(results)
 
# 示例
holding_periods = [1, 5, 10, 20]  # 1天、1周、2周、1月
results = holding_period_study(predictions, holding_periods)

3. 样本外验证

3.1 滚动窗口验证

def rolling_window_backtest(data, strategy_func, 
                               window_size=252, test_size=63):
    """
    滚动窗口验证
    
    参数:
        data: 数据
        strategy_func: 策略函数
        window_size: 训练窗口大小
        test_size: 测试窗口大小
    
    返回:
        results: 验证结果
    """
    results = []
    
    for i in range(window_size, len(data), test_size):
        # 训练集
        train_data = data[i-window_size:i]
        
        # 测试集
        test_data = data[i:i+test_size]
        
        # 训练模型
        model = train_model(train_data)
        
        # 生成预测
        predictions = model.predict(test_data)
        
        # 回测
        performance = backtest(predictions, test_data)
        
        results.append({
            'test_start': test_data.index[0],
            'test_end': test_data.index[-1],
            'return': performance['return'],
            'sharpe': performance['sharpe']
        })
    
    return pd.DataFrame(results)
 
# 示例
results = rolling_window_backtest(data, strategy_func, 
                                 window_size=252, test_size=63)

3.2 扩展窗口验证

def expanding_window_backtest(data, strategy_func, 
                                initial_size=252, test_size=63):
    """
    扩展窗口验证
    
    参数:
        data: 数据
        strategy_func: 策略函数
        initial_size: 初始训练集大小
        test_size: 测试窗口大小
    
    返回:
        results: 验证结果
    """
    results = []
    
    for i in range(initial_size, len(data), test_size):
        # 训练集（不断扩大）
        train_data = data[:i]
        
        # 测试集
        test_data = data[i:i+test_size]
        
        # 训练模型
        model = train_model(train_data)
        
        # 生成预测
        predictions = model.predict(test_data)
        
        # 回测
        performance = backtest(predictions, test_data)
        
        results.append({
            'test_start': test_data.index[0],
            'test_end': test_data.index[-1],
            'train_size': len(train_data),
            'return': performance['return'],
            'sharpe': performance['sharpe']
        })
    
    return pd.DataFrame(results)
 
# 示例
results = expanding_window_backtest(data, strategy_func, 
                                  initial_size=252, test_size=63)

4. 避免过拟合

4.1 样本外验证

def out_of_sample_validation(data, split_ratio=0.7):
    """
    样本外验证
    
    参数:
        data: 数据
        split_ratio: 训练集比例
    
    返回:
        train_data: 训练集
        test_data: 测试集
    """
    # 划分数据
    split_idx = int(len(data) * split_ratio)
    
    train_data = data[:split_idx]
    test_data = data[split_idx:]
    
    return train_data, test_data
 
# 使用示例
train_data, test_data = out_of_sample_validation(data, split_ratio=0.7)
 
# 训练模型
model = train_model(train_data)
 
# 测试模型
predictions = model.predict(test_data)
performance = evaluate(predictions, test_data)

4.2 参数限制

# ❌ 错误做法：过度优化
params = {
    'k': 23,
    'lookback': 17,
    'threshold': 0.032,
    'smoothing': 0.7,
    # ... 太多参数
}
 
# ✅ 正确做法：少量关键参数
params = {
    'k': 20,        # 清晰的含义
    'lookback': 20   # 基于经济逻辑
}

4.3 交叉验证

from sklearn.model_selection import TimeSeriesSplit
 
def time_series_cv(X, y, model_func, n_splits=5):
    """
    时间序列交叉验证
    
    参数:
        X: 特征
        y: 目标
        model_func: 模型函数
        n_splits: 折数
    
    返回:
        scores: 每折的分数
    """
    tscv = TimeSeriesSplit(n_splits=n_splits)
    scores = []
    
    for train_idx, test_idx in tscv.split(X):
        X_train, X_test = X[train_idx], X[test_idx]
        y_train, y_test = y[train_idx], y[test_idx]
        
        # 训练模型
        model = model_func()
        model.fit(X_train, y_train)
        
        # 预测
        y_pred = model.predict(X_test)
        
        # 评估
        score = evaluate(y_pred, y_test)
        scores.append(score)
    
    return scores
 
# 示例
scores = time_series_cv(X, y, model_func, n_splits=5)
print(f"平均分数: {np.mean(scores):.4f}")
print(f"标准差: {np.std(scores):.4f}")

总结

实验分析是验证策略有效性的关键：

1. 参数敏感性：了解参数变化的影响
2. 持仓周期：研究调仓频率的影响
3. 样本外验证：防止过拟合
4. 交叉验证：提高结果的可靠性

建议：

• 务必进行样本外验证
• 使用多种验证方法
• 关注参数稳定性
• 避免过度优化