320 lines
10 KiB
Python
320 lines
10 KiB
Python
"""
|
||
策略交付包导出器 — 生成策略文件 + Markdown 报告 + CSV 明细
|
||
|
||
交付包结构:
|
||
deliverables/{strategy_name}_{timestamp}/
|
||
├── {strategy_name}.py 策略源文件 (从 strategies/ 复制)
|
||
├── STRATEGY_REPORT.md 策略交付报告 (完整分析)
|
||
├── optimization.csv 参数优化响应面
|
||
├── walkforward.csv Walk-forward 逐窗口明细
|
||
└── backtest_metrics.csv 最优参数回测指标
|
||
|
||
用法:
|
||
from exporter import StrategyExporter
|
||
exporter = StrategyExporter()
|
||
path = exporter.deliver(
|
||
strategy_name="sma_cross",
|
||
opt_result=opt_result,
|
||
wf_result=wf_result,
|
||
accept_report=accept_report,
|
||
df=df,
|
||
symbol="XAUUSD",
|
||
)
|
||
print(f"交付包: {path}")
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
import shutil
|
||
from datetime import datetime
|
||
from typing import Optional
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
import raptorbt
|
||
|
||
from strategies import get_strategy
|
||
from strategies.base import Strategy
|
||
from .optimizer import OptimizationResult
|
||
from .walk_forward import WalkForwardResult
|
||
from .acceptance import AcceptanceReport
|
||
|
||
|
||
class StrategyExporter:
|
||
"""
|
||
策略交付包导出器
|
||
|
||
将策略文件 + 优化结果 + walk-forward 结果 + 验收报告
|
||
打包成一个完整的交付目录
|
||
"""
|
||
|
||
def __init__(self, output_dir: str = "deliverables"):
|
||
self.output_dir = os.path.join(
|
||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), output_dir
|
||
)
|
||
|
||
def deliver(
|
||
self,
|
||
strategy_name: str,
|
||
df: pd.DataFrame,
|
||
symbol: str,
|
||
opt_result: OptimizationResult,
|
||
wf_result: WalkForwardResult,
|
||
accept_report: AcceptanceReport,
|
||
param_grid: Optional[dict] = None,
|
||
data_info: Optional[dict] = None,
|
||
) -> str:
|
||
"""
|
||
生成完整交付包
|
||
|
||
参数:
|
||
strategy_name: 策略名称
|
||
df: K 线数据
|
||
symbol: 标的
|
||
opt_result: 优化结果
|
||
wf_result: walk-forward 结果
|
||
accept_report: 验收报告
|
||
param_grid: 参数搜索空间 (用于报告)
|
||
data_info: 数据信息 (如 {"source": "Mt5Bridge", "range": "2026-01~2026-06"})
|
||
|
||
返回:
|
||
交付包目录路径
|
||
"""
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
pkg_name = f"{strategy_name}_{timestamp}"
|
||
pkg_dir = os.path.join(self.output_dir, pkg_name)
|
||
os.makedirs(pkg_dir, exist_ok=True)
|
||
|
||
strategy = get_strategy(strategy_name)
|
||
|
||
# 1. 复制策略源文件
|
||
src_path = os.path.join(
|
||
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
|
||
"strategies", f"{strategy_name}.py",
|
||
)
|
||
if os.path.exists(src_path):
|
||
shutil.copy2(src_path, os.path.join(pkg_dir, f"{strategy_name}.py"))
|
||
|
||
# 2. 导出 CSV 明细
|
||
opt_result.export(os.path.join(pkg_dir, "optimization.csv"))
|
||
wf_result.export(os.path.join(pkg_dir, "walkforward.csv"))
|
||
|
||
# 3. 用最优参数跑完整回测, 导出指标
|
||
best_metrics = self._run_best_backtest(strategy_name, opt_result.best_params, df, symbol)
|
||
if best_metrics:
|
||
pd.DataFrame(list(best_metrics.items()), columns=["metric", "value"]).to_csv(
|
||
os.path.join(pkg_dir, "backtest_metrics.csv"),
|
||
index=False, encoding="utf-8-sig",
|
||
)
|
||
|
||
# 4. 生成 Markdown 报告
|
||
report = self._generate_report(
|
||
strategy_name=strategy_name,
|
||
strategy=strategy,
|
||
symbol=symbol,
|
||
opt_result=opt_result,
|
||
wf_result=wf_result,
|
||
accept_report=accept_report,
|
||
param_grid=param_grid,
|
||
data_info=data_info,
|
||
best_metrics=best_metrics,
|
||
)
|
||
report_path = os.path.join(pkg_dir, "STRATEGY_REPORT.md")
|
||
with open(report_path, "w", encoding="utf-8") as f:
|
||
f.write(report)
|
||
|
||
return pkg_dir
|
||
|
||
def _run_best_backtest(self, strategy_name, best_params, df, symbol) -> Optional[dict]:
|
||
"""用最优参数跑完整回测, 返回指标字典"""
|
||
try:
|
||
strategy = get_strategy(strategy_name)
|
||
# 用最优参数重新实例化
|
||
strategy_cls = type(strategy)
|
||
strategy = strategy_cls(**best_params)
|
||
|
||
signals = strategy.generate_signals(df)
|
||
config = strategy.build_config()
|
||
arr = Strategy.to_arrays(df)
|
||
|
||
result = raptorbt.run_single_backtest(
|
||
timestamps=arr["timestamps"],
|
||
open=arr["open"], high=arr["high"], low=arr["low"], close=arr["close"],
|
||
volume=arr["volume"],
|
||
entries=signals.entries, exits=signals.exits,
|
||
direction=signals.direction, weight=1.0, symbol=symbol,
|
||
config=config,
|
||
)
|
||
m = result.metrics
|
||
return {
|
||
"total_return_pct": m.total_return_pct,
|
||
"sharpe_ratio": m.sharpe_ratio,
|
||
"max_drawdown_pct": m.max_drawdown_pct,
|
||
"total_trades": m.total_trades,
|
||
"win_rate_pct": m.win_rate_pct,
|
||
"profit_factor": m.profit_factor,
|
||
}
|
||
except Exception as e:
|
||
print(f" ⚠️ 最优参数回测失败: {e}")
|
||
return None
|
||
|
||
def _generate_report(
|
||
self,
|
||
strategy_name,
|
||
strategy,
|
||
symbol,
|
||
opt_result,
|
||
wf_result,
|
||
accept_report,
|
||
param_grid,
|
||
data_info,
|
||
best_metrics,
|
||
) -> str:
|
||
"""生成 Markdown 交付报告"""
|
||
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
params_str = ", ".join(f"{k}={v}" for k, v in opt_result.best_params.items())
|
||
grid_str = ", ".join(
|
||
f'{k}={v}' for k, v in (param_grid or {}).items()
|
||
) if param_grid else "N/A"
|
||
data_info = data_info or {}
|
||
|
||
# 验收结果表
|
||
accept_lines = []
|
||
for c in accept_report.criteria:
|
||
status = "✅" if c.passed else "❌"
|
||
accept_lines.append(
|
||
f"| {c.name} | {c.threshold:.2f} | {c.actual:.2f} | {status} |"
|
||
)
|
||
accept_table = "\n".join(accept_lines) or "| (无) | | | |"
|
||
|
||
# 参数稳定性
|
||
stability_lines = []
|
||
for param, dist in wf_result.param_stability.items():
|
||
dist_str = ", ".join(
|
||
f"{k}×{v}" for k, v in sorted(dist.items(), key=lambda x: -x[1])
|
||
)
|
||
stability_lines.append(f"- **{param}**: {dist_str}")
|
||
stability_str = "\n".join(stability_lines) or "- (无)"
|
||
|
||
# 最优参数回测指标
|
||
if best_metrics:
|
||
metrics_lines = [
|
||
f"| {k} | {v:.4f} |" for k, v in best_metrics.items()
|
||
]
|
||
metrics_table = "\n".join(metrics_lines)
|
||
else:
|
||
metrics_table = "| (回测失败) | |"
|
||
|
||
# Top 5 参数组合
|
||
top5 = opt_result.top_n(5)
|
||
if len(top5) > 0:
|
||
cols = opt_result.param_names + [opt_result.metric, "total_trades"]
|
||
cols = [c for c in cols if c in top5.columns]
|
||
top5_lines = ["| " + " | ".join(cols) + " |",
|
||
"|" + "|".join(["---"] * len(cols)) + "|"]
|
||
for _, row in top5.iterrows():
|
||
vals = []
|
||
for c in cols:
|
||
v = row[c]
|
||
if isinstance(v, float):
|
||
vals.append(f"{v:.4f}")
|
||
else:
|
||
vals.append(str(v))
|
||
top5_lines.append("| " + " | ".join(vals) + " |")
|
||
top5_table = "\n".join(top5_lines)
|
||
else:
|
||
top5_table = "(无有效结果)"
|
||
|
||
verdict = "✅ 通过验收，可交付" if accept_report.passed else "❌ 未通过验收，不可交付"
|
||
|
||
return f"""# 策略交付报告: {strategy_name}
|
||
|
||
> 生成时间: {now}
|
||
> 引擎: RaptorBT v{raptorbt.__version__}
|
||
|
||
## 1. 策略概述
|
||
|
||
| 项目 | 内容 |
|
||
|------|------|
|
||
| 策略名称 | `{strategy_name}` |
|
||
| 描述 | {strategy.description()} |
|
||
| 交易标的 | {symbol} |
|
||
| 最优参数 | {params_str} |
|
||
| 参数搜索空间 | {grid_str} |
|
||
| 预热期 | {strategy.warmup_bars()} bars |
|
||
|
||
## 2. 参数优化
|
||
|
||
**目标指标**: {opt_result.metric} ({opt_result.direction})
|
||
**搜索组合数**: {len(opt_result.results)}
|
||
**有效结果数**: {opt_result.results[opt_result.metric].notna().sum()}
|
||
|
||
### Top 5 参数组合
|
||
|
||
{top5_table}
|
||
|
||
## 3. Walk-Forward 验证
|
||
|
||
| 指标 | IS (训练集) | OOS (测试集) |
|
||
|------|------------|-------------|
|
||
| 平均夏普比率 | {wf_result.is_sharpe_avg:.4f} | {wf_result.oos_sharpe_avg:.4f} |
|
||
| 平均收益率 | - | {wf_result.oos_return_avg:.2f}% |
|
||
| 平均最大回撤 | - | {wf_result.oos_max_drawdown_avg:.2f}% |
|
||
| 总交易数 | - | {wf_result.oos_trades_total} |
|
||
| 平均胜率 | - | {wf_result.oos_win_rate_avg:.1f}% |
|
||
|
||
**衰减比 (OOS/IS)**: {wf_result.decay_ratio:.2%}
|
||
**过拟合判定**: {'⚠️ 是 (衰减比 < 50%)' if wf_result.is_overfit else '✅ 否'}
|
||
**验证窗口数**: {wf_result.n_windows}
|
||
|
||
### 参数稳定性
|
||
|
||
{stability_str}
|
||
|
||
## 4. 验收结果
|
||
|
||
**总体结论**: {verdict}
|
||
|
||
| 标准 | 阈值 | 实际值 | 结果 |
|
||
|------|------|--------|------|
|
||
{accept_table}
|
||
|
||
## 5. 最优参数完整回测
|
||
|
||
| 指标 | 值 |
|
||
|------|-----|
|
||
{metrics_table}
|
||
|
||
## 6. 数据信息
|
||
|
||
| 项目 | 内容 |
|
||
|------|------|
|
||
| 数据源 | {data_info.get('source', 'Mt5Bridge')} |
|
||
| 数据范围 | {data_info.get('range', 'N/A')} |
|
||
| K 线数量 | {data_info.get('bars', 'N/A')} |
|
||
| 时间周期 | {data_info.get('timeframe', 'N/A')} |
|
||
|
||
## 7. 风险提示
|
||
|
||
1. **回测不等于实盘**: 历史表现不保证未来收益，滑点和延迟可能影响实际执行
|
||
2. **参数敏感性**: 请关注参数稳定性分析，参数值频繁变化的策略稳健性较差
|
||
3. **市场环境**: 策略可能在特定市场条件下表现优异，其他条件下表现不佳
|
||
4. **过拟合风险**: {'⚠️ 检测到过拟合，OOS 表现显著低于 IS' if wf_result.is_overfit else '未检测到明显过拟合'}
|
||
5. **交易成本**: 回测已包含手续费和滑点，但实际成本可能更高
|
||
|
||
## 8. 复现方式
|
||
|
||
```bash
|
||
# 使用最优参数回测
|
||
python main.py run --strategy {strategy_name} --symbol {symbol} --export
|
||
|
||
# 重新验证
|
||
python main.py validate --strategy {strategy_name} --param <参数>=<值>
|
||
```
|
||
|
||
---
|
||
|
||
*本报告由 RaptorBT 策略自动化框架自动生成*
|
||
"""
|