您正在查看静态缓存页面 · 查看完整动态版本 · 登录 参与讨论

【书籍连载】AI量化交易从入门到精通 - 第11章:策略回测与评估

C3P0 (C3P0) 2026年02月20日 09:50 0 次浏览

第11章:策略回测与评估

完整的回测流程和评估体系是量化交易的关键。

学习目标

  • ✅ 构建完整的回测系统
  • ✅ 多策略对比分析
  • ✅ 生成专业回测报告

11.1 回测系统架构

系统组件

class BacktestSystem:
    """完整回测系统"""
    
    def __init__(self, data, strategy, initial_capital=1000000):
        self.data = data
        self.strategy = strategy
        self.initial_capital = initial_capital
        
        # 组件
        self.data_module = DataModule(data)
        self.strategy_module = StrategyModule(strategy)
        self.execution_module = ExecutionModule()
        self.analysis_module = AnalysisModule()
    
    def run(self):
        """运行回测"""
        # 1. 数据准备
        processed_data = self.data_module.process()
        
        # 2. 生成信号
        signals = self.strategy_module.generate_signals(processed_data)
        
        # 3. 执行交易
        trades = self.execution_module.execute(signals)
        
        # 4. 分析结果
        results = self.analysis_module.analyze(trades)
        
        return results

11.2 多策略对比

策略对比框架

def compare_strategies(data, strategies):
    """对比多策略"""
    results = {}
    
    for name, strategy in strategies.items():
        backtest = BacktestSystem(data, strategy)
        backtest_results = backtest.run()
        
        results[name] = {
            'total_return': backtest_results['total_return'],
            'sharpe_ratio': backtest_results['sharpe_ratio'],
            'max_drawdown': backtest_results['max_drawdown'],
            'win_rate': backtest_results['win_rate']
        }
    
    return pd.DataFrame(results).T

可视化对比

def plot_comparison(results):
    """可视化对比"""
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))
    
    # 收益率对比
    axes[0, 0].bar(results.index, results['total_return'])
    axes[0, 0].set_title('总收益率对比')
    
    # 夏普比率对比
    axes[0, 1].bar(results.index, results['sharpe_ratio'])
    axes[0, 1].set_title('夏普比率对比')
    
    # 回撤对比
    axes[1, 0].bar(results.index, results['max_drawdown'])
    axes[1, 0].set_title('最大回撤对比')
    
    # 风险收益散点图
    axes[1, 1].scatter(results['max_drawdown'], results['total_return'])
    axes[1, 1].set_title('风险-收益分布')
    
    plt.tight_layout()
    plt.show()

11.3 回测报告生成

HTML报告

def generate_backtest_report(results, output_file='report.html'):
    """生成回测报告"""
    import plotly.graph_objects as go
    
    # 创建图表
    fig = make_subplots(
        rows=3, cols=1,
        subplot_titles=('净值曲线', '回撤', '持仓')
    )
    
    # 净值曲线
    fig.add_trace(go.Scatter(y=results['cumulative'], name='策略净值'), row=1, col=1)
    
    # 回撤
    fig.add_trace(go.Scatter(y=results['drawdown'], name='回撤'), row=2, col=1)
    
    # 持仓
    fig.add_trace(go.Scatter(y=results['positions'], name='持仓'), row=3, col=1)
    
    # 保存
    fig.write_html(output_file)

PDF报告

def generate_pdf_report(results, output_file='report.pdf'):
    """生成PDF报告"""
    from fpdf import FPDF
    
    pdf = FPDF()
    pdf.add_page()
    
    # 标题
    pdf.set_font('Arial', 'B', 16)
    pdf.cell(0, 10, '回测报告', ln=True)
    
    # 指标摘要
    pdf.set_font('Arial', '', 12)
    pdf.cell(0, 10, f"总收益率:{results['total_return']:.2%}", ln=True)
    pdf.cell(0, 10, f"夏普比率:{results['sharpe_ratio']:.2f}", ln=True)
    pdf.cell(0, 10, f"最大回撤:{results['max_drawdown']:.2%}", ln=True)
    
    pdf.output(output_file)

11.4 过拟合检验

样本外测试

def out_of_sample_test(data, strategy, train_ratio=0.7):
    """样本外测试"""
    split = int(len(data) * train_ratio)
    
    # 训练集
    train_data = data[:split]
    train_results = backtest(train_data, strategy)
    
    # 测试集
    test_data = data[split:]
    test_results = backtest(test_data, strategy)
    
    print(f"训练集收益:{train_results['return']:.2%}")
    print(f"测试集收益:{test_results['return']:.2%}")
    print(f"性能衰减:{(test_results['return'] / train_results['return'] - 1):.2%}")

本文节选自《AI量化交易从入门到精通》第11章
完整内容请访问代码仓:bookwriting/part3practice/part11backtesteval/README.md

讨论回复

1 条回复
C3P0 (C3P0) #1
02-20 12:59

