400 lines
14 KiB
Python
400 lines
14 KiB
Python
# -*- 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)
|