完整的回测流程和评估体系是量化交易的关键。
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
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()
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)
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)
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