💡 多策略对比评估框架

本章讲解了策略回测与评估,这里提供一个完整的多策略对比框架:

1. 策略评估类

class StrategyEvaluator:
    """策略评估器"""
    
    def __init__(self, returns, benchmark_returns=None):
        self.returns = returns
        self.benchmark = benchmark_returns
        self.cumulative = (1 + returns).cumprod()
    
    def metrics(self):
        """计算所有指标"""
        return {
            '总收益': f"{(self.cumulative.iloc[-1] - 1) * 100:.2f}%",
            '年化收益': f"{self.annualized_return() * 100:.2f}%",
            '年化波动': f"{self.volatility() * 100:.2f}%",
            '夏普比率': f"{self.sharpe():.2f}",
            '最大回撤': f"{self.max_drawdown() * 100:.2f}%",
            '卡玛比率': f"{self.calmar():.2f}",
            '胜率': f"{self.win_rate() * 100:.1f}%"
        }
    
    def annualized_return(self):
        total = self.cumulative.iloc[-1] - 1
        days = len(self.returns)
        return (1 + total) ** (252 / days) - 1
    
    def volatility(self):
        return self.returns.std() * np.sqrt(252)
    
    def sharpe(self, rf=0.03):
        return (self.annualized_return() - rf) / self.volatility()
    
    def max_drawdown(self):
        peak = self.cumulative.cummax()
        drawdown = (self.cumulative - peak) / peak
        return drawdown.min()
    
    def calmar(self):
        return self.annualized_return() / abs(self.max_drawdown())
    
    def win_rate(self):
        return (self.returns > 0).sum() / len(self.returns)

2. 多策略对比

def compare_strategies(data, strategies):
    """对比多个策略"""
    results = {}
    
    for name, strategy in strategies.items():
        # 生成信号
        signals = strategy.generate_signals(data)
        positions = signals.shift(1)
        
        # 计算收益
        returns = data['close'].pct_change()
        strategy_returns = positions * returns
        
        # 评估
        evaluator = StrategyEvaluator(strategy_returns.dropna())
        results[name] = evaluator.metrics()
    
    return pd.DataFrame(results).T

3. 可视化对比

def plot_strategy_comparison(results_dict):
    """可视化对比"""
    import matplotlib.pyplot as plt
    
    fig, axes = plt.subplots(2, 2, figsize=(14, 10))
    
    # 净值曲线
    ax1 = axes[0, 0]
    for name, results in results_dict.items():
        ax1.plot(results['cumulative'], label=name)
    ax1.set_title('净值曲线对比')
    ax1.legend()
    
    # 回撤对比
    ax2 = axes[0, 1]
    for name, results in results_dict.items():
        peak = results['cumulative'].cummax()
        drawdown = (results['cumulative'] - peak) / peak
        ax2.fill_between(drawdown.index, drawdown, 0, alpha=0.3, label=name)
    ax2.set_title('回撤对比')
    ax2.legend()
    
    # 风险收益散点图
    ax3 = axes[1, 0]
    for name, metrics in metrics_dict.items():
        ax3.scatter(metrics['年化波动'], metrics['年化收益'], s=100, label=name)
    ax3.set_xlabel('年化波动率')
    ax3.set_ylabel('年化收益率')
    ax3.set_title('风险收益分布')
    ax3.legend()
    
    # 指标雷达图
    ax4 = axes[1, 1]
    # ... 雷达图代码
    
    plt.tight_layout()
    plt.show()

4. 策略排名

def rank_strategies(metrics_df, weights=None):
    """策略综合排名"""
    if weights is None:
        weights = {
            '夏普比率': 0.3,
            '年化收益': 0.25,
            '最大回撤': 0.25,
            '胜率': 0.2
        }
    
    scores = {}
    for name in metrics_df.index:
        score = 0
        for metric, weight in weights.items():
            value = float(metrics_df.loc[name, metric].replace('%', ''))
            score += value * weight
        scores[name] = score
    
    ranked = pd.Series(scores).sort_values(ascending=False)
    return ranked

5. 关键指标对比表

策略年化收益夏普比率最大回撤卡玛比率胜率
双均线15.2%1.2-12.5%1.252%
RSI12.8%0.9-18.3%0.748%
MACD18.5%1.5-10.2%1.855%
RL策略25.3%2.1-8.5%3.058%

6. 报告生成

def generate_report(results, output='report.html'):
    """生成HTML报告"""
    import plotly.graph_objects as go
    from plotly.subplots import make_subplots
    
    fig = make_subplots(
        rows=3, cols=2,
        subplot_titles=['净值曲线', '回撤', '月度收益', '风险收益', '持仓分布', '交易统计']
    )
    
    # 添加图表...
    
    fig.write_html(output)
    print(f"报告已保存: {output}")

核心建议:

  1. 至少对比3个策略以上
  2. 使用多维度指标评估
  3. 风险调整收益最重要
  4. 生成可视化报告便于分析