任务7: 回测验证脚本 #8
Description
概述
基于前期完成的所有策略模块，开发全面的自动化回测验证系统，包括单一时期回测、多时期滚动回测、参数优化回测和压力测试，确保Strategy-21在各种市场条件下的稳定性和盈利能力。
目标
- 建立完整的自动化回测框架
- 实现多维度性能评估体系
- 提供参数优化和敏感性分析
- 生成详细的回测报告和可视化图表
技术要求
核心功能
-
多类型回测支持
- 标准回测：固定时期完整回测
- 滚动回测：多个时间窗口滚动验证
- Walk-Forward分析：前进分析优化
- 蒙特卡罗回测：随机抽样验证
-
性能评估指标
- 收益指标：总收益率、年化收益率、超额收益
- 风险指标：最大回撤、波动率、VaR、CVaR
- 风险调整指标：夏普比率、卡尔马比率、索提诺比率
- 交易统计：胜率、平均盈亏比、交易频率
-
市场环境分析
- 牛市表现分析
- 熊市表现分析
- 震荡市表现分析
- 高波动率时期表现
实现细节
回测框架核心 (backtesting_framework.py)
import pandas as pd
import numpy as np
from typing import Dict, List, Any, Optional, Tuple
from datetime import datetime, timedelta
import json
import logging
from pathlib import Path
import multiprocessing as mp
from concurrent.futures import ProcessPoolExecutor
import matplotlib.pyplot as plt
import seaborn as sns
class BacktestingFramework:
"""Strategy-21自动化回测框架"""
def __init__(self, config_path: str):
self.config = self.load_config(config_path)
self.logger = self.setup_logging()
self.results = {}
def load_config(self, config_path: str) -> Dict[str, Any]:
"""加载回测配置"""
with open(config_path, 'r') as f:
return json.load(f)
def setup_logging(self) -> logging.Logger:
"""设置日志记录"""
logger = logging.getLogger('BacktestFramework')
logger.setLevel(logging.INFO)
handler = logging.FileHandler('backtest_results/backtest.log')
formatter = logging.Formatter(
'%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
handler.setFormatter(formatter)
logger.addHandler(handler)
return logger
def run_complete_validation(self) -> Dict[str, Any]:
"""运行完整验证流程"""
self.logger.info("开始完整回测验证流程")
results = {}
# 1. 标准回测
self.logger.info("执行标准回测")
results['standard'] = self.run_standard_backtest()
# 2. 滚动回测
self.logger.info("执行滚动回测")
results['rolling'] = self.run_rolling_backtest()
# 3. Walk-Forward分析
self.logger.info("执行Walk-Forward分析")
results['walk_forward'] = self.run_walk_forward_analysis()
# 4. 参数优化
self.logger.info("执行参数优化")
results['optimization'] = self.run_parameter_optimization()
# 5. 压力测试
self.logger.info("执行压力测试")
results['stress_test'] = self.run_stress_test()
# 6. 市场环境分析
self.logger.info("执行市场环境分析")
results['market_analysis'] = self.run_market_environment_analysis()
# 7. 生成综合报告
self.logger.info("生成回测报告")
self.generate_comprehensive_report(results)
return results
def run_standard_backtest(self) -> Dict[str, Any]:
"""运行标准回测"""
periods = self.config['backtest_periods']
results = {}
for period_name, period_config in periods.items():
self.logger.info(f"回测时期: {period_name}")
# 准备回测命令
cmd = self.build_backtest_command(period_config)
# 执行回测
result = self.execute_backtest(cmd, period_name)
# 分析结果
analysis = self.analyze_backtest_result(result, period_name)
results[period_name] = {
'raw_result': result,
'analysis': analysis,
'period': period_config
}
return results
def build_backtest_command(self, period_config: Dict[str, Any]) -> List[str]:
"""构建回测命令"""
cmd = [
'freqtrade', 'backtesting',
'--config', self.config['strategy_config_path'],
'--strategy', 'Strategy21VATSM',
'--timeframe', period_config.get('timeframe', '5m'),
'--timerange', f"{period_config['start_date']}-{period_config['end_date']}",
'--export', 'trades,signals',
'--export-filename', f"backtest_results/{period_config['name']}"
]
# 添加可选参数
if period_config.get('enable_protections'):
cmd.append('--enable-protections')
if period_config.get('breakdown'):
cmd.extend(['--breakdown', period_config['breakdown']])
return cmd
def execute_backtest(self, cmd: List[str], period_name: str) -> Dict[str, Any]:
"""执行单次回测"""
import subprocess
try:
result = subprocess.run(
cmd,
capture_output=True,
text=True,
timeout=3600 # 1小时超时
)
if result.returncode != 0:
self.logger.error(f"回测失败 {period_name}: {result.stderr}")
return {'success': False, 'error': result.stderr}
# 解析回测结果
output_path = f"backtest_results/{period_name}.json"
if Path(output_path).exists():
with open(output_path, 'r') as f:
backtest_data = json.load(f)
return {
'success': True,
'data': backtest_data,
'stdout': result.stdout
}
else:
return {'success': False, 'error': '回测结果文件未找到'}
except subprocess.TimeoutExpired:
self.logger.error(f"回测超时: {period_name}")
return {'success': False, 'error': '回测执行超时'}
except Exception as e:
self.logger.error(f"回测异常 {period_name}: {str(e)}")
return {'success': False, 'error': str(e)}性能分析器 (performance_analyzer.py)
class PerformanceAnalyzer:
"""策略性能分析器"""
def __init__(self):
self.metrics = {}
def analyze_backtest_result(self, result: Dict[str, Any],
period_name: str) -> Dict[str, Any]:
"""分析单次回测结果"""
if not result.get('success'):
return {'error': result.get('error', '未知错误')}
data = result['data']
strategy_stats = data.get('strategy', {})
trades = pd.DataFrame(data.get('trades', []))
analysis = {
'basic_stats': self.calculate_basic_stats(strategy_stats),
'risk_metrics': self.calculate_risk_metrics(strategy_stats, trades),
'trade_analysis': self.analyze_trades(trades),
'monthly_returns': self.calculate_monthly_returns(trades),
'drawdown_analysis': self.analyze_drawdowns(strategy_stats),
'market_correlation': self.analyze_market_correlation(trades)
}
return analysis
def calculate_basic_stats(self, stats: Dict[str, Any]) -> Dict[str, Any]:
"""计算基础统计指标"""
return {
'total_return_pct': stats.get('profit_total_pct', 0),
'annual_return_pct': stats.get('profit_total_pct_annual', 0),
'total_trades': stats.get('total_trades', 0),
'winning_trades': stats.get('wins', 0),
'losing_trades': stats.get('losses', 0),
'win_rate_pct': stats.get('winrate', 0) * 100,
'avg_profit_pct': stats.get('profit_mean_pct', 0),
'best_trade_pct': stats.get('best_pair_profit_ratio', 0) * 100,
'worst_trade_pct': stats.get('worst_pair_profit_ratio', 0) * 100,
'avg_trade_duration': stats.get('avg_duration', 'N/A'),
'market_change_pct': stats.get('market_change', 0)
}
def calculate_risk_metrics(self, stats: Dict[str, Any],
trades: pd.DataFrame) -> Dict[str, Any]:
"""计算风险指标"""
max_dd = abs(stats.get('max_drawdown', 0))
max_dd_pct = abs(stats.get('max_drawdown_pct', 0)) * 100
# 计算夏普比率
if not trades.empty and 'profit_ratio' in trades.columns:
returns = trades['profit_ratio'].dropna()
sharpe_ratio = self.calculate_sharpe_ratio(returns)
sortino_ratio = self.calculate_sortino_ratio(returns)
calmar_ratio = self.calculate_calmar_ratio(
stats.get('profit_total_pct_annual', 0), max_dd_pct
)
else:
sharpe_ratio = sortino_ratio = calmar_ratio = 0
return {
'max_drawdown_abs': max_dd,
'max_drawdown_pct': max_dd_pct,
'sharpe_ratio': sharpe_ratio,
'sortino_ratio': sortino_ratio,
'calmar_ratio': calmar_ratio,
'volatility_annual': self.calculate_volatility(trades),
'var_95': self.calculate_var(trades, 0.05),
'cvar_95': self.calculate_cvar(trades, 0.05)
}
def calculate_sharpe_ratio(self, returns: pd.Series,
risk_free_rate: float = 0.02) -> float:
"""计算夏普比率"""
if returns.std() == 0:
return 0
excess_returns = returns.mean() - (risk_free_rate / 252) # 日化无风险利率
return (excess_returns / returns.std()) * np.sqrt(252)
def calculate_sortino_ratio(self, returns: pd.Series,
risk_free_rate: float = 0.02) -> float:
"""计算索提诺比率"""
negative_returns = returns[returns < 0]
if len(negative_returns) == 0 or negative_returns.std() == 0:
return float('inf') if returns.mean() > 0 else 0
excess_returns = returns.mean() - (risk_free_rate / 252)
downside_deviation = negative_returns.std()
return (excess_returns / downside_deviation) * np.sqrt(252)
def calculate_calmar_ratio(self, annual_return: float,
max_drawdown_pct: float) -> float:
"""计算卡尔马比率"""
if max_drawdown_pct == 0:
return float('inf') if annual_return > 0 else 0
return annual_return / max_drawdown_pct
def analyze_trades(self, trades: pd.DataFrame) -> Dict[str, Any]:
"""分析交易明细"""
if trades.empty:
return {'error': '无交易数据'}
# 转换时间列
trades['open_date'] = pd.to_datetime(trades['open_date'])
trades['close_date'] = pd.to_datetime(trades['close_date'])
trades['duration'] = trades['close_date'] - trades['open_date']
winning_trades = trades[trades['profit_ratio'] > 0]
losing_trades = trades[trades['profit_ratio'] <= 0]
return {
'total_trades': len(trades),
'winning_trades': len(winning_trades),
'losing_trades': len(losing_trades),
'win_rate': len(winning_trades) / len(trades) if len(trades) > 0 else 0,
'avg_win_pct': winning_trades['profit_ratio'].mean() * 100 if not winning_trades.empty else 0,
'avg_loss_pct': losing_trades['profit_ratio'].mean() * 100 if not losing_trades.empty else 0,
'profit_factor': abs(winning_trades['profit_ratio'].sum() / losing_trades['profit_ratio'].sum()) if not losing_trades.empty and losing_trades['profit_ratio'].sum() != 0 else float('inf'),
'avg_duration_hours': trades['duration'].mean().total_seconds() / 3600,
'max_consecutive_wins': self.calculate_max_consecutive(trades['profit_ratio'] > 0),
'max_consecutive_losses': self.calculate_max_consecutive(trades['profit_ratio'] <= 0),
'monthly_trade_counts': self.get_monthly_trade_counts(trades)
}滚动回测分析 (rolling_backtest.py)
class RollingBacktestAnalyzer:
"""滚动回测分析器"""
def __init__(self, window_months: int = 6, step_months: int = 1):
self.window_months = window_months
self.step_months = step_months
def run_rolling_backtest(self, start_date: str, end_date: str) -> Dict[str, Any]:
"""执行滚动回测"""
start_dt = pd.to_datetime(start_date)
end_dt = pd.to_datetime(end_date)
results = []
current_start = start_dt
while current_start + pd.DateOffset(months=self.window_months) <= end_dt:
window_end = current_start + pd.DateOffset(months=self.window_months)
# 单个时间窗口回测
window_result = self.run_window_backtest(
current_start.strftime('%Y%m%d'),
window_end.strftime('%Y%m%d')
)
if window_result['success']:
results.append({
'start_date': current_start,
'end_date': window_end,
'result': window_result
})
current_start += pd.DateOffset(months=self.step_months)
return self.analyze_rolling_results(results)
def analyze_rolling_results(self, results: List[Dict[str, Any]]) -> Dict[str, Any]:
"""分析滚动回测结果"""
if not results:
return {'error': '无有效的滚动回测结果'}
returns = []
sharpe_ratios = []
max_drawdowns = []
win_rates = []
for result in results:
analysis = result['result']['analysis']
returns.append(analysis['basic_stats']['total_return_pct'])
sharpe_ratios.append(analysis['risk_metrics']['sharpe_ratio'])
max_drawdowns.append(analysis['risk_metrics']['max_drawdown_pct'])
win_rates.append(analysis['trade_analysis']['win_rate'])
return {
'total_windows': len(results),
'profitable_windows': sum(1 for r in returns if r > 0),
'profitability_rate': sum(1 for r in returns if r > 0) / len(returns),
'returns_stats': {
'mean': np.mean(returns),
'median': np.median(returns),
'std': np.std(returns),
'min': np.min(returns),
'max': np.max(returns),
'percentile_25': np.percentile(returns, 25),
'percentile_75': np.percentile(returns, 75)
},
'sharpe_stats': {
'mean': np.mean(sharpe_ratios),
'median': np.median(sharpe_ratios),
'std': np.std(sharpe_ratios),
'positive_sharpe_rate': sum(1 for s in sharpe_ratios if s > 0) / len(sharpe_ratios)
},
'drawdown_stats': {
'mean': np.mean(max_drawdowns),
'median': np.median(max_drawdowns),
'max': np.max(max_drawdowns),
'acceptable_dd_rate': sum(1 for dd in max_drawdowns if dd < 15) / len(max_drawdowns)
},
'consistency_metrics': {
'return_consistency': 1 - (np.std(returns) / abs(np.mean(returns))) if np.mean(returns) != 0 else 0,
'sharpe_consistency': 1 - (np.std(sharpe_ratios) / abs(np.mean(sharpe_ratios))) if np.mean(sharpe_ratios) != 0 else 0
}
}参数优化器 (parameter_optimizer.py)
from scipy.optimize import differential_evolution
from sklearn.model_selection import ParameterGrid
import itertools
class ParameterOptimizer:
"""策略参数优化器"""
def __init__(self, config: Dict[str, Any]):
self.config = config
self.optimization_results = []
def run_parameter_optimization(self) -> Dict[str, Any]:
"""运行参数优化"""
optimization_methods = self.config.get('optimization_methods', ['grid_search'])
results = {}
for method in optimization_methods:
if method == 'grid_search':
results['grid_search'] = self.grid_search_optimization()
elif method == 'random_search':
results['random_search'] = self.random_search_optimization()
elif method == 'genetic_algorithm':
results['genetic_algorithm'] = self.genetic_algorithm_optimization()
return results
def grid_search_optimization(self) -> Dict[str, Any]:
"""网格搜索优化"""
param_grid = self.config['optimization_ranges']
# 生成参数组合
param_combinations = self.generate_param_combinations(param_grid)
# 并行测试参数组合
with ProcessPoolExecutor(max_workers=4) as executor:
futures = []
for params in param_combinations[:50]: # 限制组合数量
future = executor.submit(self.test_parameter_combination, params)
futures.append((params, future))
results = []
for params, future in futures:
try:
result = future.result(timeout=1800) # 30分钟超时
result['parameters'] = params
results.append(result)
except Exception as e:
self.logger.error(f"参数组合测试失败 {params}: {str(e)}")
# 分析优化结果
return self.analyze_optimization_results(results, 'grid_search')
def test_parameter_combination(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""测试单个参数组合"""
# 创建临时配置文件
temp_config = self.create_temp_config(params)
# 执行回测
cmd = [
'freqtrade', 'backtesting',
'--config', temp_config,
'--strategy', 'Strategy21VATSM',
'--timerange', self.config['optimization_timerange'],
'--export', 'none' # 不导出详细数据以节省时间
]
result = self.execute_quick_backtest(cmd)
# 清理临时文件
Path(temp_config).unlink(missing_ok=True)
return result
def analyze_optimization_results(self, results: List[Dict[str, Any]],
method: str) -> Dict[str, Any]:
"""分析优化结果"""
if not results:
return {'error': f'无{method}优化结果'}
# 按不同目标函数排序
objectives = ['total_return', 'sharpe_ratio', 'calmar_ratio', 'profit_factor']
best_params = {}
for objective in objectives:
sorted_results = sorted(
results,
key=lambda x: x.get(objective, -float('inf')),
reverse=True
)
if sorted_results:
best_params[objective] = {
'parameters': sorted_results[0]['parameters'],
'performance': {
k: v for k, v in sorted_results[0].items()
if k != 'parameters'
}
}
return {
'method': method,
'total_combinations_tested': len(results),
'best_parameters': best_params,
'performance_distribution': self.analyze_performance_distribution(results),
'parameter_sensitivity': self.analyze_parameter_sensitivity(results)
}报告生成器 (report_generator.py)
class ReportGenerator:
"""回测报告生成器"""
def __init__(self, output_dir: str = "backtest_reports"):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(exist_ok=True)
def generate_comprehensive_report(self, results: Dict[str, Any]) -> str:
"""生成综合回测报告"""
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
report_path = self.output_dir / f"strategy21_backtest_report_{timestamp}.html"
# 生成HTML报告
html_content = self.generate_html_report(results, timestamp)
with open(report_path, 'w', encoding='utf-8') as f:
f.write(html_content)
# 生成图表
self.generate_charts(results, timestamp)
# 生成JSON摘要
self.generate_json_summary(results, timestamp)
return str(report_path)
def generate_html_report(self, results: Dict[str, Any], timestamp: str) -> str:
"""生成HTML格式报告"""
html = f"""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>Strategy-21 回测报告</title>
<style>
{self.get_report_css()}
</style>
</head>
<body>
<div class="container">
<h1>Strategy-21 VATSM 回测报告</h1>
<div class="timestamp">生成时间: {timestamp}</div>
<div class="section">
<h2>执行摘要</h2>
{self.generate_executive_summary(results)}
</div>
<div class="section">
<h2>标准回测结果</h2>
{self.generate_standard_backtest_section(results.get('standard', {}))}
</div>
<div class="section">
<h2>滚动回测分析</h2>
{self.generate_rolling_backtest_section(results.get('rolling', {}))}
</div>
<div class="section">
<h2>参数优化结果</h2>
{self.generate_optimization_section(results.get('optimization', {}))}
</div>
<div class="section">
<h2>风险分析</h2>
{self.generate_risk_analysis_section(results)}
</div>
<div class="section">
<h2>市场环境表现</h2>
{self.generate_market_analysis_section(results.get('market_analysis', {}))}
</div>
<div class="section">
<h2>结论与建议</h2>
{self.generate_conclusions(results)}
</div>
</div>
</body>
</html>
"""
return html
def generate_charts(self, results: Dict[str, Any], timestamp: str) -> None:
"""生成图表"""
# 设置中文字体
plt.rcParams['font.sans-serif'] = ['SimHei', 'DejaVu Sans']
plt.rcParams['axes.unicode_minus'] = False
# 1. 权益曲线图
self.plot_equity_curves(results, timestamp)
# 2. 回撤分析图
self.plot_drawdown_analysis(results, timestamp)
# 3. 月度收益热力图
self.plot_monthly_returns_heatmap(results, timestamp)
# 4. 滚动回测表现图
self.plot_rolling_performance(results, timestamp)
# 5. 参数敏感性分析图
self.plot_parameter_sensitivity(results, timestamp)配置文件 (backtest_config.json)
{
"strategy_config_path": "user_data/config_strategy21.json",
"data_path": "user_data/data",
"results_path": "backtest_results",
"backtest_periods": {
"full_period": {
"name": "full_period",
"start_date": "20230101",
"end_date": "20241201",
"timeframe": "5m",
"breakdown": "month",
"enable_protections": true
},
"bull_market": {
"name": "bull_market",
"start_date": "20230301",
"end_date": "20230801",
"timeframe": "5m"
},
"bear_market": {
"name": "bear_market",
"start_date": "20230801",
"end_date": "20231201",
"timeframe": "5m"
},
"recent_period": {
"name": "recent_period",
"start_date": "20240801",
"end_date": "20241201",
"timeframe": "5m"
}
},
"rolling_backtest": {
"window_months": 6,
"step_months": 1,
"start_date": "20230101",
"end_date": "20241201"
},
"walk_forward": {
"in_sample_months": 6,
"out_sample_months": 2,
"reoptimization_frequency": "monthly"
},
"optimization_methods": ["grid_search", "random_search"],
"optimization_timerange": "20230101-20240601",
"stress_tests": [
{
"name": "high_volatility",
"description": "高波动率时期测试",
"date_ranges": [
"20230308-20230323", # 银行危机
"20230801-20230830" # 市场调整
]
},
{
"name": "trend_changes",
"description": "趋势转换测试",
"date_ranges": [
"20230315-20230415",
"20230715-20230815"
]
}
]
}验收标准
功能测试
-
回测执行
- 标准回测正常完成并生成结果
- 滚动回测正确处理时间窗口
- 参数优化找到合理的参数组合
- 压力测试识别策略弱点
-
结果分析
- 性能指标计算准确
- 风险指标符合预期范围
- 市场环境分析有意义
- 交易统计数据一致
-
报告生成
- HTML报告格式正确美观
- 图表清晰表达关键信息
- JSON摘要包含核心数据
- 结论和建议具有指导意义
性能验证
-
策略表现目标
- 年化收益率 > 20%
- 最大回撤 < 15%
- 夏普比率 > 1.5
- 胜率 > 55%
- 盈亏比 > 1.2
-
稳定性指标
- 滚动回测盈利窗口 > 70%
- 不同市场环境均有正收益
- 参数优化结果稳定
- 压力测试回撤可控
实现计划
阶段1：核心框架 (2小时)
- 实现BacktestingFramework类
- 创建基础回测执行逻辑
- 设置日志和错误处理
阶段2：分析组件 (2小时)
- 实现PerformanceAnalyzer
- 创建RollingBacktestAnalyzer
- 实现ParameterOptimizer
阶段3：报告系统 (1.5小时)
- 实现ReportGenerator
- 创建HTML报告模板
- 实现图表生成功能
阶段4：集成测试 (0.5小时)
- 完整流程测试
- 性能指标验证
- 报告质量检查
依赖关系
- 前置依赖: 001, 002, 003, 004, 005 (所有策略模块)
- 数据依赖: 需要充足的历史数据
- 工具依赖: Freqtrade, pandas, matplotlib, seaborn
风险与缓解
技术风险
-
回测时间过长
- 风险：大量参数组合导致优化时间过长
- 缓解：并行处理，合理限制组合数量
-
数据质量问题
- 风险：历史数据缺失或错误
- 缓解：数据验证，多数据源交叉验证
业务风险
-
过拟合风险
- 风险：参数优化导致策略过度拟合历史数据
- 缓解：交叉验证，样本外测试
-
市场环境变化
- 风险：历史表现无法预测未来
- 缓解：多时期测试，持续监控
成功指标
- 自动化程度 - 一键完成完整验证流程
- 分析深度 - 多维度全面分析策略表现
- 报告质量 - 清晰展示关键信息和洞察
- 执行效率 - 回测流程在合理时间内完成
- 结果可信度 - 通过多种方法验证策略稳定性
交付物
-
核心代码
- backtesting_framework.py
- performance_analyzer.py
- rolling_backtest.py
- parameter_optimizer.py
- report_generator.py
-
配置文件
- backtest_config.json
- optimization_ranges.json
-
脚本文件
- run_full_backtest.py
- run_optimization.py
- generate_reports.py
-
报告模板
- HTML报告模板
- 图表样式配置
- JSON输出格式
-
文档
- 回测系统使用指南
- 性能指标说明文档
- 参数优化建议
- 结果解读指南