# -*- coding: utf-8 -*- """ 量化策略回测分析工具 用于分析策略回测结果，计算各种绩效指标 """ import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns from datetime import datetime, timedelta import warnings warnings.filterwarnings('ignore') # 设置中文字体 plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans'] plt.rcParams['axes.unicode_minus'] = False class BacktestAnalyzer: """回测结果分析器""" def __init__(self, returns, benchmark_returns=None, risk_free_rate=0.03): """ 初始化分析器 参数: returns: 策略收益率序列 (pandas Series) benchmark_returns: 基准收益率序列 (pandas Series) risk_free_rate: 无风险利率 """ self.returns = returns self.benchmark_returns = benchmark_returns self.risk_free_rate = risk_free_rate self.results = {} def calculate_basic_metrics(self): """计算基本指标""" # 总收益率 total_return = (1 + self.returns).prod() - 1 # 年化收益率 n_years = len(self.returns) / 252 # 假设一年252个交易日 annual_return = (1 + total_return) ** (1 / n_years) - 1 # 年化波动率 annual_volatility = self.returns.std() * np.sqrt(252) # 夏普比率 sharpe_ratio = (annual_return - self.risk_free_rate) / annual_volatility # 最大回撤 cumulative_returns = (1 + self.returns).cumprod() running_max = cumulative_returns.expanding().max() drawdown = (cumulative_returns - running_max) / running_max max_drawdown = drawdown.min() # 胜率 win_rate = (self.returns > 0).sum() / len(self.returns) # 盈亏比 avg_win = self.returns[self.returns > 0].mean() avg_loss = abs(self.returns[self.returns <= 0].mean()) profit_factor = avg_win / avg_loss if avg_loss > 0 else np.inf basic_metrics = { '总收益率': f'{total_return:.2%}', '年化收益率': f'{annual_return:.2%}', '年化波动率': f'{annual_volatility:.2%}', '夏普比率': f'{sharpe_ratio:.2f}', '最大回撤': f'{max_drawdown:.2%}', '胜率': f'{win_rate:.2%}', '盈亏比': f'{profit_factor:.2f}', '交易次数': len(self.returns) } self.results['basic_metrics'] = basic_metrics self.results['cumulative_returns'] = cumulative_returns self.results['drawdown'] = drawdown return basic_metrics def calculate_relative_metrics(self): """计算相对指标（如果有基准）""" if self.benchmark_returns is None: return None # 计算相对收益率 excess_returns = self.returns - self.benchmark_returns # 信息比率 tracking_error = excess_returns.std() * np.sqrt(252) annual_excess_return = excess_returns.mean() * 252 information_ratio = annual_excess_return / tracking_error if tracking_error > 0 else 0 # Beta值 covariance = np.cov(self.returns, self.benchmark_returns)[0, 1] benchmark_variance = np.var(self.benchmark_returns) beta = covariance / benchmark_variance if benchmark_variance > 0 else 0 # Alpha值 annual_return = self.returns.mean() * 252 benchmark_annual_return = self.benchmark_returns.mean() * 252 alpha = annual_return - (self.risk_free_rate + beta * (benchmark_annual_return - self.risk_free_rate)) relative_metrics = { '信息比率': f'{information_ratio:.2f}', 'Beta值': f'{beta:.2f}', 'Alpha值': f'{alpha:.2%}', '相对收益率': f'{excess_returns.mean():.2%}', '跟踪误差': f'{tracking_error:.2%}' } self.results['relative_metrics'] = relative_metrics self.results['excess_returns'] = excess_returns return relative_metrics def plot_returns_comparison(self, save_path=None): """绘制收益对比图""" fig, axes = plt.subplots(3, 1, figsize=(12, 10)) # 1. 累计收益对比 ax1 = axes[0] cumulative_returns = self.results['cumulative_returns'] ax1.plot(cumulative_returns.index, cumulative_returns.values, label='策略', linewidth=2) if self.benchmark_returns is not None: benchmark_cumulative = (1 + self.benchmark_returns).cumprod() ax1.plot(benchmark_cumulative.index, benchmark_cumulative.values, label='基准', linewidth=2, alpha=0.7) ax1.set_title('累计收益对比', fontsize=14) ax1.legend() ax1.grid(True, alpha=0.3) ax1.set_ylabel('累计收益') # 2. 回撤图 ax2 = axes[1] drawdown = self.results['drawdown'] ax2.fill_between(drawdown.index, drawdown.values, 0, color='red', alpha=0.3) ax2.plot(drawdown.index, drawdown.values, color='red', linewidth=1) ax2.set_title('策略回撤', fontsize=14) ax2.grid(True, alpha=0.3) ax2.set_ylabel('回撤') # 3. 收益分布 ax3 = axes[2] self.returns.hist(bins=30, ax=ax3, alpha=0.7, density=True) ax3.axvline(self.returns.mean(), color='red', linestyle='--', label=f'均值: {self.returns.mean():.2%}') ax3.set_title('收益分布', fontsize=14) ax3.legend() ax3.grid(True, alpha=0.3) ax3.set_xlabel('日收益率') ax3.set_ylabel('密度') plt.tight_layout() if save_path: plt.savefig(save_path, dpi=300, bbox_inches='tight') return fig def plot_monthly_returns_heatmap(self, save_path=None): """绘制月度收益热力图""" # 创建月度收益数据 monthly_returns = self.returns.resample('M').apply(lambda x: (1 + x).prod() - 1) monthly_returns.index = monthly_returns.index.to_period('M') # 创建透视表 monthly_returns_df = pd.DataFrame({ 'Year': monthly_returns.index.year, 'Month': monthly_returns.index.month, 'Return': monthly_returns.values }) pivot_table = monthly_returns_df.pivot(index='Year', columns='Month', values='Return') # 绘制热力图 plt.figure(figsize=(12, 8)) sns.heatmap(pivot_table, annot=True, fmt='.1%', cmap='RdYlGn', center=0, cbar_kws={'label': '月度收益率'}) plt.title('月度收益热力图', fontsize=16) plt.xlabel('月份') plt.ylabel('年份') if save_path: plt.savefig(save_path, dpi=300, bbox_inches='tight') return plt.gcf() def generate_report(self, save_path=None): """生成完整的分析报告""" # 计算所有指标 basic_metrics = self.calculate_basic_metrics() relative_metrics = self.calculate_relative_metrics() # 创建报告 report = [] report.append("=" * 60) report.append("量化策略回测分析报告") report.append("=" * 60) report.append(f"分析时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") report.append(f"回测期间: {self.returns.index[0].strftime('%Y-%m-%d')} 至 {self.returns.index[-1].strftime('%Y-%m-%d')}") report.append(f"交易日数: {len(self.returns)}") report.append("") # 基本指标 report.append("基本指标:") report.append("-" * 30) for key, value in basic_metrics.items(): report.append(f"{key}: {value}") report.append("") # 相对指标 if relative_metrics: report.append("相对指标 (vs 基准):") report.append("-" * 30) for key, value in relative_metrics.items(): report.append(f"{key}: {value}") report.append("") # 风险提示 report.append("风险提示:") report.append("-" * 30) report.append("1. 本报告基于历史数据，不构成投资建议") report.append("2. 过往业绩不代表未来表现") report.append("3. 投资有风险，入市需谨慎") report_text = "\n".join(report) if save_path: with open(save_path, 'w', encoding='utf-8') as f: f.write(report_text) return report_text def create_sample_backtest_data(): """创建示例回测数据""" # 生成日期序列 dates = pd.date_range(start='2020-01-01', end='2023-12-31', freq='D') # 生成随机收益率（正态分布） np.random.seed(42) returns = np.random.normal(0.0005, 0.02, len(dates)) # 日收益率0.05%，波动率2% # 创建收益率序列 returns_series = pd.Series(returns, index=dates) # 生成基准收益率（相关性0.7） benchmark_returns = 0.7 * returns + np.random.normal(0.0003, 0.015, len(dates)) benchmark_series = pd.Series(benchmark_returns, index=dates) return returns_series, benchmark_series def analyze_strategy_performance(returns_data, benchmark_data=None, strategy_name="策略"): """分析策略表现的主函数""" print(f"正在分析 {strategy_name} 的表现...") # 创建分析器 analyzer = BacktestAnalyzer(returns_data, benchmark_data) # 生成报告 report = analyzer.generate_report() print(report) # 绘制图表 fig = analyzer.plot_returns_comparison() plt.show() # 月度热力图 heatmap_fig = analyzer.plot_monthly_returns_heatmap() plt.show() return analyzer # 策略优化工具 class StrategyOptimizer: """策略参数优化器""" def __init__(self, strategy_func, param_ranges, returns_data): """ 初始化优化器 参数: strategy_func: 策略函数 param_ranges: 参数范围字典 returns_data: 收益率数据 """ self.strategy_func = strategy_func self.param_ranges = param_ranges self.returns_data = returns_data self.results = [] def grid_search(self, metric='sharpe_ratio'): """网格搜索最优参数""" print("开始网格搜索参数优化...") # 生成参数组合 import itertools param_names = list(self.param_ranges.keys()) param_values = list(self.param_ranges.values()) best_params = None best_metric = -np.inf # 遍历所有参数组合 for param_combo in itertools.product(*param_values): params = dict(zip(param_names, param_combo)) try: # 运行策略 strategy_returns = self.strategy_func(self.returns_data, **params) # 计算指标 analyzer = BacktestAnalyzer(strategy_returns) basic_metrics = analyzer.calculate_basic_metrics() # 获取优化指标值 if metric == 'sharpe_ratio': current_metric = float(basic_metrics['夏普比率'].replace('%', '')) elif metric == 'total_return': current_metric = float(basic_metrics['总收益率'].replace('%', '')) elif metric == 'max_drawdown': current_metric = -float(basic_metrics['最大回撤'].replace('%', '')) # 记录结果 self.results.append({ 'params': params, 'metric': current_metric, 'basic_metrics': basic_metrics }) # 更新最优参数 if current_metric > best_metric: best_metric = current_metric best_params = params print(f"参数: {params}, {metric}: {current_metric:.4f}") except Exception as e: print(f"参数 {params} 运行失败: {str(e)}") continue print(f"\n最优参数: {best_params}") print(f"最优{metric}: {best_metric:.4f}") return best_params, best_metric def plot_optimization_results(self, param1, param2, metric='sharpe_ratio'): """绘制参数优化结果热力图""" # 提取两个参数的结果 param1_values = [] param2_values = [] metric_values = [] for result in self.results: if param1 in result['params'] and param2 in result['params']: param1_values.append(result['params'][param1]) param2_values.append(result['params'][param2]) metric_values.append(result['metric']) # 创建DataFrame optimization_df = pd.DataFrame({ param1: param1_values, param2: param2_values, metric: metric_values }) # 创建透视表 pivot_table = optimization_df.pivot(index=param2, columns=param1, values=metric) # 绘制热力图 plt.figure(figsize=(10, 8)) sns.heatmap(pivot_table, annot=True, fmt='.3f', cmap='viridis') plt.title(f'{param1} vs {param2} 对 {metric} 的影响') plt.xlabel(param1) plt.ylabel(param2) return plt.gcf() if __name__ == "__main__": # 创建示例数据 print("创建示例回测数据...") returns_data, benchmark_data = create_sample_backtest_data() # 分析策略表现 analyzer = analyze_strategy_performance(returns_data, benchmark_data, "双均线策略") print("\n分析完成！") print("=" * 60) print("工具使用说明:") print("1. BacktestAnalyzer: 回测结果分析器") print("2. StrategyOptimizer: 策略参数优化器") print("3. 支持多种图表绘制和报告生成") print("=" * 60)