Initial commit: MT5 EA 回测报告分析工具集
通用可复用的 MT5 回测分析工具，不针对任何具体 EA： - mt5_report_parser.py: 解析 MT5 中文 xlsx 报告，重建逐笔交易 - run_analysis.py: 主分析器，IS/OSS 对比 + What-If + 蒙特卡洛 + 数据驱动建议 - walk_forward.py: Walk-Forward 滚动 IS→OOS 验证 (WFE/泛化相关性) - param_scan.py: 参数敏感度扫描，高原检测 + 输出优化后 .set - mae_mfe.py: MAE/MFE 分析 + TP/SL 扫描，附 MQL5 导出代码 所有输出为自包含 HTML (内联 SVG，无图片依赖)。
This commit is contained in:
+51
@@ -0,0 +1,51 @@
|
||||
# Python
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
*.so
|
||||
.Python
|
||||
*.egg-info/
|
||||
.eggs/
|
||||
build/
|
||||
dist/
|
||||
.pytest_cache/
|
||||
.mypy_cache/
|
||||
.ruff_cache/
|
||||
.venv/
|
||||
venv/
|
||||
env/
|
||||
|
||||
# IDE / 编辑器
|
||||
.vscode/
|
||||
.idea/
|
||||
*.swp
|
||||
*.swo
|
||||
*~
|
||||
.DS_Store
|
||||
Thumbs.db
|
||||
|
||||
# MT5 回测原始报告（含交易数据，敏感，不入库）
|
||||
# 命名约定：IS-*.xlsx / OSS-*.xlsx / ReportTester-*.xlsx
|
||||
IS-*.xlsx
|
||||
OSS-*.xlsx
|
||||
ReportTester-*.xlsx
|
||||
ReportTester-*.htm
|
||||
ReportTester-*.xml
|
||||
|
||||
# 参数扫描中间产物（批量回测的 .set 与报告，可由 gen-set 重新生成）
|
||||
scan_*/
|
||||
*.set.bak
|
||||
|
||||
# MAE/MFE 原始 CSV（由 EA 导出，含逐笔数据）
|
||||
mae_mfe_*.csv
|
||||
|
||||
# 本地配置 / 凭证
|
||||
.env
|
||||
.env.local
|
||||
*.secret
|
||||
config_local.ini
|
||||
|
||||
# 临时与日志
|
||||
*.log
|
||||
tmp/
|
||||
temp/
|
||||
@@ -0,0 +1,180 @@
|
||||
# MT5 EA 回测报告分析工具集
|
||||
|
||||
一套**通用、可复用**的 MetaTrader 5 EA 回测报告分析工具。读取 MT5 Strategy Tester 导出的 xlsx 报告，输出自包含 HTML（内联 SVG，无图片依赖，AI 可解析表格、人类可读可视化），辅助 EA 优化决策。
|
||||
|
||||
不针对任何具体 EA —— 所有分析逻辑数据驱动，换一份报告即自动重新计算与诊断。
|
||||
|
||||
## 功能概览
|
||||
|
||||
| 模块 | 作用 | 输出 |
|
||||
|---|---|---|
|
||||
| `mt5_report_parser.py` | 解析 MT5 中文 xlsx 报告（设置/结果/订单/成交），重建逐笔交易 | `MT5Report` 对象 |
|
||||
| `run_analysis.py` | 主分析器：核心指标对比、What-If 假设、蒙特卡洛、方向/时段诊断、数据驱动建议 | `report.html` |
|
||||
| `walk_forward.py` | Walk-Forward 滚动 IS→OOS 验证，算 WFE 与泛化相关性 | `walk_forward.html` |
|
||||
| `param_scan.py` | 参数敏感度扫描：网格 .set 生成 → 响应面 → 高原检测 → **输出优化后 .set** | `param_scan.html` + `.set` |
|
||||
| `mae_mfe.py` | MAE/MFE 分析：散点 + TP/SL 扫描热力图，附 MQL5 导出代码 | `mae_mfe.html` |
|
||||
|
||||
## 快速开始
|
||||
|
||||
### 环境要求
|
||||
- Python 3.9+
|
||||
- 依赖：`openpyxl`、`pandas`、`numpy`
|
||||
|
||||
```bash
|
||||
pip install openpyxl pandas numpy
|
||||
```
|
||||
|
||||
### 基本用法
|
||||
|
||||
把两份 MT5 回测报告（通常 IS / OSS 各一份）放进目录，命名为 `IS-*.xlsx` 和 `OSS-*.xlsx`：
|
||||
|
||||
```bash
|
||||
python run_analysis.py
|
||||
# 或显式指定：
|
||||
python run_analysis.py IS-ReportTester-12345.xlsx OSS-ReportTester-12345.xlsx
|
||||
```
|
||||
|
||||
生成的 `output/report.html` 可直接浏览器打开。
|
||||
|
||||
### 各模块独立使用
|
||||
|
||||
```bash
|
||||
# Walk-Forward 滚动验证
|
||||
python walk_forward.py IS-ReportTester-12345.xlsx --is-days 60 --oos-days 30
|
||||
|
||||
# 参数敏感度扫描（三步）
|
||||
python param_scan.py gen-set template.set grid.json out_setdir --ea MyEA
|
||||
# （在 MT5 里跑 out_setdir/run_scan.bat，把报告放到 out_setdir/reports/）
|
||||
python param_scan.py analyze out_setdir/reports out_setdir/manifest.csv \
|
||||
--x FastMA --y SlowMA --metric pf \
|
||||
--template template.set --out-set output/optimized.set --strategy plateau
|
||||
|
||||
# MAE/MFE 分析
|
||||
python mae_mfe.py --show-snippet > mae_mfe_snippet.mqh # 取 MQL5 代码
|
||||
# （把代码粘进 EA，回测后得到 mae_mfe_*.csv）
|
||||
python mae_mfe.py mae_mfe_2026-06-28.csv
|
||||
```
|
||||
|
||||
## 项目结构
|
||||
|
||||
```
|
||||
.
|
||||
├── mt5_report_parser.py # 通用 MT5 xlsx 解析器
|
||||
├── run_analysis.py # 主分析器（含 What-If/蒙特卡洛/规则建议/Walk-Forward 集成）
|
||||
├── walk_forward.py # Walk-Forward 滚动验证
|
||||
├── param_scan.py # 参数敏感度扫描 + 优化 .set 输出
|
||||
├── mae_mfe.py # MAE/MFE 分析 + MQL5 代码片段
|
||||
├── output/ # 生成产物（HTML/.set，已入库便于在线预览）
|
||||
│ ├── report.html
|
||||
│ ├── walk_forward.html
|
||||
│ ├── param_scan.html
|
||||
│ └── mae_mfe.html
|
||||
├── .gitignore
|
||||
└── README.md
|
||||
```
|
||||
|
||||
## 各模块详解
|
||||
|
||||
### 1. mt5_report_parser.py — 解析器
|
||||
|
||||
解析 MT5 Strategy Tester 导出的中文 xlsx 报告，结构化为：
|
||||
|
||||
- `meta`：元信息（EA、品种、期间、初始资金、杠杆、输入参数字典）
|
||||
- `summary` / `summary_norm`：结果区汇总指标（数值化后的字典）
|
||||
- `orders` / `deals`：订单与成交明细 DataFrame
|
||||
- `trades`：由 in/out 成交对**重建的逐笔交易** DataFrame（含 open_time、close_time、direction、net_profit、duration_min 等）
|
||||
|
||||
```python
|
||||
import mt5_report_parser as mp
|
||||
rep = mp.parse_report("IS-ReportTester-12345.xlsx")
|
||||
print(rep.meta["专家"], rep.summary["总净盈利"])
|
||||
print(rep.trades[["open_time","direction","net_profit"]].head())
|
||||
```
|
||||
|
||||
`compute_trade_metrics(trades)` 一次算出：PF、胜率、Sortino、Calmar、滚动PF、最大连败、按方向/小时/星期/持仓时间分桶等扩展指标。
|
||||
|
||||
### 2. run_analysis.py — 主分析器
|
||||
|
||||
读两份报告，输出 `report.html`，包含 10 个章节：
|
||||
|
||||
1. 核心指标对比（PF/胜率/回撤/夏普/Sortino/Calmar/滚动PF）
|
||||
2. 资金曲线（内联 SVG）
|
||||
3. 方向性诊断（多空各自的胜率/PF/盈亏比/期望）
|
||||
4. **What-If 假设分析**（9 个通用场景：sell-only/buy-only/信号反向/过滤最差时段/仓位减半/盈利单放大/亏损截断/启用BE）
|
||||
5. 蒙特卡洛回撤模拟（1000 次打乱，检验顺序自相关）
|
||||
6. 时段诊断（小时/星期热力表，标 IS/OSS 双负窗口）
|
||||
7. 持仓时间分桶
|
||||
8. **数据驱动的优化建议**（8 条规则按数据特征自动触发，每条附"触发条件→数据→动作"）
|
||||
9. Walk-Forward 滚动验证（内嵌）
|
||||
10. 可选扩展分析指引
|
||||
|
||||
**关键设计**：建议规则不预设任何策略特定结论。例如"信号方向自检"规则仅当两份报告的"信号反向"场景净盈利均转正且 PF>1 时才触发，提示用户复核源码方向逻辑。
|
||||
|
||||
### 3. walk_forward.py — Walk-Forward 验证
|
||||
|
||||
将单份报告按时间切成多个滚动窗口（IS段 + OOS段），逐窗算指标：
|
||||
|
||||
- **WFE**（Walk-Forward Efficiency）= ΣOOS净盈利 / ΣIS净盈利
|
||||
- **IS-OOS PF 相关性**：高=参数泛化好，低=过拟合风险
|
||||
- OOS 盈利窗口占比
|
||||
- 滚动 IS/OOS PF 曲线 SVG
|
||||
|
||||
可独立运行，也已集成进主报告第 9 节。
|
||||
|
||||
### 4. param_scan.py — 参数敏感度扫描
|
||||
|
||||
三个子命令：
|
||||
|
||||
- `gen-set`：读 `grid.json` + .set 模板，批量生成每个参数组合的 .set + manifest.csv + MT5 批处理脚本 + tester.ini 模板
|
||||
- `analyze`：读 manifest + 一批报告，构建响应面
|
||||
- `demo`：合成数据演示
|
||||
|
||||
**响应面分析**：
|
||||
- 2D 热力图（参数X × 参数Y → 指标，标出高原 P1/P2/P3）
|
||||
- 3D 等距投影曲面
|
||||
- **高原检测**：4-连通区域 ≥4 格且高于 80 分位 → 稳健参数区
|
||||
- **过拟合评分**：峰值孤立度 + 高原覆盖率 → 稳健性 0~1
|
||||
- **直接输出优化后 .set**：支持两种策略
|
||||
- `--strategy plateau`（默认，推荐）：取最大高原中心，抗扰动
|
||||
- `--strategy peak`：取单点峰值，激进易过拟合
|
||||
|
||||
`grid.json` 示例：
|
||||
```json
|
||||
{"FastMA": [5,10,15,20,25], "SlowMA": [20,30,40,50], "StopLoss": [50,100]}
|
||||
```
|
||||
|
||||
### 5. mae_mfe.py — MAE/MFE 分析
|
||||
|
||||
MT5 标准 xlsx 不含逐笔 MAE/MFE。本模块提供：
|
||||
|
||||
- `--show-snippet`：输出 MQL5 代码，粘进 EA 的 OnDeinit，自动扫描每笔持仓期的 iHigh/iLow 算 MAE/MFE 导出 CSV
|
||||
- 分析 CSV：MAE/MFE vs 盈亏散点（绿赢红输 + SL/TP 候选阈值线）
|
||||
- TP×SL 网格扫描热力图：扫描所有组合重算理论净盈利，标最优点
|
||||
- 推荐：SL = 亏损笔 MAE 75 分位；TP = 盈利笔 MFE 中位
|
||||
|
||||
## 设计原则
|
||||
|
||||
1. **通用可复用**：脚本零硬编码，不依赖任何具体 EA 的参数名/阈值/结论
|
||||
2. **数据驱动**：所有"建议"由当前数据的特征触发，非预写
|
||||
3. **自包含输出**：HTML 全部用纯表格 + 内联 SVG，无外部图片/JS/CSS 依赖，AI 可直接解析数据
|
||||
4. **输入仅依赖** mt5_report_parser 解析出的结构化数据
|
||||
|
||||
## 关于示例数据
|
||||
|
||||
出于交易策略数据敏感性，示例回测报告（`IS-*.xlsx` / `OSS-*.xlsx`）已通过 `.gitignore` 排除。`output/` 下的 HTML 为真实数据生成的示例产物，可供预览效果。
|
||||
|
||||
如需本地试跑，将自己的 MT5 报告按 `IS-*.xlsx` / `OSS-*.xlsx` 命名放入根目录即可。
|
||||
|
||||
## 依赖
|
||||
|
||||
| 包 | 用途 |
|
||||
|---|---|
|
||||
| `openpyxl` | 读取 xlsx |
|
||||
| `pandas` | 数据处理 |
|
||||
| `numpy` | 数值计算 |
|
||||
|
||||
无 matplotlib 等重型绘图库，所有可视化用纯 SVG。
|
||||
|
||||
## License
|
||||
|
||||
MIT
|
||||
+441
@@ -0,0 +1,441 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
MAE / MFE 分析
|
||||
==============
|
||||
MT5 标准回测 xlsx 不含逐笔 MAE/MFE（最大不利/有利偏移），需要由 EA 在
|
||||
OnDeinit 里把每笔交易的 MAE/MFE 写到 CSV。本模块：
|
||||
|
||||
1. mql5_snippet() : 返回可粘贴进 EA 的 MQL5 代码，自动导出 mae_mfe.csv
|
||||
2. analyze(df) : 由 CSV 计算最优 TP/SL 区间
|
||||
3. build_html(df) : 散点图 + 推荐区间
|
||||
|
||||
CSV 列约定（首行表头）:
|
||||
ticket,direction,open_time,close_time,open_price,close_price,
|
||||
mae_points,mfe_points,profit_money
|
||||
|
||||
direction: long / short
|
||||
mae_points: 持仓期间最大不利偏移（点，正数表示逆向走了多少点）
|
||||
mfe_points: 持仓期间最大有利偏移（点，正数）
|
||||
profit_money: 该笔最终盈亏（金额）
|
||||
|
||||
分析原理:
|
||||
- MFE vs 盈亏散点：能涨到 X 点的笔里，最终盈利占比？→ 若大量高 MFE 笔最终
|
||||
亏损，说明 TP 太贪；把 TP 放到 MFE 高胜率区间可改善。
|
||||
- MAE vs 盈亏散点：逆向超过 Y 点的笔几乎全亏 → SL 设在 Y 内可避免大损。
|
||||
- 用扫描法找使"理论净盈利"最大的 TP/SL 组合（基于历史 MAE/MFE 重算）。
|
||||
|
||||
用法:
|
||||
python mae_mfe.py <mae_mfe.csv>
|
||||
python mae_mfe.py --show-snippet # 打印 MQL5 代码
|
||||
python mae_mfe.py --demo # 合成数据演示
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import html
|
||||
import os
|
||||
import sys
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
OUT_DIR = "output"
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# 1. MQL5 代码片段
|
||||
# =========================================================================== #
|
||||
def mql5_snippet() -> str:
|
||||
"""返回 MQL5 代码：在 OnDeinit 里把每笔历史仓位的 MAE/MFE 写入 CSV。"""
|
||||
return r'''// ====== MAE/MFE 日志片段（粘贴进 EA） ======
|
||||
// 在文件顶部加:
|
||||
#include <Trade\DealInfo.mqh>
|
||||
#include <Trade\PositionInfo.mqh>
|
||||
|
||||
// 在 OnDeinit(const int reason) 末尾加:
|
||||
void ExportMAEMFE()
|
||||
{
|
||||
string fname = "mae_mfe_" + TimeToString(TimeCurrent(), TIME_DATE) + ".csv";
|
||||
int h = FileOpen(fname, FILE_WRITE | FILE_CSV | FILE_ANSI, ',');
|
||||
if(h == INVALID_HANDLE) { Print("MAE/MFE: 文件打开失败"); return; }
|
||||
FileWrite(h, "ticket","direction","open_time","close_time","open_price","close_price",
|
||||
"mae_points","mfe_points","profit_money");
|
||||
|
||||
HistorySelect(0, TimeCurrent() + 86400);
|
||||
int total = HistoryDealsTotal();
|
||||
// 按订单(order)配对 in/out
|
||||
for(int i = 0; i < total; i++)
|
||||
{
|
||||
ulong dealIn = HistoryDealGetTicket(i);
|
||||
if(HistoryDealGetInteger(dealIn, DEAL_ENTRY) != DEAL_ENTRY_IN) continue;
|
||||
|
||||
long order = HistoryDealGetInteger(dealIn, DEAL_ORDER);
|
||||
ulong dealOut = 0;
|
||||
// 找同 order 的 out 成交
|
||||
for(int j = i + 1; j < total; j++)
|
||||
{
|
||||
ulong d = HistoryDealGetTicket(j);
|
||||
if(HistoryDealGetInteger(d, DEAL_ORDER) == order &&
|
||||
HistoryDealGetInteger(d, DEAL_ENTRY) == DEAL_ENTRY_OUT)
|
||||
{ dealOut = d; break; }
|
||||
}
|
||||
if(dealOut == 0) continue;
|
||||
|
||||
datetime tIn = (datetime)HistoryDealGetInteger(dealIn, DEAL_TIME);
|
||||
datetime tOut = (datetime)HistoryDealGetInteger(dealOut, DEAL_TIME);
|
||||
double pIn = HistoryDealGetDouble(dealIn, DEAL_PRICE);
|
||||
double pOut = HistoryDealGetDouble(dealOut, DEAL_PRICE);
|
||||
long type = HistoryDealGetInteger(dealIn, DEAL_TYPE); // DEAL_TYPE_BUY / SELL
|
||||
double profit = HistoryDealGetDouble(dealOut, DEAL_PROFIT)
|
||||
+ HistoryDealGetDouble(dealOut, DEAL_SWAP)
|
||||
+ HistoryDealGetDouble(dealOut, DEAL_COMMISSION);
|
||||
string dir = (type == DEAL_TYPE_BUY) ? "long" : "short";
|
||||
|
||||
// 扫描持仓期间最高/最低价算 MAE/MFE（点）
|
||||
double mfe = 0, mae = 0;
|
||||
double pt = SymbolInfoDouble(_Symbol, SYMBOL_POINT);
|
||||
for(datetime t = tIn; t <= tOut; t += PeriodSeconds())
|
||||
{
|
||||
double hi = iHigh(_Symbol, PERIOD_M1, iBarShift(_Symbol, PERIOD_M1, t, false));
|
||||
double lo = iLow (_Symbol, PERIOD_M1, iBarShift(_Symbol, PERIOD_M1, t, false));
|
||||
if(hi <= 0 || lo <= 0) continue;
|
||||
if(type == DEAL_TYPE_BUY)
|
||||
{
|
||||
double fav = (hi - pIn) / pt; // 有利
|
||||
double adv = (pIn - lo) / pt; // 不利
|
||||
if(fav > mfe) mfe = fav;
|
||||
if(adv > mae) mae = adv;
|
||||
}
|
||||
else
|
||||
{
|
||||
double fav = (pIn - lo) / pt;
|
||||
double adv = (hi - pIn) / pt;
|
||||
if(fav > mfe) mfe = fav;
|
||||
if(adv > mae) mae = adv;
|
||||
}
|
||||
}
|
||||
FileWrite(h, order, dir, TimeToString(tIn), TimeToString(tOut),
|
||||
DoubleToString(pIn, _Digits), DoubleToString(pOut, _Digits),
|
||||
DoubleToString(mae, 1), DoubleToString(mfe, 1),
|
||||
DoubleToString(profit, 2));
|
||||
}
|
||||
FileClose(h);
|
||||
Print("MAE/MFE 已导出: ", fname);
|
||||
}
|
||||
// 在 OnDeinit 里调用: ExportMAEMFE();
|
||||
// ====== 片段结束 ======
|
||||
'''
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# 2. 加载
|
||||
# =========================================================================== #
|
||||
def load_csv(path: str) -> pd.DataFrame:
|
||||
df = pd.read_csv(path)
|
||||
# 列名容错
|
||||
rename = {
|
||||
"mae_points": "mae", "mfe_points": "mfe",
|
||||
"profit_money": "profit",
|
||||
"open_price": "open", "close_price": "close",
|
||||
}
|
||||
df = df.rename(columns={k: v for k, v in rename.items() if k in df.columns})
|
||||
for c in ("mae", "mfe", "profit"):
|
||||
if c in df.columns:
|
||||
df[c] = pd.to_numeric(df[c], errors="coerce")
|
||||
return df
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# 3. 分析：最优 TP/SL
|
||||
# =========================================================================== #
|
||||
def analyze(df: pd.DataFrame, tp_grid: Optional[List[float]] = None,
|
||||
sl_grid: Optional[List[float]] = None) -> Dict[str, Any]:
|
||||
"""
|
||||
扫描 TP/SL 组合，找使理论净盈利最大的参数。
|
||||
理论模型：对每笔，若 mfe >= TP 则按 TP 离场（盈利 = TP×点值）；
|
||||
若 mae >= SL 则按 SL 离场（亏损 = -SL×点值）；
|
||||
否则按原 profit。
|
||||
这里用"点数口径"近似：盈利/亏损点数替代金额（点值统一假设）。
|
||||
"""
|
||||
if df.empty:
|
||||
return {}
|
||||
if tp_grid is None:
|
||||
mfe_max = float(df["mfe"].max())
|
||||
tp_grid = list(np.linspace(0, mfe_max, 30))
|
||||
if sl_grid is None:
|
||||
mae_max = float(df["mae"].max())
|
||||
sl_grid = list(np.linspace(0, mae_max, 30))
|
||||
|
||||
mae = df["mae"].to_numpy()
|
||||
mfe = df["mfe"].to_numpy()
|
||||
n = len(df)
|
||||
|
||||
best = None
|
||||
grid = []
|
||||
for sl in sl_grid:
|
||||
for tp in tp_grid:
|
||||
# 重算每笔点数结果
|
||||
# 优先 SL 触发（更保守：先假设逆向先到）
|
||||
result_pts = np.empty(n)
|
||||
for k in range(n):
|
||||
if mae[k] >= sl and sl > 0:
|
||||
result_pts[k] = -sl
|
||||
elif mfe[k] >= tp and tp > 0:
|
||||
result_pts[k] = tp
|
||||
else:
|
||||
# 都没触发，按原方向小盈小亏（用 profit 符号 × mfe 比例近似）
|
||||
result_pts[k] = (mfe[k] - mae[k]) * 0.5
|
||||
net = float(result_pts.sum())
|
||||
wins = int((result_pts > 0).sum())
|
||||
pf = float(result_pts[result_pts > 0].sum() / -result_pts[result_pts <= 0].sum()) if (result_pts <= 0).any() else np.inf
|
||||
grid.append({"tp": float(tp), "sl": float(sl), "net": net,
|
||||
"win_rate": wins / n * 100, "pf": pf})
|
||||
if best is None or net > best["net"]:
|
||||
best = {"tp": float(tp), "sl": float(sl), "net": net,
|
||||
"win_rate": wins / n * 100, "pf": pf}
|
||||
grid_df = pd.DataFrame(grid)
|
||||
|
||||
# MAE/MFE 分布统计
|
||||
mae_threshold = float(df.loc[df["profit"] <= 0, "mae"].quantile(0.75)) if (df["profit"] <= 0).any() else 0
|
||||
mfe_threshold = float(df.loc[df["profit"] > 0, "mfe"].quantile(0.5)) if (df["profit"] > 0).any() else 0
|
||||
|
||||
return {
|
||||
"best": best,
|
||||
"grid": grid_df,
|
||||
"mae_q3_of_losers": mae_threshold, # 75% 亏损笔的 MAE 上限 → SL 候选
|
||||
"mfe_median_of_winners": mfe_threshold, # 盈利笔 MFE 中位 → TP 候选
|
||||
"n": n,
|
||||
"profit_corr_mae": float(df["profit"].corr(df["mae"])),
|
||||
"profit_corr_mfe": float(df["profit"].corr(df["mfe"])),
|
||||
}
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# 4. SVG 散点
|
||||
# =========================================================================== #
|
||||
def svg_scatter(x: np.ndarray, y: np.ndarray, profit: np.ndarray,
|
||||
xlabel: str, ylabel: str, title: str,
|
||||
threshold: Optional[float] = None, threshold_label: str = "",
|
||||
w: int = 460, h: int = 320) -> str:
|
||||
top, bottom, left, right = 30, 40, 50, 16
|
||||
plot_w = w - left - right
|
||||
plot_h = h - top - bottom
|
||||
x_min, x_max = float(x.min()), float(x.max())
|
||||
y_min, y_max = float(y.min()), float(y.max())
|
||||
if x_max == x_min: x_max = x_min + 1
|
||||
if y_max == y_min: y_max = y_min + 1
|
||||
|
||||
def sx(v): return left + (v - x_min) / (x_max - x_min) * plot_w
|
||||
def sy(v): return top + plot_h - (v - y_min) / (y_max - y_min) * plot_h
|
||||
|
||||
pts = ""
|
||||
for xi, yi, pi in zip(x, y, profit):
|
||||
col = "#16a34a" if pi > 0 else "#dc2626"
|
||||
pts += f'<circle cx="{sx(xi):.1f}" cy="{sy(yi):.1f}" r="2.2" fill="{col}" opacity="0.45"><title>{xlabel}={xi:.1f}, {ylabel}={yi:.2f}, profit={pi:.2f}</title></circle>'
|
||||
|
||||
thr = ""
|
||||
if threshold is not None and x_min <= threshold <= x_max:
|
||||
tx = sx(threshold)
|
||||
thr = (f'<line x1="{tx:.1f}" y1="{top}" x2="{tx:.1f}" y2="{top+plot_h}" stroke="#1e3a8a" stroke-width="1.5" stroke-dasharray="4,3"/>'
|
||||
f'<text x="{tx+4:.1f}" y="{top+12:.1f}" font-size="9" fill="#1e3a8a">{html.escape(threshold_label)}={threshold:.1f}</text>')
|
||||
|
||||
grid = ""
|
||||
for frac in (0, 0.5, 1.0):
|
||||
xv = x_min + frac * (x_max - x_min)
|
||||
yv = y_min + frac * (y_max - y_min)
|
||||
gx = left + frac * plot_w
|
||||
gy = top + plot_h - frac * plot_h
|
||||
grid += (f'<line x1="{gx:.1f}" y1="{top}" x2="{gx:.1f}" y2="{top+plot_h}" stroke="#f1f5f9" stroke-width="0.5"/>'
|
||||
f'<text x="{gx:.1f}" y="{h-bottom+14:.1f}" font-size="9" fill="#6b7280" text-anchor="middle">{xv:.0f}</text>'
|
||||
f'<line x1="{left}" y1="{gy:.1f}" x2="{w-right}" y2="{gy:.1f}" stroke="#f1f5f9" stroke-width="0.5"/>'
|
||||
f'<text x="{left-6:.1f}" y="{gy+3:.1f}" font-size="9" fill="#6b7280" text-anchor="end">{yv:.0f}</text>')
|
||||
|
||||
return (f'<svg viewBox="0 0 {w} {h}" class="chart">'
|
||||
f'<text x="{w//2}" y="16" font-size="11" fill="#374151" text-anchor="middle">{html.escape(title)}</text>'
|
||||
f'{grid}{pts}{thr}'
|
||||
f'<text x="{w//2}" y="{h-4}" font-size="9" fill="#6b7280" text-anchor="middle">{html.escape(xlabel)} →</text>'
|
||||
f'<text x="14" y="{h//2}" font-size="9" fill="#6b7280" text-anchor="middle" transform="rotate(-90 14 {h//2})">{html.escape(ylabel)} →</text>'
|
||||
f'</svg>')
|
||||
|
||||
|
||||
def svg_tp_sl_heatmap(grid_df: pd.DataFrame, w: int = 460, h: int = 360) -> str:
|
||||
"""TP×SL 净盈利热力图。"""
|
||||
p = grid_df.pivot_table(index="sl", columns="tp", values="net", aggfunc="first")
|
||||
p = p.sort_index().sort_index(axis=1)
|
||||
Z = p.to_numpy()
|
||||
xs = list(p.columns)
|
||||
ys = list(p.index)
|
||||
top, bottom, left, right = 30, 40, 60, 16
|
||||
plot_w = w - left - right
|
||||
plot_h = h - top - bottom
|
||||
ny, nx = Z.shape
|
||||
cw = plot_w / nx
|
||||
ch = plot_h / ny
|
||||
valid = Z[~np.isnan(Z)]
|
||||
vmin, vmax = float(valid.min()), float(valid.max())
|
||||
if vmax == vmin: vmax = vmin + 1
|
||||
|
||||
def col(v):
|
||||
if np.isnan(v): return "#f3f4f6"
|
||||
r = (v - vmin) / (vmax - vmin)
|
||||
if r < 0.5:
|
||||
t = r * 2
|
||||
return f"rgb({int(220-60*t)},{int(60+160*t)},{int(60+40*t)})"
|
||||
t = (r - 0.5) * 2
|
||||
return f"rgb({int(160-100*t)},{int(220-40*t)},{int(100-40*t)})"
|
||||
|
||||
cells = ""
|
||||
for j in range(ny):
|
||||
for i in range(nx):
|
||||
v = Z[j, i]
|
||||
x = left + i * cw
|
||||
y = top + (ny - 1 - j) * ch
|
||||
cells += f'<rect x="{x:.1f}" y="{y:.1f}" width="{cw:.1f}" height="{ch:.1f}" fill="{col(v)}" stroke="#fff" stroke-width="0.4"/>'
|
||||
# 最优点
|
||||
bi = np.unravel_index(np.nanargmax(Z), Z.shape)
|
||||
bx = left + bi[1] * cw + cw / 2
|
||||
by = top + (ny - 1 - bi[0]) * ch + ch / 2
|
||||
marker = f'<circle cx="{bx:.1f}" cy="{by:.1f}" r="6" fill="none" stroke="#1e3a8a" stroke-width="2"/>'
|
||||
|
||||
xlabs = "".join(f'<text x="{left+(i+0.5)*cw:.1f}" y="{h-bottom+14:.1f}" font-size="8" fill="#374151" text-anchor="middle">{xs[i]:.0f}</text>' for i in range(0, nx, 2))
|
||||
ylabs = "".join(f'<text x="{left-6:.1f}" y="{top+(ny-0.5-j)*ch+3:.1f}" font-size="8" fill="#374151" text-anchor="end">{ys[j]:.0f}</text>' for j in range(0, ny, 2))
|
||||
return (f'<svg viewBox="0 0 {w} {h}" class="chart">'
|
||||
f'<text x="{w//2}" y="16" font-size="11" fill="#374151" text-anchor="middle">理论净盈利 (TP × SL 扫描) — 圈=最优</text>'
|
||||
f'{cells}{marker}{xlabs}{ylabs}'
|
||||
f'<text x="{w//2}" y="{h-4}" font-size="9" fill="#6b7280" text-anchor="middle">TP(点) →</text>'
|
||||
f'<text x="14" y="{h//2}" font-size="9" fill="#6b7280" text-anchor="middle" transform="rotate(-90 14 {h//2})">SL(点) →</text>'
|
||||
f'</svg>')
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# 5. HTML
|
||||
# =========================================================================== #
|
||||
def build_html(df: pd.DataFrame, res: Dict[str, Any]) -> str:
|
||||
mae_svg = svg_scatter(
|
||||
df["mae"].to_numpy(), df["profit"].to_numpy(), df["profit"].to_numpy(),
|
||||
"MAE(点)", "盈利($)", "MAE vs 盈亏（绿=赢/红=输）",
|
||||
threshold=res["mae_q3_of_losers"], threshold_label="SL候选",
|
||||
)
|
||||
mfe_svg = svg_scatter(
|
||||
df["mfe"].to_numpy(), df["profit"].to_numpy(), df["profit"].to_numpy(),
|
||||
"MFE(点)", "盈利($)", "MFE vs 盈亏（绿=赢/红=输）",
|
||||
threshold=res["mfe_median_of_winners"], threshold_label="TP候选",
|
||||
)
|
||||
tpsl_svg = svg_tp_sl_heatmap(res["grid"])
|
||||
b = res["best"]
|
||||
|
||||
cards = (
|
||||
"<div class='cards'>"
|
||||
f"<div class='card'><div class='k'>样本笔数</div><div class='v'>{res['n']}</div></div>"
|
||||
f"<div class='card'><div class='k'>最优 TP(点)</div><div class='v'>{b['tp']:.0f}</div></div>"
|
||||
f"<div class='card'><div class='k'>最优 SL(点)</div><div class='v'>{b['sl']:.0f}</div></div>"
|
||||
f"<div class='card'><div class='k'>理论净盈利(点)</div><div class='v pos'>{b['net']:.0f}</div></div>"
|
||||
f"<div class='card'><div class='k'>理论胜率</div><div class='v'>{b['win_rate']:.1f}%</div></div>"
|
||||
f"<div class='card'><div class='k'>理论 PF</div><div class='v'>{b['pf']:.2f}</div></div>"
|
||||
"</div>"
|
||||
)
|
||||
|
||||
interp = ""
|
||||
if res["profit_corr_mae"] < -0.3:
|
||||
interp += f"<p class='muted'>盈亏与 MAE 负相关 ({res['profit_corr_mae']:.2f})：逆向偏移越大越易亏，设 SL 合理。SL 候选 = 亏损笔 MAE 75 分位 ≈ <b>{res['mae_q3_of_losers']:.0f} 点</b>。</p>"
|
||||
if res["profit_corr_mfe"] > 0.3:
|
||||
interp += f"<p class='muted'>盈亏与 MFE 正相关 ({res['profit_corr_mfe']:.2f})：能跑到 X 点的笔更易盈利，TP 候选 = 盈利笔 MFE 中位 ≈ <b>{res['mfe_median_of_winners']:.0f} 点</b>。</p>"
|
||||
if not interp:
|
||||
interp = "<p class='muted'>MAE/MFE 与盈亏相关性弱，TP/SL 优化空间有限，建议关注入场逻辑。</p>"
|
||||
|
||||
interp += (f"<div class='note'><b>理论最优 TP/SL</b>：TP={b['tp']:.0f}点 / SL={b['sl']:.0f}点，"
|
||||
f"在该组合下重算净盈利 {b['net']:.0f}点、胜率 {b['win_rate']:.1f}%、PF {b['pf']:.2f}。"
|
||||
f"此为基于历史 MAE/MFE 的理论值，需回测验证。</div>")
|
||||
|
||||
return f"""<!doctype html><html lang="zh-CN"><head><meta charset="utf-8">
|
||||
<title>MAE/MFE 分析</title>
|
||||
<style>
|
||||
body {{ font-family:-apple-system,"Microsoft YaHei",sans-serif; background:#f8fafc; color:#111827; margin:0; padding:20px;}}
|
||||
.wrap {{ max-width:1100px; margin:0 auto;}}
|
||||
h1,h2 {{ color:#1e3a8a;}} h1 {{ border-bottom:3px solid #1e3a8a; padding-bottom:8px;}}
|
||||
.muted {{ color:#6b7280; font-size:12px;}}
|
||||
.cards {{ display:grid; grid-template-columns:repeat(auto-fit,minmax(150px,1fr)); gap:12px; margin:16px 0;}}
|
||||
.card {{ background:#fff; padding:12px; border-radius:8px; border:1px solid #e2e8f0;}}
|
||||
.card .k {{ color:#6b7280; font-size:11px;}} .card .v {{ font-size:18px; font-weight:600; margin-top:4px;}}
|
||||
.card .v.pos {{ color:#16a34a;}}
|
||||
.chart {{ width:100%; height:auto; background:#fff; border:1px solid #e2e8f0; border-radius:6px;}}
|
||||
.grid2 {{ display:grid; grid-template-columns:1fr 1fr; gap:12px;}}
|
||||
.note {{ background:#fef3c7; border-left:4px solid #f59e0b; padding:10px 14px; margin:12px 0; border-radius:4px;}}
|
||||
pre {{ background:#1e293b; color:#e2e8f0; padding:14px; border-radius:6px; overflow:auto; font-size:11px;}}
|
||||
footer {{ color:#6b7280; font-size:11px; margin-top:32px; text-align:center;}}
|
||||
</style></head><body><div class="wrap">
|
||||
<h1>MAE / MFE 分析报告</h1>
|
||||
<p class='muted'>样本: {res['n']} 笔 · 盈亏-MAE 相关 {res['profit_corr_mae']:.2f} · 盈亏-MFE 相关 {res['profit_corr_mfe']:.2f}</p>
|
||||
{cards}
|
||||
{interp}
|
||||
<h2>1. MAE / MFE 散点</h2>
|
||||
<div class="grid2"><div>{mae_svg}</div><div>{mfe_svg}</div></div>
|
||||
<p class='muted'>蓝虚线 = 推荐的 SL/TP 候选阈值。MAE 图：右侧红点=逆向走很远还亏=该早止损；MFE 图：右侧绿点=涨得高最终赚=TP 可放到此处。</p>
|
||||
<h2>2. TP × SL 扫描热力图</h2>
|
||||
{tpsl_svg}
|
||||
<p class='muted'>圈=理论最优组合。色越绿=净盈利越高。注意"宽绿区"比"单点深绿"更稳健。</p>
|
||||
<h2>3. 如何获取 MAE/MFE 数据</h2>
|
||||
<p class='muted'>MT5 标准 xlsx 报告不含逐笔 MAE/MFE。需在 EA 的 OnDeinit 里加如下代码导出 CSV，再喂给本工具：</p>
|
||||
<pre>{html.escape(mql5_snippet())}</pre>
|
||||
<footer>由 mae_mfe.py 生成 · {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M')}</footer>
|
||||
</div></body></html>"""
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# 6. demo
|
||||
# =========================================================================== #
|
||||
def _demo_df() -> pd.DataFrame:
|
||||
rng = np.random.default_rng(3)
|
||||
n = 400
|
||||
direction = rng.choice(["long", "short"], n)
|
||||
mae = np.abs(rng.normal(20, 15, n))
|
||||
mfe = np.abs(rng.normal(40, 20, n))
|
||||
# 盈亏与 mfe 正相关、与 mae 负相关
|
||||
base = (mfe - mae) * 0.3 + rng.normal(0, 5, n)
|
||||
profit = np.where(base > 0, mfe * 0.2, -mae * 0.25) + rng.normal(0, 2, n)
|
||||
return pd.DataFrame({
|
||||
"ticket": np.arange(n),
|
||||
"direction": direction,
|
||||
"mae": np.round(mae, 1),
|
||||
"mfe": np.round(mfe, 1),
|
||||
"profit": np.round(profit, 2),
|
||||
})
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# CLI
|
||||
# =========================================================================== #
|
||||
def main(argv: List[str]) -> int:
|
||||
ap = argparse.ArgumentParser(description="MAE/MFE 分析")
|
||||
ap.add_argument("csv", nargs="?", help="mae_mfe.csv 路径")
|
||||
ap.add_argument("--show-snippet", action="store_true", help="打印 MQL5 代码片段")
|
||||
ap.add_argument("--demo", action="store_true")
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
os.makedirs(OUT_DIR, exist_ok=True)
|
||||
|
||||
if args.show_snippet:
|
||||
print(mql5_snippet())
|
||||
return 0
|
||||
|
||||
if args.demo:
|
||||
df = _demo_df()
|
||||
else:
|
||||
if not args.csv:
|
||||
ap.error("需要 csv 路径，或 --demo，或 --show-snippet")
|
||||
df = load_csv(args.csv)
|
||||
|
||||
res = analyze(df)
|
||||
html_doc = build_html(df, res)
|
||||
out = os.path.join(OUT_DIR, "mae_mfe.html")
|
||||
with open(out, "w", encoding="utf-8") as f:
|
||||
f.write(html_doc)
|
||||
print(f"HTML 报告: {out}")
|
||||
print(f"最优 TP={res['best']['tp']:.0f}点 SL={res['best']['sl']:.0f}点 "
|
||||
f"理论净盈利={res['best']['net']:.0f}点 PF={res['best']['pf']:.2f}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main(sys.argv[1:]))
|
||||
@@ -0,0 +1,414 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
MT5 策略测试报告解析器（中文版 xlsx）
|
||||
=====================================
|
||||
读取 MT5 Strategy Tester 导出的 .xlsx 报告，结构化为：
|
||||
- meta : 元信息（EA、品种、期间、经纪商、初始资金、杠杆、输入参数）
|
||||
- summary : “结果”区的汇总指标字典
|
||||
- orders : 订单明细 DataFrame
|
||||
- deals : 成交明细 DataFrame
|
||||
- trades : 由 in/out 成交对重建的“逐笔交易”DataFrame
|
||||
|
||||
设计为可复用：后续新的回测报告可直接喂给本解析器。
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
import numpy as np
|
||||
import openpyxl
|
||||
import pandas as pd
|
||||
|
||||
# 报告中区段标题（A 列出现的精确字符串）
|
||||
SEC_SETTINGS = "设置"
|
||||
SEC_RESULTS = "结果"
|
||||
SEC_ORDERS = "订单"
|
||||
SEC_DEALS = "成交"
|
||||
|
||||
|
||||
@dataclass
|
||||
class MT5Report:
|
||||
meta: Dict[str, Any] = field(default_factory=dict)
|
||||
summary: Dict[str, Any] = field(default_factory=dict)
|
||||
orders: Optional[pd.DataFrame] = None
|
||||
deals: Optional[pd.DataFrame] = None
|
||||
trades: Optional[pd.DataFrame] = None
|
||||
source_file: str = ""
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 工具函数
|
||||
# --------------------------------------------------------------------------- #
|
||||
def _find_section_rows(ws) -> Dict[str, int]:
|
||||
"""扫描 A 列，定位各段标题所在行。"""
|
||||
rows: Dict[str, int] = {}
|
||||
for r in range(1, ws.max_row + 1):
|
||||
v = ws.cell(r, 1).value
|
||||
if v in (SEC_SETTINGS, SEC_RESULTS, SEC_ORDERS, SEC_DEALS):
|
||||
rows[v] = r
|
||||
return rows
|
||||
|
||||
|
||||
def _kv_in_row(ws, r: int, max_col: int = 14) -> Dict[str, Any]:
|
||||
"""
|
||||
按行解析 “标签: 值” 对。
|
||||
MT5 报告里标签以全角冒号 ':' 结尾，值位于其右侧下一个非空单元格。
|
||||
一行可能含多组 标签/值。
|
||||
"""
|
||||
out: Dict[str, Any] = {}
|
||||
cells = [ws.cell(r, c).value for c in range(1, max_col + 1)]
|
||||
i = 0
|
||||
while i < len(cells):
|
||||
v = cells[i]
|
||||
if isinstance(v, str) and v.endswith(":"):
|
||||
label = v[:-1].strip()
|
||||
# 向右找第一个非空值
|
||||
j = i + 1
|
||||
while j < len(cells) and cells[j] is None:
|
||||
j += 1
|
||||
if j < len(cells):
|
||||
out[label] = cells[j]
|
||||
i = j + 1
|
||||
continue
|
||||
i += 1
|
||||
return out
|
||||
|
||||
|
||||
def _parse_table(
|
||||
ws, header_row: int, end_row: int, max_col: int = 14
|
||||
) -> pd.DataFrame:
|
||||
"""从 header_row 读取列名，读取其后到 end_row 的数据，返回 DataFrame。"""
|
||||
headers = [ws.cell(header_row, c).value for c in range(1, max_col + 1)]
|
||||
# 去掉末尾连续的 None 列
|
||||
while headers and headers[-1] is None:
|
||||
headers.pop()
|
||||
ncol = len(headers)
|
||||
|
||||
records: List[List[Any]] = []
|
||||
for r in range(header_row + 1, end_row + 1):
|
||||
row = [ws.cell(r, c).value for c in range(1, ncol + 1)]
|
||||
# 跳过整行空
|
||||
if all(v is None for v in row):
|
||||
continue
|
||||
records.append(row)
|
||||
|
||||
df = pd.DataFrame(records, columns=headers)
|
||||
return df
|
||||
|
||||
|
||||
def _to_float(x: Any) -> Optional[float]:
|
||||
"""从形如 '971.54 (96.32%)' 或 '0.69415' 的字符串中提取首个数值。"""
|
||||
if x is None:
|
||||
return None
|
||||
if isinstance(x, (int, float)):
|
||||
return float(x)
|
||||
if isinstance(x, str):
|
||||
m = re.search(r"-?\d+\.?\d*", x)
|
||||
return float(m.group()) if m else None
|
||||
return None
|
||||
|
||||
|
||||
def _pct_to_float(x: Any) -> Optional[float]:
|
||||
"""从 '96.32%' 提取 96.32。"""
|
||||
if x is None:
|
||||
return None
|
||||
if isinstance(x, (int, float)):
|
||||
return float(x)
|
||||
m = re.search(r"(-?\d+\.?\d*)\s*%", str(x))
|
||||
return float(m.group(1)) if m else None
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 主入口
|
||||
# --------------------------------------------------------------------------- #
|
||||
def parse_report(path: str) -> MT5Report:
|
||||
"""解析一份 MT5 xlsx 报告。"""
|
||||
wb = openpyxl.load_workbook(path, data_only=True)
|
||||
ws = wb[wb.sheetnames[0]]
|
||||
rep = MT5Report(source_file=path)
|
||||
|
||||
sec = _find_section_rows(ws)
|
||||
r_settings = sec.get(SEC_SETTINGS, 1)
|
||||
r_results = sec.get(SEC_RESULTS, r_settings)
|
||||
r_orders = sec.get(SEC_ORDERS, r_results)
|
||||
r_deals = sec.get(SEC_DEALS, r_orders)
|
||||
|
||||
# ---- 元信息 / 设置 ----
|
||||
# 整个设置块（设置标题行之后到结果段之前）逐行解析 “标签: 值” 对
|
||||
meta: Dict[str, Any] = {}
|
||||
for r in range(r_settings + 1, r_results):
|
||||
kv = _kv_in_row(ws, r)
|
||||
for k, v in kv.items():
|
||||
if k == "输入": # “输入:” 是子段标题，其值为分节字符串，跳过
|
||||
continue
|
||||
meta[k] = v
|
||||
|
||||
# 输入参数：在 “输入:” 段落里，列 D 形如 Key=Value
|
||||
inputs: Dict[str, str] = {}
|
||||
for r in range(r_settings + 1, r_results):
|
||||
for c in (4,): # 经验上输入参数在 D 列
|
||||
v = ws.cell(r, c).value
|
||||
if isinstance(v, str) and "=" in v and not v.endswith(":"):
|
||||
# 跳过分节标题（如 "=== PAINEL ===="）
|
||||
if v.strip().startswith("==="):
|
||||
continue
|
||||
k, _, val = v.partition("=")
|
||||
inputs[k.strip()] = val.strip()
|
||||
meta["inputs"] = inputs
|
||||
|
||||
# ---- 结果汇总 ----
|
||||
summary: Dict[str, Any] = {}
|
||||
for r in range(r_results, r_orders):
|
||||
summary.update(_kv_in_row(ws, r))
|
||||
|
||||
# ---- 订单表 ----
|
||||
rep.orders = _parse_table(ws, header_row=r_orders + 1, end_row=r_deals - 1)
|
||||
|
||||
# ---- 成交表 ----
|
||||
rep.deals = _parse_table(ws, header_row=r_deals + 1, end_row=ws.max_row)
|
||||
|
||||
rep.meta = meta
|
||||
rep.summary = summary
|
||||
rep.trades = _reconstruct_trades(rep.deals)
|
||||
_normalize_summary_numbers(rep)
|
||||
return rep
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 逐笔交易重建
|
||||
# --------------------------------------------------------------------------- #
|
||||
DEALS_COLS = {
|
||||
"time": "时间",
|
||||
"deal_id": "成交",
|
||||
"symbol": "交易品种",
|
||||
"type": "类型", # buy / sell / balance
|
||||
"entry": "趋势", # in / out / None
|
||||
"volume": "交易量",
|
||||
"price": "价位",
|
||||
"order": "订单",
|
||||
"commission": "手续费",
|
||||
"swap": "库存费",
|
||||
"profit": "盈利",
|
||||
"balance": "结余",
|
||||
"comment": "注释",
|
||||
}
|
||||
|
||||
|
||||
def _reconstruct_trades(deals: pd.DataFrame) -> pd.DataFrame:
|
||||
"""
|
||||
由成交明细重建逐笔交易。
|
||||
规则：每笔 'in' 成交后紧跟其 'out' 平仓成交，两两配对。
|
||||
返回字段：open_time, close_time, direction, volume, open_price,
|
||||
close_price, profit, swap, commission, net_profit, duration_min
|
||||
"""
|
||||
if deals is None or deals.empty:
|
||||
return pd.DataFrame()
|
||||
|
||||
d = deals.copy()
|
||||
# 统一列名
|
||||
d = d.rename(columns={v: k for k, v in DEALS_COLS.items() if v in d.columns})
|
||||
|
||||
# 类型转换
|
||||
d["time"] = pd.to_datetime(d["time"], errors="coerce")
|
||||
for col in ("price", "commission", "swap", "profit", "balance"):
|
||||
if col in d.columns:
|
||||
d[col] = pd.to_numeric(d[col], errors="coerce")
|
||||
# 成交量形如 '0.01 / 0.01'，取前半
|
||||
if "volume" in d.columns:
|
||||
d["volume"] = d["volume"].astype(str).str.split("/").str[0].str.strip()
|
||||
d["volume"] = pd.to_numeric(d["volume"], errors="coerce")
|
||||
|
||||
# 仅保留 in/out 行（balance 行无 entry）
|
||||
d = d[d["entry"].isin(["in", "out"])].reset_index(drop=True)
|
||||
|
||||
trades: List[Dict[str, Any]] = []
|
||||
i = 0
|
||||
while i < len(d) - 1:
|
||||
in_row = d.iloc[i]
|
||||
out_row = d.iloc[i + 1]
|
||||
if in_row["entry"] != "in" or out_row["entry"] != "out":
|
||||
i += 1
|
||||
continue
|
||||
direction = "short" if in_row["type"] == "sell" else "long"
|
||||
profit = float(out_row.get("profit", 0.0) or 0.0)
|
||||
swap = float(out_row.get("swap", 0.0) or 0.0)
|
||||
commission = float(out_row.get("commission", 0.0) or 0.0)
|
||||
dur = (out_row["time"] - in_row["time"]).total_seconds() / 60.0
|
||||
trades.append(
|
||||
{
|
||||
"open_time": in_row["time"],
|
||||
"close_time": out_row["time"],
|
||||
"direction": direction,
|
||||
"volume": in_row.get("volume"),
|
||||
"open_price": in_row.get("price"),
|
||||
"close_price": out_row.get("price"),
|
||||
"profit": profit,
|
||||
"swap": swap,
|
||||
"commission": commission,
|
||||
"net_profit": profit + swap + commission,
|
||||
"duration_min": dur,
|
||||
}
|
||||
)
|
||||
i += 2
|
||||
|
||||
return pd.DataFrame(trades)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 汇总指标数值化（便于程序对比）
|
||||
# --------------------------------------------------------------------------- #
|
||||
def _normalize_summary_numbers(rep: MT5Report) -> None:
|
||||
s = rep.summary
|
||||
norm: Dict[str, Optional[float]] = {}
|
||||
|
||||
# 直接数值字段
|
||||
direct = {
|
||||
"总净盈利": "total_net_profit",
|
||||
"毛利": "gross_profit",
|
||||
"毛损": "gross_loss",
|
||||
"盈利因子": "profit_factor",
|
||||
"预期收益": "expected_payoff",
|
||||
"夏普比率": "sharpe",
|
||||
"采收率": "recovery",
|
||||
"LR 相关性": "lr_correlation",
|
||||
"LR 标准误差": "lr_std_error",
|
||||
"总成交": "total_deals",
|
||||
"交易总计": "total_trades",
|
||||
}
|
||||
for cn, en in direct.items():
|
||||
norm[en] = _to_float(s.get(cn))
|
||||
|
||||
# 回撤类形如 "971.54 (96.32%)" —— 同时拆金额与百分比
|
||||
dd_pairs = {
|
||||
"最大结余亏损": "max_balance_dd",
|
||||
"最大净值亏损": "max_equity_dd",
|
||||
"相对结余亏损": "rel_balance_dd",
|
||||
"相对净值亏损": "rel_equity_dd",
|
||||
}
|
||||
for cn, en in dd_pairs.items():
|
||||
val = _to_float(s.get(cn)) # 取金额
|
||||
# 单独找百分比
|
||||
pct = _pct_to_float(s.get(cn))
|
||||
norm[en] = val
|
||||
norm[en + "_pct"] = pct
|
||||
|
||||
# 胜率类形如 "732 (38.28%)" —— 取百分比
|
||||
wr_pairs = {
|
||||
"盈利交易 (% 全部)": "win_rate_pct",
|
||||
"亏损交易 (% 全部)": "loss_rate_pct",
|
||||
"卖出交易 (赢得 %)": "sell_win_rate_pct",
|
||||
"买入交易 (赢得 %)": "buy_win_rate_pct",
|
||||
}
|
||||
for cn, en in wr_pairs.items():
|
||||
norm[en] = _pct_to_float(s.get(cn))
|
||||
|
||||
# 平均盈亏
|
||||
norm["avg_win"] = _to_float(s.get("平均 获利交易"))
|
||||
norm["avg_loss"] = _to_float(s.get("平均 亏损交易"))
|
||||
norm["max_win"] = _to_float(s.get("最大 获利交易"))
|
||||
norm["max_loss"] = _to_float(s.get("最大 亏损交易"))
|
||||
|
||||
rep.summary_norm = norm # type: ignore[attr-defined]
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 便捷：从 trades 直接计算扩展指标
|
||||
# --------------------------------------------------------------------------- #
|
||||
def compute_trade_metrics(trades: pd.DataFrame) -> Dict[str, Any]:
|
||||
"""由逐笔交易 DataFrame 计算扩展指标。"""
|
||||
if trades is None or trades.empty:
|
||||
return {}
|
||||
|
||||
t = trades
|
||||
n = len(t)
|
||||
net = t["net_profit"].astype(float)
|
||||
wins = net[net > 0]
|
||||
losses = net[net <= 0]
|
||||
gross_profit = wins.sum()
|
||||
gross_loss = -losses.sum()
|
||||
pf = gross_profit / gross_loss if gross_loss > 0 else np.inf
|
||||
|
||||
# 连续盈亏
|
||||
streak_win = streak_loss = cur_w = cur_l = 0
|
||||
max_streak_win = max_streak_loss = 0
|
||||
for v in net:
|
||||
if v > 0:
|
||||
cur_w += 1
|
||||
cur_l = 0
|
||||
max_streak_win = max(max_streak_win, cur_w)
|
||||
else:
|
||||
cur_l += 1
|
||||
cur_w = 0
|
||||
max_streak_loss = max(max_streak_loss, cur_l)
|
||||
|
||||
# 资金曲线与回撤
|
||||
equity = net.cumsum()
|
||||
running_max = equity.cummax()
|
||||
dd = equity - running_max
|
||||
max_dd = dd.min()
|
||||
|
||||
# 按方向
|
||||
by_dir = {}
|
||||
for direction, g in t.groupby("direction"):
|
||||
gn = g["net_profit"].astype(float)
|
||||
w = gn[gn > 0]
|
||||
l = gn[gn <= 0]
|
||||
gp = w.sum()
|
||||
gl = -l.sum()
|
||||
by_dir[direction] = {
|
||||
"n": len(g),
|
||||
"win_rate": len(w) / len(g) * 100 if len(g) else 0,
|
||||
"net_profit": gn.sum(),
|
||||
"profit_factor": gp / gl if gl > 0 else np.inf,
|
||||
"avg_win": w.mean() if len(w) else 0,
|
||||
"avg_loss": l.mean() if len(l) else 0,
|
||||
"expectancy": gn.mean(),
|
||||
}
|
||||
|
||||
# 按小时 / 星期 / 月份
|
||||
t2 = t.copy()
|
||||
t2["hour"] = t2["open_time"].dt.hour
|
||||
t2["weekday"] = t2["open_time"].dt.dayofweek # 0=Mon
|
||||
t2["month"] = t2["open_time"].dt.to_period("M").astype(str)
|
||||
by_hour = t2.groupby("hour")["net_profit"].agg(["count", "sum", "mean"]).to_dict("index")
|
||||
by_weekday = t2.groupby("weekday")["net_profit"].agg(["count", "sum", "mean"]).to_dict("index")
|
||||
by_month = t2.groupby("month")["net_profit"].agg(["count", "sum", "mean"]).to_dict("index")
|
||||
|
||||
return {
|
||||
"n_trades": n,
|
||||
"net_profit": float(net.sum()),
|
||||
"win_rate": float(len(wins) / n * 100),
|
||||
"profit_factor": float(pf),
|
||||
"avg_win": float(wins.mean()) if len(wins) else 0.0,
|
||||
"avg_loss": float(losses.mean()) if len(losses) else 0.0,
|
||||
"expectancy": float(net.mean()),
|
||||
"max_streak_win": int(max_streak_win),
|
||||
"max_streak_loss": int(max_streak_loss),
|
||||
"max_dd": float(max_dd),
|
||||
"avg_duration_min": float(t["duration_min"].mean()),
|
||||
"median_duration_min": float(t["duration_min"].median()),
|
||||
"by_direction": by_dir,
|
||||
"by_hour": by_hour,
|
||||
"by_weekday": by_weekday,
|
||||
"by_month": by_month,
|
||||
"_equity": equity,
|
||||
"_dd": dd,
|
||||
}
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import sys
|
||||
|
||||
for f in sys.argv[1:]:
|
||||
rep = parse_report(f)
|
||||
print(f"== {f} ==")
|
||||
print("EA:", rep.meta.get("专家"))
|
||||
print("symbol:", rep.meta.get("交易品种"))
|
||||
print("period:", rep.meta.get("期间"))
|
||||
print("inputs:", rep.meta.get("inputs"))
|
||||
print("trades rows:", 0 if rep.trades is None else len(rep.trades))
|
||||
print("deals rows:", 0 if rep.deals is None else len(rep.deals))
|
||||
print("summary_norm sample:", rep.summary_norm) # type: ignore[attr-defined]
|
||||
print()
|
||||
File diff suppressed because one or more lines are too long
@@ -0,0 +1,10 @@
|
||||
; === 由 param_scan.py 优化生成 ===
|
||||
; 来源: 单点峰值（PF=2.281，来源 —）
|
||||
; 生成时间: 2026-06-28 02:22
|
||||
; 应用参数: {'FastMA': '25.0', 'SlowMA': '60.0'}
|
||||
|
||||
; inputs
|
||||
FastMA=25.0||1||1||100||N
|
||||
SlowMA=60.0||1||1||500||N
|
||||
StopLoss=100||1||1||10000||N
|
||||
TakeProfit=200||1||1||10000||N
|
||||
@@ -0,0 +1,10 @@
|
||||
; === 由 param_scan.py 优化生成 ===
|
||||
; 来源: 最大高原中心（面积 5，PF 均值 1.399）
|
||||
; 生成时间: 2026-06-28 02:21
|
||||
; 应用参数: {'FastMA': '10', 'SlowMA': '40'}
|
||||
|
||||
; inputs
|
||||
FastMA=10||1||1||100||N
|
||||
SlowMA=40||1||1||500||N
|
||||
StopLoss=100||1||1||10000||N
|
||||
TakeProfit=200||1||1||10000||N
|
||||
File diff suppressed because one or more lines are too long
File diff suppressed because one or more lines are too long
@@ -0,0 +1,23 @@
|
||||
<!doctype html><html lang="zh-CN"><head><meta charset="utf-8">
|
||||
<title>Walk-Forward 报告 — demo</title>
|
||||
<style>
|
||||
body { font-family: -apple-system, "Microsoft YaHei", sans-serif; background:#f8fafc; color:#111827; margin:0; padding:20px;}
|
||||
.wrap { max-width:1100px; margin:0 auto;}
|
||||
h1 { color:#1e3a8a; border-bottom:3px solid #1e3a8a; padding-bottom:8px;}
|
||||
table.data { border-collapse:collapse; width:100%; background:#fff; margin:8px 0 16px; font-size:12px;}
|
||||
table.data th, table.data td { border:1px solid #e2e8f0; padding:5px 7px; text-align:left;}
|
||||
table.data th { background:#f1f5f9;}
|
||||
td.pos, tr.pos td { color:#16a34a;} td.neg, tr.neg td { color:#dc2626;}
|
||||
.muted { color:#6b7280; font-size:12px;}
|
||||
.cards { display:grid; grid-template-columns:repeat(auto-fit,minmax(200px,1fr)); gap:12px; margin:16px 0;}
|
||||
.card { background:#fff; padding:14px; border-radius:8px; border:1px solid #e2e8f0;}
|
||||
.card .k { color:#6b7280; font-size:12px;} .card .v { font-size:20px; font-weight:600; margin-top:4px;}
|
||||
.card .v.pos { color:#16a34a;} .card .v.neg { color:#dc2626;}
|
||||
.chart { width:100%; height:auto; background:#fff; border:1px solid #e2e8f0; border-radius:6px;}
|
||||
.note { background:#fef3c7; border-left:4px solid #f59e0b; padding:10px 14px; margin:12px 0; border-radius:4px;}
|
||||
.warn { background:#fee2e2; border-left:4px solid #dc2626; padding:10px 14px; margin:12px 0; border-radius:4px;}
|
||||
</style></head><body><div class="wrap">
|
||||
<h1>Walk-Forward 滚动验证报告</h1>
|
||||
<p class='muted'>来源: demo</p>
|
||||
<div class='cards'><div class='card'><div class='k'>窗口数</div><div class='v'>3</div></div><div class='card'><div class='k'>WFE (ΣOOS净/ΣIS净)</div><div class='v pos'>+0.51</div></div><div class='card'><div class='k'>OOS 盈利窗口占比</div><div class='v'>0%</div></div><div class='card'><div class='k'>IS-OOS PF 相关性</div><div class='v'>-0.28</div></div></div><svg viewBox="0 0 760 260" class="chart"><line x1="46" y1="224.0" x2="744" y2="224.0" stroke="#e5e7eb" stroke-width="0.5"/><text x="40" y="227.0" font-size="9" fill="#6b7280" text-anchor="end">0.00</text><line x1="46" y1="174.0" x2="744" y2="174.0" stroke="#e5e7eb" stroke-width="0.5"/><text x="40" y="177.0" font-size="9" fill="#6b7280" text-anchor="end">0.25</text><line x1="46" y1="124.0" x2="744" y2="124.0" stroke="#e5e7eb" stroke-width="0.5"/><text x="40" y="127.0" font-size="9" fill="#6b7280" text-anchor="end">0.50</text><line x1="46" y1="74.0" x2="744" y2="74.0" stroke="#e5e7eb" stroke-width="0.5"/><text x="40" y="77.0" font-size="9" fill="#6b7280" text-anchor="end">0.75</text><line x1="46" y1="24.0" x2="744" y2="24.0" stroke="#e5e7eb" stroke-width="0.5"/><text x="40" y="27.0" font-size="9" fill="#6b7280" text-anchor="end">1.00</text><line x1="46" y1="24.0" x2="744" y2="24.0" stroke="#9ca3af" stroke-dasharray="3,3" stroke-width="0.8"/><polyline points="46.0,79.4 395.0,64.9 744.0,63.7" fill="none" stroke="#2563eb" stroke-width="1.6"><title></title></polyline><polyline points="46.0,66.6 395.0,60.7 744.0,79.2" fill="none" stroke="#dc2626" stroke-width="1.6"><title></title></polyline><text x="46.0" y="238.0" font-size="9" fill="#374151" text-anchor="middle">#0</text><text x="395.0" y="238.0" font-size="9" fill="#374151" text-anchor="middle">#1</text><text x="744.0" y="238.0" font-size="9" fill="#374151" text-anchor="middle">#2</text><text x="380" y="16" font-size="11" fill="#374151" text-anchor="middle">滚动窗口 IS vs OOS 盈利因子 (PF=1 为盈亏平衡线)</text><text x="744" y="32" font-size="10" fill="#2563eb" text-anchor="end">IS PF</text><text x="744" y="46" font-size="10" fill="#dc2626" text-anchor="end">OOS PF</text><text x="380" y="256" font-size="9" fill="#6b7280" text-anchor="middle">窗口序号 →</text></svg><table class='data'><thead><tr><th>窗口</th><th>IS区间</th><th>OOS区间</th><th>IS笔</th><th>IS净</th><th>IS PF</th><th>IS胜率</th><th>OOS笔</th><th>OOS净</th><th>OOS PF</th><th>OOS胜率</th><th>OOS回撤</th></tr></thead><tbody><tr><td>#0</td><td>2024-01-01~2024-03-01</td><td>2024-03-01~2024-03-31</td><td>480</td><td>-171.12</td><td>0.72</td><td>43.3%</td><td>240</td><td class='neg'>-63.73</td><td>0.79</td><td>45.0%</td><td>-103.50</td></tr><tr><td>#1</td><td>2024-01-31~2024-03-31</td><td>2024-03-31~2024-04-30</td><td>480</td><td>-122.89</td><td>0.80</td><td>45.0%</td><td>240</td><td class='neg'>-55.43</td><td>0.82</td><td>45.0%</td><td>-99.96</td></tr><tr><td>#2</td><td>2024-03-01~2024-04-30</td><td>2024-04-30~2024-05-30</td><td>480</td><td>-119.16</td><td>0.80</td><td>45.0%</td><td>240</td><td class='neg'>-91.91</td><td>0.72</td><td>41.7%</td><td>-120.94</td></tr></tbody></table><p class='muted'>IS 与 OOS 的 PF 相关性低 → IS 表现难以预测 OOS，参数不稳健，警惕过拟合单段行情。</p><div class='note'>WFE > 0.5：OOS 能保留 IS 的一半以上盈利，泛化较好。</div>
|
||||
</div></body></html>
|
||||
+766
@@ -0,0 +1,766 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
参数敏感度扫描
|
||||
==============
|
||||
对 EA 关键输入参数做网格回测，分析响应面，找稳健参数高原
|
||||
（plateau，即大面积参数都能盈利的区域，而非单点过拟合峰值）。
|
||||
|
||||
工作流：
|
||||
1. gen-set : 根据 grid 定义 + .set 模板，批量生成每个参数组合的 .set 文件，
|
||||
并写 manifest.csv（记录文件名↔参数值），生成 MT5 批处理脚本。
|
||||
2. (用户在 MT5 里跑批处理，得到一批 ReportTester-*.xlsx)
|
||||
3. analyze : 读取 manifest + 所有报告，构建响应面：
|
||||
- 2D 热力图（参数X × 参数Y → 指标值，最适合找高原）
|
||||
- 3D 等距投影曲面
|
||||
- 高原检测：识别"指标高于阈值且邻域也高"的连通区域
|
||||
- 与单点峰值对比，标记过拟合风险
|
||||
|
||||
用法:
|
||||
python param_scan.py gen-set template.set grid.json out_setdir \\
|
||||
--ea "MyEA" --symbol XAUUSD --period M1 --from 2024-01-01 --to 2024-06-01
|
||||
python param_scan.py analyze reports_dir manifest.csv --x FastMA --y SlowMA --metric PF
|
||||
python param_scan.py demo # 用合成数据演示响应面与高原检测
|
||||
|
||||
grid.json 示例:
|
||||
{"FastMA": [5,10,15,20,25], "SlowMA": [20,30,40,50], "StopLoss": [50,100]}
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import html
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
from itertools import product
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
import mt5_report_parser as mp
|
||||
|
||||
OUT_DIR = "output"
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# 1. 网格 .set 文件生成
|
||||
# =========================================================================== #
|
||||
def _apply_params_to_template(template: str, params: Dict[str, str]) -> str:
|
||||
"""
|
||||
MT5 .set 文件每行形如: varname=value||0||0||0||0||0||...
|
||||
本函数把指定 varname 的 value 替换为新值，保留行尾的 || 字段。
|
||||
若模板中不存在该变量，追加一行。
|
||||
"""
|
||||
lines = template.splitlines()
|
||||
out: List[str] = []
|
||||
replaced = set()
|
||||
for line in lines:
|
||||
if "=" in line and not line.strip().startswith(";"):
|
||||
key, _, rest = line.partition("=")
|
||||
k = key.strip()
|
||||
if k in params:
|
||||
# rest 形如 value||... 只替换第一个 || 之前的部分
|
||||
if "||" in rest:
|
||||
_, _, tail = rest.partition("||")
|
||||
out.append(f"{key}={params[k]}||{tail}")
|
||||
else:
|
||||
out.append(f"{key}={params[k]}")
|
||||
replaced.add(k)
|
||||
continue
|
||||
out.append(line)
|
||||
for k, v in params.items():
|
||||
if k not in replaced:
|
||||
out.append(f"{k}={v}||0||0||0||0||0")
|
||||
return "\n".join(out) + "\n"
|
||||
|
||||
|
||||
def generate_set_files(
|
||||
template_set_path: str,
|
||||
grid: Dict[str, List[Any]],
|
||||
out_dir: str,
|
||||
ea_name: str = "EA",
|
||||
symbol: str = "XAUUSD",
|
||||
period: str = "M1",
|
||||
date_from: str = "2024-01-01",
|
||||
date_to: str = "2024-06-01",
|
||||
terminal_path: str = r"C:\Program Files\MetaTrader 5\terminal64.exe",
|
||||
) -> str:
|
||||
"""
|
||||
生成所有参数组合的 .set 文件 + manifest.csv + 批处理脚本。
|
||||
返回 manifest.csv 路径。
|
||||
"""
|
||||
os.makedirs(out_dir, exist_ok=True)
|
||||
with open(template_set_path, "r", encoding="utf-8", errors="replace") as f:
|
||||
template = f.read()
|
||||
|
||||
keys = list(grid.keys())
|
||||
combos = list(product(*[grid[k] for k in keys]))
|
||||
|
||||
manifest_rows: List[Dict[str, Any]] = []
|
||||
for i, combo in enumerate(combos):
|
||||
params = {k: str(v) for k, v in zip(keys, combo)}
|
||||
set_text = _apply_params_to_template(template, params)
|
||||
set_name = f"{ea_name}_{i:04d}.set"
|
||||
set_path = os.path.join(out_dir, set_name)
|
||||
with open(set_path, "w", encoding="utf-8") as f:
|
||||
f.write(set_text)
|
||||
row = {"idx": i, "set_file": set_name}
|
||||
row.update(params)
|
||||
manifest_rows.append(row)
|
||||
|
||||
manifest_path = os.path.join(out_dir, "manifest.csv")
|
||||
pd.DataFrame(manifest_rows).to_csv(manifest_path, index=False, encoding="utf-8-sig")
|
||||
|
||||
# 批处理脚本：逐个跑回测，输出 ReportTester-XXXX.xlsx
|
||||
bat_lines = [
|
||||
"@echo off",
|
||||
f"REM MT5 批量回测脚本 — 由 param_scan.py 生成",
|
||||
f"REM 共 {len(combos)} 个参数组合",
|
||||
f'set TERMINAL="{terminal_path}"',
|
||||
f'set EA={ea_name}',
|
||||
f'set SYMBOL={symbol}',
|
||||
f'set PERIOD={period}',
|
||||
f'set FROM={date_from}',
|
||||
f'set TO={date_to}',
|
||||
f'set OUTDIR={out_dir}\\reports',
|
||||
f'if not exist "%OUTDIR%" mkdir "%OUTDIR%"',
|
||||
"",
|
||||
"REM 说明：MT5 terminal.exe 单实例运行，每次跑一个 .set",
|
||||
"REM 配置文件 (.ini) 指定 ReportTester 输出路径，跑完把 xlsx 移到 OUTDIR",
|
||||
"REM 这里给出调用骨架；实际需配合 Common/tester.ini 模板",
|
||||
]
|
||||
for r in manifest_rows:
|
||||
bat_lines.append(
|
||||
f'REM [{r["idx"]}] {r["set_file"]} '
|
||||
f'{", ".join(f"{k}={r[k]}" for k in keys)}'
|
||||
)
|
||||
bat_lines.append(
|
||||
f'echo Running {r["set_file"]} ...'
|
||||
)
|
||||
bat_lines.append(
|
||||
f'%TERMINAL% /portable /config:tester_{r["idx"]:04d}.ini'
|
||||
)
|
||||
bat_lines.append("echo All done.")
|
||||
bat_path = os.path.join(out_dir, "run_scan.bat")
|
||||
with open(bat_path, "w", encoding="utf-8") as f:
|
||||
f.write("\r\n".join(bat_lines))
|
||||
|
||||
# tester.ini 模板说明
|
||||
ini_note = (
|
||||
"; tester.ini 模板示例（每组合一份，替换 <...>）\n"
|
||||
"[Tester]\n"
|
||||
f"Expert=Experts\\{ea_name}.ex5\n"
|
||||
f"Symbol={symbol}\n"
|
||||
f"Period={period}\n"
|
||||
f"FromDate={date_from}\n"
|
||||
f"ToDate={date_to}\n"
|
||||
"Deposit=10000\n"
|
||||
"Currency=USD\n"
|
||||
"Leverage=100\n"
|
||||
"Model=1\n"
|
||||
"Optimization=0\n"
|
||||
"Visual=0\n"
|
||||
"ForwardMode=0\n"
|
||||
"Report=<OUTDIR>\\ReportTester_<IDX>\n"
|
||||
"ReplaceReport=1\n"
|
||||
"ShutdownTerminal=1\n"
|
||||
"UseLocal=1\n"
|
||||
"InputSet=<SET_FILE_FULL_PATH>\n"
|
||||
)
|
||||
with open(os.path.join(out_dir, "tester_ini_template.txt"), "w", encoding="utf-8") as f:
|
||||
f.write(ini_note)
|
||||
|
||||
print(f"已生成 {len(combos)} 个 .set 文件 + manifest.csv + run_scan.bat")
|
||||
print(f" 目录: {out_dir}")
|
||||
print(f" 下一步: 编辑 tester_ini_template.txt，在 MT5 里跑 run_scan.bat，")
|
||||
print(f" 把生成的 ReportTester-*.xlsx 放到 {out_dir}\\reports\\ 后执行 analyze。")
|
||||
return manifest_path
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# 2. 加载扫描结果
|
||||
# =========================================================================== #
|
||||
def load_scan_results(reports_dir: str, manifest_csv: str) -> pd.DataFrame:
|
||||
"""
|
||||
读取 manifest + reports_dir 下所有 xlsx，解析每份报告的关键指标，
|
||||
合并参数 → 返回 DataFrame[参数列..., net, pf, win_rate, max_dd_pct]。
|
||||
"""
|
||||
manifest = pd.read_csv(manifest_csv)
|
||||
rows: List[Dict[str, Any]] = []
|
||||
for _, mrow in manifest.iterrows():
|
||||
# 报告文件名约定：ReportTester_<idx>.xlsx 或 manifest 里的 set_file 改后缀
|
||||
idx = int(mrow["idx"])
|
||||
candidates = [
|
||||
os.path.join(reports_dir, f"ReportTester-{idx:04d}.xlsx"),
|
||||
os.path.join(reports_dir, f"ReportTester-{idx}.xlsx"),
|
||||
os.path.join(reports_dir, str(mrow["set_file"]).replace(".set", ".xlsx")),
|
||||
]
|
||||
path = next((c for c in candidates if os.path.exists(c)), None)
|
||||
if path is None:
|
||||
continue
|
||||
try:
|
||||
rep = mp.parse_report(path)
|
||||
s = rep.summary_norm
|
||||
rows.append({
|
||||
**{k: mrow[k] for k in manifest.columns},
|
||||
"net": s.get("total_net_profit"),
|
||||
"pf": s.get("profit_factor"),
|
||||
"win_rate": s.get("win_rate_pct"),
|
||||
"max_dd_pct": s.get("max_balance_dd_pct"),
|
||||
"sharpe": s.get("sharpe"),
|
||||
})
|
||||
except Exception as e:
|
||||
print(f" 跳过 {path}: {e}")
|
||||
return pd.DataFrame(rows)
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# 3. 响应面 + 高原检测
|
||||
# =========================================================================== #
|
||||
def build_pivot(df: pd.DataFrame, x: str, y: str, metric: str) -> Tuple[np.ndarray, List, List]:
|
||||
"""透视成 2D 矩阵 (行=y, 列=x)。"""
|
||||
p = df.pivot_table(index=y, columns=x, values=metric, aggfunc="mean")
|
||||
p = p.sort_index().sort_index(axis=1)
|
||||
return p.to_numpy(), list(p.columns), list(p.index)
|
||||
|
||||
|
||||
def detect_plateaus(
|
||||
Z: np.ndarray, percentile: float = 80, min_cluster: int = 4
|
||||
) -> List[Dict[str, Any]]:
|
||||
"""
|
||||
高原检测：指标值高于 percentile 阈值，且 4-邻域连通区域 ≥ min_cluster 个点。
|
||||
返回每个高原的边界、中心、均值、面积。越大越宽的高原 = 越稳健。
|
||||
"""
|
||||
if Z.size == 0 or np.all(np.isnan(Z)):
|
||||
return []
|
||||
valid = Z[~np.isnan(Z)]
|
||||
if valid.size == 0:
|
||||
return []
|
||||
thr = float(np.percentile(valid, percentile))
|
||||
binary = (Z >= thr) & ~np.isnan(Z)
|
||||
|
||||
# 4-连通 DFS
|
||||
visited = np.zeros_like(binary, dtype=bool)
|
||||
ny, nx = Z.shape
|
||||
clusters: List[List[Tuple[int, int]]] = []
|
||||
for sy in range(ny):
|
||||
for sx in range(nx):
|
||||
if binary[sy, sx] and not visited[sy, sx]:
|
||||
stack = [(sy, sx)]
|
||||
comp = []
|
||||
while stack:
|
||||
cy, cx = stack.pop()
|
||||
if cy < 0 or cy >= ny or cx < 0 or cx >= nx:
|
||||
continue
|
||||
if visited[cy, cx] or not binary[cy, cx]:
|
||||
continue
|
||||
visited[cy, cx] = True
|
||||
comp.append((cy, cx))
|
||||
stack.extend([(cy+1, cx), (cy-1, cx), (cy, cx+1), (cy, cx-1)])
|
||||
if len(comp) >= min_cluster:
|
||||
clusters.append(comp)
|
||||
|
||||
plateaus: List[Dict[str, Any]] = []
|
||||
for comp in clusters:
|
||||
ys = [c[0] for c in comp]
|
||||
xs = [c[1] for c in comp]
|
||||
vals = [Z[c] for c in comp]
|
||||
plateaus.append({
|
||||
"size": len(comp),
|
||||
"y_range": (int(min(ys)), int(max(ys))),
|
||||
"x_range": (int(min(xs)), int(max(xs))),
|
||||
"metric_mean": float(np.mean(vals)),
|
||||
"metric_max": float(max(vals)),
|
||||
})
|
||||
plateaus.sort(key=lambda p: p["size"], reverse=True)
|
||||
return plateaus
|
||||
|
||||
|
||||
def overfit_score(Z: np.ndarray, plateaus: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||||
"""
|
||||
过拟合评分：
|
||||
peak_value = 全局最大值
|
||||
peak_isolation = 最大值周围邻域均值 / 最大值（越小说明越孤立=过拟合）
|
||||
plateau_coverage = 高原面积占总面积比例（越高越稳健）
|
||||
"""
|
||||
if Z.size == 0 or np.all(np.isnan(Z)):
|
||||
return {}
|
||||
valid = Z[~np.isnan(Z)]
|
||||
peak = float(np.nanmax(Z))
|
||||
py, px = np.unravel_index(np.nanargmax(Z), Z.shape)
|
||||
# 邻域均值（3x3 减自身）
|
||||
nbrs = []
|
||||
for dy in (-1, 0, 1):
|
||||
for dx in (-1, 0, 1):
|
||||
if dy == 0 and dx == 0:
|
||||
continue
|
||||
ny, nx = py + dy, px + dx
|
||||
if 0 <= ny < Z.shape[0] and 0 <= nx < Z.shape[1] and not np.isnan(Z[ny, nx]):
|
||||
nbrs.append(Z[ny, nx])
|
||||
nbr_mean = float(np.mean(nbrs)) if nbrs else float("nan")
|
||||
isolation = nbr_mean / peak if peak != 0 else float("nan")
|
||||
plateau_cov = sum(p["size"] for p in plateaus) / Z.size
|
||||
# 评分：越接近 1 越稳健，越接近 0 越过拟合
|
||||
robust = float(isolation * 0.5 + min(1, plateau_cov * 4) * 0.5) if not np.isnan(isolation) else float("nan")
|
||||
return {
|
||||
"peak": peak,
|
||||
"peak_neighbor_mean": nbr_mean,
|
||||
"peak_isolation_ratio": isolation,
|
||||
"plateau_coverage": plateau_cov,
|
||||
"robust_score": robust,
|
||||
}
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# 4. SVG 可视化
|
||||
# =========================================================================== #
|
||||
def svg_heatmap(
|
||||
Z: np.ndarray, xticks: List, yticks: List,
|
||||
metric: str, xname: str, yname: str,
|
||||
plateaus: List[Dict[str, Any]],
|
||||
w: int = 560, h: int = 420,
|
||||
) -> str:
|
||||
"""2D 热力图。高原用粗边框标出。"""
|
||||
if Z.size == 0:
|
||||
return ""
|
||||
top, bottom, left, right = 50, 50, 70, 16
|
||||
plot_w = w - left - right
|
||||
plot_h = h - top - bottom
|
||||
ny, nx = Z.shape
|
||||
cw = plot_w / nx
|
||||
ch = plot_h / ny
|
||||
valid = Z[~np.isnan(Z)]
|
||||
vmin = float(np.min(valid)) if valid.size else 0
|
||||
vmax = float(np.max(valid)) if valid.size else 1
|
||||
if vmax == vmin:
|
||||
vmax = vmin + 1
|
||||
|
||||
def color(v):
|
||||
if np.isnan(v):
|
||||
return "#f3f4f6"
|
||||
r = (v - vmin) / (vmax - vmin)
|
||||
if r < 0.5:
|
||||
t = r * 2
|
||||
return f"rgb({int(220-60*t)},{int(60+160*t)},{int(60+40*t)})"
|
||||
else:
|
||||
t = (r - 0.5) * 2
|
||||
return f"rgb({int(160-100*t)},{int(220-40*t)},{int(100-40*t)})"
|
||||
|
||||
cells = ""
|
||||
for j in range(ny):
|
||||
for i in range(nx):
|
||||
v = Z[j, i]
|
||||
x = left + i * cw
|
||||
y = top + (ny - 1 - j) * ch
|
||||
vstr = f"{v:.3f}" if not np.isnan(v) else "—"
|
||||
cells += (f'<rect x="{x:.1f}" y="{y:.1f}" width="{cw:.1f}" height="{ch:.1f}" '
|
||||
f'fill="{color(v)}" stroke="#fff" stroke-width="0.5">'
|
||||
f'<title>{xticks[i]}, {yticks[j]}: {vstr}</title></rect>')
|
||||
if not np.isnan(v):
|
||||
cells += f'<text x="{x+cw/2:.1f}" y="{y+ch/2+3:.1f}" font-size="9" fill="#111827" text-anchor="middle">{v:.2f}</text>'
|
||||
|
||||
# 高原边框
|
||||
plat_svg = ""
|
||||
palette = ["#1e3a8a", "#7c3aed", "#0f766e"]
|
||||
for k, p in enumerate(plateaus[:3]):
|
||||
y0, y1 = p["y_range"]
|
||||
x0, x1 = p["x_range"]
|
||||
rx = left + x0 * cw
|
||||
ry = top + (ny - 1 - y1) * ch
|
||||
rw = (x1 - x0 + 1) * cw
|
||||
rh = (y1 - y0 + 1) * ch
|
||||
col = palette[k % len(palette)]
|
||||
plat_svg += (f'<rect x="{rx:.1f}" y="{ry:.1f}" width="{rw:.1f}" height="{rh:.1f}" '
|
||||
f'fill="none" stroke="{col}" stroke-width="2.5" rx="2"/>')
|
||||
plat_svg += f'<text x="{rx+4:.1f}" y="{ry+13:.1f}" font-size="10" fill="{col}" font-weight="bold">P{k+1}</text>'
|
||||
|
||||
# 轴标签
|
||||
xlabs = "".join(
|
||||
f'<text x="{left+(i+0.5)*cw:.1f}" y="{h-bottom+14:.1f}" font-size="9" fill="#374151" text-anchor="middle">{xticks[i]}</text>'
|
||||
for i in range(nx)
|
||||
)
|
||||
ylabs = "".join(
|
||||
f'<text x="{left-6:.1f}" y="{top+(ny-0.5-j)*ch+3:.1f}" font-size="9" fill="#374151" text-anchor="end">{yticks[j]}</text>'
|
||||
for j in range(ny)
|
||||
)
|
||||
# 图例
|
||||
leg = ""
|
||||
for i in range(6):
|
||||
r = i / 5
|
||||
v = vmin + r * (vmax - vmin)
|
||||
leg += (f'<rect x="{left+i*30:.0f}" y="18" width="30" height="10" fill="{color(v)}"/>'
|
||||
f'<text x="{left+i*30+15:.0f}" y="42" font-size="8" fill="#374151" text-anchor="middle">{v:.2f}</text>')
|
||||
|
||||
return (f'<svg viewBox="0 0 {w} {h}" class="chart">'
|
||||
f'<text x="{w//2}" y="12" font-size="11" fill="#374151" text-anchor="middle">'
|
||||
f'{html.escape(metric)} 响应面热力图（{html.escape(xname)} × {html.escape(yname)}）</text>'
|
||||
f'{leg}{cells}{plat_svg}{xlabs}{ylabs}'
|
||||
f'<text x="{w//2}" y="{h-4}" font-size="10" fill="#374151" text-anchor="middle">{html.escape(xname)} →</text>'
|
||||
f'<text x="14" y="{h//2}" font-size="10" fill="#374151" text-anchor="middle" transform="rotate(-90 14 {h//2})">{html.escape(yname)} →</text>'
|
||||
f'</svg>')
|
||||
|
||||
|
||||
def svg_surface_iso(
|
||||
Z: np.ndarray, xticks: List, yticks: List, metric: str,
|
||||
w: int = 560, h: int = 420,
|
||||
) -> str:
|
||||
"""3D 等距投影曲面（四边形网格，画家算法排序）。"""
|
||||
if Z.size == 0:
|
||||
return ""
|
||||
Zf = np.where(np.isnan(Z), 0.0, Z)
|
||||
ny, nx = Zf.shape
|
||||
vmin = float(Zf.min())
|
||||
vmax = float(Zf.max())
|
||||
if vmax == vmin:
|
||||
vmax = vmin + 1
|
||||
|
||||
cx, cy = w / 2, h * 0.62
|
||||
scale_xy = min((w - 80) / (nx + ny), 36)
|
||||
scale_z = (h * 0.45) / (vmax - vmin)
|
||||
ang = 0.55 # 投影角
|
||||
|
||||
def proj(i, j, z):
|
||||
# i 列(x), j 行(y)
|
||||
px = cx + (i - j) * scale_xy * 0.9
|
||||
py = cy + (i + j) * scale_xy * ang - (z - vmin) * scale_z
|
||||
return px, py
|
||||
|
||||
quads = []
|
||||
for j in range(ny - 1):
|
||||
for i in range(nx - 1):
|
||||
z = Zf[j, i]
|
||||
# 颜色
|
||||
r = (z - vmin) / (vmax - vmin)
|
||||
if r < 0.5:
|
||||
t = r * 2
|
||||
col = f"rgb({int(220-60*t)},{int(60+160*t)},{int(60+40*t)})"
|
||||
else:
|
||||
t = (r - 0.5) * 2
|
||||
col = f"rgb({int(160-100*t)},{int(220-40*t)},{int(100-40*t)})"
|
||||
p1 = proj(i, j, Zf[j, i])
|
||||
p2 = proj(i + 1, j, Zf[j, i + 1])
|
||||
p3 = proj(i + 1, j + 1, Zf[j + 1, i + 1])
|
||||
p4 = proj(i, j + 1, Zf[j + 1, i])
|
||||
depth = i + j # 越大越靠前
|
||||
pts = " ".join(f"{x:.1f},{y:.1f}" for x, y in [p1, p2, p3, p4])
|
||||
quads.append((depth, f'<polygon points="{pts}" fill="{col}" stroke="#1f2937" stroke-width="0.3" opacity="0.92"/>'))
|
||||
quads.sort(key=lambda q: -q[0]) # 远的先画
|
||||
body = "".join(q[1] for q in quads)
|
||||
return (f'<svg viewBox="0 0 {w} {h}" class="chart">'
|
||||
f'<text x="{w//2}" y="16" font-size="11" fill="#374151" text-anchor="middle">{html.escape(metric)} 3D 响应面（等距投影）</text>'
|
||||
f'{body}</svg>')
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# 5. HTML 报告
|
||||
# =========================================================================== #
|
||||
def build_html(
|
||||
df: pd.DataFrame, x: str, y: str, metric: str,
|
||||
Z: np.ndarray, xticks: List, yticks: List,
|
||||
plateaus: List[Dict[str, Any]], of_score: Dict[str, Any],
|
||||
) -> str:
|
||||
heat = svg_heatmap(Z, xticks, yticks, metric, x, y, plateaus)
|
||||
surf = svg_surface_iso(Z, xticks, yticks, metric)
|
||||
|
||||
# 高原表
|
||||
if plateaus:
|
||||
plat_rows = ""
|
||||
for k, p in enumerate(plateaus[:6]):
|
||||
xr = f"{xticks[p['x_range'][0]]}~{xticks[p['x_range'][1]]}"
|
||||
yr = f"{yticks[p['y_range'][0]]}~{yticks[p['y_range'][1]]}"
|
||||
plat_rows += (f"<tr><td>P{k+1}</td><td>{p['size']}</td><td>{xr}</td><td>{yr}</td>"
|
||||
f"<td>{p['metric_mean']:.3f}</td><td>{p['metric_max']:.3f}</td></tr>")
|
||||
plat_table = (
|
||||
"<table class='data'><thead><tr><th>高原</th><th>面积(格数)</th>"
|
||||
f"<th>{html.escape(x)} 范围</th><th>{html.escape(y)} 范围</th>"
|
||||
f"<th>{html.escape(metric)} 均值</th><th>{html.escape(metric)} 峰值</th></tr></thead><tbody>"
|
||||
+ plat_rows + "</tbody></table>"
|
||||
)
|
||||
else:
|
||||
plat_table = "<p class='muted'>未检测到稳健高原（所有高分点都是孤立单点，过拟合风险高）。</p>"
|
||||
|
||||
iso_str = (f"{of_score['peak_isolation_ratio']:.2f}" if of_score and not np.isnan(of_score.get('peak_isolation_ratio', np.nan)) else "—")
|
||||
rob_str = (f"{of_score['robust_score']:.2f}" if of_score and not np.isnan(of_score.get('robust_score', np.nan)) else "—")
|
||||
cards = (
|
||||
"<div class='cards'>"
|
||||
f"<div class='card'><div class='k'>全局峰值 {html.escape(metric)}</div><div class='v'>{of_score['peak']:.3f}</div></div>"
|
||||
f"<div class='card'><div class='k'>峰值邻域均值/峰值</div><div class='v'>{iso_str}</div></div>"
|
||||
f"<div class='card'><div class='k'>高原覆盖率</div><div class='v'>{of_score['plateau_coverage']*100:.0f}%</div></div>"
|
||||
f"<div class='card'><div class='k'>稳健性评分(0~1)</div><div class='v'>{rob_str}</div></div>"
|
||||
"</div>"
|
||||
)
|
||||
|
||||
interp = ""
|
||||
if of_score and not np.isnan(of_score.get('robust_score', np.nan)):
|
||||
r = of_score['robust_score']
|
||||
if r < 0.4:
|
||||
interp = "<div class='warn'>稳健性评分低：最优参数是孤立尖峰，邻域快速退化，过拟合风险高。应在高原中心选参而非峰值点。</div>"
|
||||
elif r > 0.7:
|
||||
interp = "<div class='note'>稳健性评分高：存在宽阔高原，参数对扰动不敏感，泛化性好。建议在最大高原中心选参。</div>"
|
||||
else:
|
||||
interp = "<div class='note'>稳健性中等：有一定高原但峰值仍较突出，建议在高原内偏离峰值一点的位置选参以留安全边际。</div>"
|
||||
|
||||
recommend = ""
|
||||
if plateaus:
|
||||
p1 = plateaus[0]
|
||||
cx_v = xticks[(p1['x_range'][0] + p1['x_range'][1]) // 2]
|
||||
cy_v = yticks[(p1['y_range'][0] + p1['y_range'][1]) // 2]
|
||||
recommend = (f"<div class='note'><b>推荐参数（最大高原中心）</b>："
|
||||
f"{html.escape(x)} = <code>{cx_v}</code>，{html.escape(y)} = <code>{cy_v}</code>。"
|
||||
f"该位置周围 {p1['size']} 个网格点 {html.escape(metric)} 均值 {p1['metric_mean']:.3f}，"
|
||||
f"对参数扰动不敏感。</div>")
|
||||
|
||||
# 顶 10 单点
|
||||
top_df = df.nlargest(10, metric) if metric in df.columns else df.head(0)
|
||||
top_rows = ""
|
||||
for _, r in top_df.iterrows():
|
||||
top_rows += "<tr>" + "".join(f"<td>{r[c]}</td>" for c in top_df.columns) + "</tr>"
|
||||
top_table = ("<h3>Top-10 单点峰值（对比用，单点 ≠ 稳健）</h3>"
|
||||
"<table class='data'><thead><tr>" + "".join(f"<th>{html.escape(c)}</th>" for c in top_df.columns) + "</tr></thead><tbody>"
|
||||
+ top_rows + "</tbody></table>") if top_rows else ""
|
||||
|
||||
return f"""<!doctype html><html lang="zh-CN"><head><meta charset="utf-8">
|
||||
<title>参数敏感度扫描 — {html.escape(metric)}</title>
|
||||
<style>
|
||||
body {{ font-family:-apple-system,"Microsoft YaHei",sans-serif; background:#f8fafc; color:#111827; margin:0; padding:20px;}}
|
||||
.wrap {{ max-width:1100px; margin:0 auto;}}
|
||||
h1,h2,h3 {{ color:#1e3a8a;}} h1 {{ border-bottom:3px solid #1e3a8a; padding-bottom:8px;}}
|
||||
table.data {{ border-collapse:collapse; width:100%; background:#fff; margin:8px 0 16px; font-size:12px;}}
|
||||
table.data th, table.data td {{ border:1px solid #e2e8f0; padding:5px 7px; text-align:left;}}
|
||||
table.data th {{ background:#f1f5f9;}}
|
||||
.muted {{ color:#6b7280; font-size:12px;}}
|
||||
code {{ background:#f1f5f9; padding:1px 5px; border-radius:3px;}}
|
||||
.chart {{ width:100%; height:auto; background:#fff; border:1px solid #e2e8f0; border-radius:6px; margin:8px 0;}}
|
||||
.grid2 {{ display:grid; grid-template-columns:1fr 1fr; gap:12px;}}
|
||||
.cards {{ display:grid; grid-template-columns:repeat(auto-fit,minmax(200px,1fr)); gap:12px; margin:16px 0;}}
|
||||
.card {{ background:#fff; padding:14px; border-radius:8px; border:1px solid #e2e8f0;}}
|
||||
.card .k {{ color:#6b7280; font-size:12px;}} .card .v {{ font-size:18px; font-weight:600; margin-top:4px;}}
|
||||
.note {{ background:#fef3c7; border-left:4px solid #f59e0b; padding:10px 14px; margin:12px 0; border-radius:4px;}}
|
||||
.warn {{ background:#fee2e2; border-left:4px solid #dc2626; padding:10px 14px; margin:12px 0; border-radius:4px;}}
|
||||
footer {{ color:#6b7280; font-size:11px; margin-top:32px; text-align:center;}}
|
||||
</style></head><body><div class="wrap">
|
||||
<h1>参数敏感度扫描报告</h1>
|
||||
<p class='muted'>指标: <code>{html.escape(metric)}</code> 参数轴: {html.escape(x)} × {html.escape(y)} 样本数: {len(df)}</p>
|
||||
{cards}
|
||||
{interp}
|
||||
{recommend}
|
||||
<h2>1. 响应面热力图（找高原用）</h2>
|
||||
{heat}
|
||||
<p class='muted'>粗框 P1/P2/P3 = 检测到的高原（连通高分区域）。高原越宽 = 参数越稳健。</p>
|
||||
<h2>2. 3D 等距投影曲面</h2>
|
||||
{surf}
|
||||
<h2>3. 高原清单</h2>
|
||||
{plat_table}
|
||||
{top_table}
|
||||
<footer>由 param_scan.py 生成 · 高原检测阈值=80分位 · {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M')}</footer>
|
||||
</div></body></html>"""
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# 6. demo（合成数据演示）
|
||||
# =========================================================================== #
|
||||
def _demo_df() -> pd.DataFrame:
|
||||
"""合成一个有高原 + 一个孤立尖峰的数据集。"""
|
||||
xs = list(range(5, 35, 5)) # FastMA
|
||||
ys = list(range(20, 70, 10)) # SlowMA
|
||||
rows = []
|
||||
for x in xs:
|
||||
for y in ys:
|
||||
# 主响应：FastMA 10-20, SlowMA 30-50 是高原（PF~1.5）
|
||||
base = 1.5 - 0.02 * abs(x - 15) - 0.015 * abs(y - 40)
|
||||
# 加一个孤立尖峰在 (25, 60)
|
||||
if x == 25 and y == 60:
|
||||
base = 2.3
|
||||
base += np.random.default_rng(x * 100 + y).normal(0, 0.05)
|
||||
rows.append({"FastMA": x, "SlowMA": y, "PF": round(base, 3)})
|
||||
return pd.DataFrame(rows)
|
||||
|
||||
|
||||
def _demo_template() -> str:
|
||||
"""合成一个最小 .set 模板，供 demo 写出 .set 用。"""
|
||||
return (
|
||||
"; inputs\n"
|
||||
"FastMA=10||1||1||100||N\n"
|
||||
"SlowMA=40||1||1||500||N\n"
|
||||
"StopLoss=100||1||1||10000||N\n"
|
||||
"TakeProfit=200||1||1||10000||N\n"
|
||||
)
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# 7. 写出优化后的 .set 文件
|
||||
# =========================================================================== #
|
||||
def _pick_optimal_params(
|
||||
df: pd.DataFrame, x: str, y: str, metric: str,
|
||||
Z: np.ndarray, xticks: List, yticks: List,
|
||||
plateaus: List[Dict[str, Any]],
|
||||
strategy: str,
|
||||
) -> Tuple[Dict[str, Any], str]:
|
||||
"""
|
||||
根据策略选出最优参数。
|
||||
返回 (参数字典, 来源说明)。
|
||||
参数字典包含 x/y 两个轴的值（高原/峰值），或单点峰值的全套参数。
|
||||
"""
|
||||
if strategy == "plateau":
|
||||
if not plateaus:
|
||||
# 回退到峰值
|
||||
return _pick_optimal_params(df, x, y, metric, Z, xticks, yticks, plateaus, "peak")
|
||||
p1 = plateaus[0]
|
||||
cx_v = xticks[(p1["x_range"][0] + p1["x_range"][1]) // 2]
|
||||
cy_v = yticks[(p1["y_range"][0] + p1["y_range"][1]) // 2]
|
||||
return ({x: cx_v, y: cy_v},
|
||||
f"最大高原中心（面积 {p1['size']}，{metric} 均值 {p1['metric_mean']:.3f}）")
|
||||
|
||||
if strategy == "peak":
|
||||
# 单点峰值：取 df 中 metric 最高的行，返回该行全套参数
|
||||
best_row = df.loc[df[metric].idxmax()]
|
||||
# 排除所有计算列与 metric 本身，剩下的是 manifest 里的参数列
|
||||
exclude = {"idx", "set_file", "net", "pf", "win_rate", "max_dd_pct", "sharpe",
|
||||
metric.lower(), metric.upper(), metric}
|
||||
param_cols = [c for c in df.columns if c not in exclude]
|
||||
params = {c: best_row[c] for c in param_cols}
|
||||
return params, f"单点峰值（{metric}={best_row[metric]:.3f}，来源 {best_row.get('set_file','—')}）"
|
||||
|
||||
raise ValueError(f"未知策略: {strategy}")
|
||||
|
||||
|
||||
def write_optimized_set(
|
||||
template_path: Optional[str],
|
||||
template_text: Optional[str],
|
||||
params: Dict[str, Any],
|
||||
out_path: str,
|
||||
source_note: str,
|
||||
) -> str:
|
||||
"""
|
||||
把最优参数套到模板上，写出 .set 文件。
|
||||
template_path 与 template_text 二选一（后者优先，供 demo 用）。
|
||||
返回写入的文件路径。
|
||||
"""
|
||||
if template_text is None:
|
||||
if template_path is None or not os.path.exists(template_path):
|
||||
raise FileNotFoundError(f"模板未提供/不存在: {template_path}")
|
||||
with open(template_path, "r", encoding="utf-8", errors="replace") as f:
|
||||
template_text = f.read()
|
||||
|
||||
# 全部转字符串
|
||||
params_str = {k: str(v) for k, v in params.items()}
|
||||
new_set = _apply_params_to_template(template_text, params_str)
|
||||
# 在文件头加注释说明来源
|
||||
header = f"; === 由 param_scan.py 优化生成 ===\n; 来源: {source_note}\n; 生成时间: {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M')}\n; 应用参数: {params_str}\n\n"
|
||||
with open(out_path, "w", encoding="utf-8") as f:
|
||||
f.write(header + new_set)
|
||||
return out_path
|
||||
|
||||
|
||||
# =========================================================================== #
|
||||
# CLI
|
||||
# =========================================================================== #
|
||||
def main(argv: List[str]) -> int:
|
||||
ap = argparse.ArgumentParser(description="参数敏感度扫描")
|
||||
sub = ap.add_subparsers(dest="cmd", required=True)
|
||||
|
||||
p_gen = sub.add_parser("gen-set", help="生成 .set 网格 + 批处理脚本")
|
||||
p_gen.add_argument("template")
|
||||
p_gen.add_argument("grid_json")
|
||||
p_gen.add_argument("out_dir")
|
||||
p_gen.add_argument("--ea", default="EA")
|
||||
p_gen.add_argument("--symbol", default="XAUUSD")
|
||||
p_gen.add_argument("--period", default="M1")
|
||||
p_gen.add_argument("--from", dest="date_from", default="2024-01-01")
|
||||
p_gen.add_argument("--to", dest="date_to", default="2024-06-01")
|
||||
p_gen.add_argument("--terminal", default=r"C:\Program Files\MetaTrader 5\terminal64.exe")
|
||||
|
||||
p_an = sub.add_parser("analyze", help="分析扫描结果，画响应面")
|
||||
p_an.add_argument("reports_dir")
|
||||
p_an.add_argument("manifest")
|
||||
p_an.add_argument("--x", required=True)
|
||||
p_an.add_argument("--y", required=True)
|
||||
p_an.add_argument("--metric", default="pf")
|
||||
p_an.add_argument("--template", default=None,
|
||||
help="原始 .set 模板路径（用于生成优化后 .set）")
|
||||
p_an.add_argument("--out-set", default=None,
|
||||
help="优化后 .set 输出路径（需配合 --template）")
|
||||
p_an.add_argument("--strategy", choices=["peak", "plateau"], default="plateau",
|
||||
help="选参策略：peak=单点峰值(激进)，plateau=高原中心(稳健,默认)")
|
||||
|
||||
p_demo = sub.add_parser("demo", help="用合成数据演示")
|
||||
p_demo.add_argument("--out-set", default=None,
|
||||
help="演示用合成模板写出优化后 .set 的路径")
|
||||
p_demo.add_argument("--strategy", choices=["peak", "plateau"], default="plateau",
|
||||
help="选参策略：peak=单点峰值(激进)，plateau=高原中心(稳健,默认)")
|
||||
|
||||
args = ap.parse_args(argv)
|
||||
os.makedirs(OUT_DIR, exist_ok=True)
|
||||
|
||||
if args.cmd == "gen-set":
|
||||
with open(args.grid_json, "r", encoding="utf-8") as f:
|
||||
grid = json.load(f)
|
||||
generate_set_files(
|
||||
args.template, grid, args.out_dir,
|
||||
ea_name=args.ea, symbol=args.symbol, period=args.period,
|
||||
date_from=args.date_from, date_to=args.date_to,
|
||||
terminal_path=args.terminal,
|
||||
)
|
||||
return 0
|
||||
|
||||
if args.cmd == "analyze":
|
||||
df = load_scan_results(args.reports_dir, args.manifest)
|
||||
if df.empty:
|
||||
print("未加载到任何报告数据，检查 reports_dir 与 manifest")
|
||||
return 1
|
||||
if args.x not in df.columns or args.y not in df.columns:
|
||||
print(f"参数 {args.x}/{args.y} 不在 manifest，可用列: {list(df.columns)}")
|
||||
return 1
|
||||
if args.metric not in df.columns:
|
||||
print(f"指标 {args.metric} 不可用，可用: {['net','pf','win_rate','max_dd_pct','sharpe']}")
|
||||
return 1
|
||||
Z, xt, yt = build_pivot(df, args.x, args.y, args.metric)
|
||||
plateaus = detect_plateaus(Z)
|
||||
of = overfit_score(Z, plateaus)
|
||||
html_doc = build_html(df, args.x, args.y, args.metric, Z, xt, yt, plateaus, of)
|
||||
out = os.path.join(OUT_DIR, "param_scan.html")
|
||||
with open(out, "w", encoding="utf-8") as f:
|
||||
f.write(html_doc)
|
||||
print(f"HTML 报告: {out}")
|
||||
|
||||
# 写出优化后的 .set
|
||||
if args.out_set:
|
||||
if not args.template:
|
||||
print("ERROR: --out-set 需要配合 --template <原始set路径>")
|
||||
return 1
|
||||
params, note = _pick_optimal_params(
|
||||
df, args.x, args.y, args.metric, Z, xt, yt, plateaus, args.strategy
|
||||
)
|
||||
write_optimized_set(args.template, None, params, args.out_set, note)
|
||||
print(f"优化后 .set 已写出: {args.out_set}")
|
||||
print(f" 策略: {args.strategy} 来源: {note}")
|
||||
print(f" 应用参数: {params}")
|
||||
return 0
|
||||
|
||||
if args.cmd == "demo":
|
||||
df = _demo_df()
|
||||
Z, xt, yt = build_pivot(df, "FastMA", "SlowMA", "PF")
|
||||
plateaus = detect_plateaus(Z)
|
||||
of = overfit_score(Z, plateaus)
|
||||
html_doc = build_html(df, "FastMA", "SlowMA", "PF", Z, xt, yt, plateaus, of)
|
||||
out = os.path.join(OUT_DIR, "param_scan.html")
|
||||
with open(out, "w", encoding="utf-8") as f:
|
||||
f.write(html_doc)
|
||||
print(f"HTML 报告: {out}")
|
||||
print(f"检测到 {len(plateaus)} 个高原")
|
||||
|
||||
if args.out_set:
|
||||
params, note = _pick_optimal_params(
|
||||
df, "FastMA", "SlowMA", "PF", Z, xt, yt, plateaus, args.strategy
|
||||
)
|
||||
write_optimized_set(None, _demo_template(), params, args.out_set, note)
|
||||
print(f"优化后 .set 已写出(用合成模板): {args.out_set}")
|
||||
print(f" 来源: {note}")
|
||||
print(f" 应用参数: {params}")
|
||||
return 0
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main(sys.argv[1:]))
|
||||
+1076
File diff suppressed because it is too large
Load Diff
+367
@@ -0,0 +1,367 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Walk-Forward 滚动验证
|
||||
====================
|
||||
将单份报告的逐笔交易按时间切成多个滚动窗口，每个窗口内
|
||||
前段做 IS（参数评估段）、后段做 OOS（样本外验证段），
|
||||
计算每个窗口的 IS/OOS 指标，再汇总：
|
||||
|
||||
- Walk-Forward Efficiency (WFE) = Σ OOS净盈利 / Σ IS净盈利
|
||||
- OOS 盈利窗口占比
|
||||
- IS PF 与 OOS PF 的相关性（高=过拟合低；低=参数不稳健）
|
||||
- 滚动 IS/OOS PF 曲线
|
||||
|
||||
相比单次 IS/OSS 划分，能反映参数在不同行情段的稳健性。
|
||||
|
||||
可独立运行，也可被 run_analysis.py 导入集成进主报告。
|
||||
|
||||
用法:
|
||||
python walk_forward.py <report.xlsx> [--is-days 60] [--oos-days 30] [--step-days 30]
|
||||
python walk_forward.py --demo
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import glob
|
||||
import html
|
||||
import os
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Dict, List, Optional, Tuple
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
import mt5_report_parser as mp
|
||||
|
||||
OUT_DIR = "output"
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 单段指标
|
||||
# --------------------------------------------------------------------------- #
|
||||
def _seg_metrics(net: pd.Series) -> Dict[str, float]:
|
||||
n = len(net)
|
||||
if n == 0:
|
||||
return {"n": 0, "net": 0.0, "pf": 0.0, "win": 0.0, "dd": 0.0, "exp": 0.0}
|
||||
wins = net[net > 0]
|
||||
losses = net[net <= 0]
|
||||
gp = wins.sum()
|
||||
gl = -losses.sum()
|
||||
pf = gp / gl if gl > 0 else np.inf
|
||||
eq = net.cumsum()
|
||||
dd = float((eq - eq.cummax()).min())
|
||||
return {
|
||||
"n": int(n),
|
||||
"net": float(net.sum()),
|
||||
"pf": float(pf),
|
||||
"win": float(len(wins) / n * 100),
|
||||
"dd": dd,
|
||||
"exp": float(net.mean()),
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class WFWindow:
|
||||
idx: int
|
||||
is_start: pd.Timestamp
|
||||
is_end: pd.Timestamp
|
||||
oos_start: pd.Timestamp
|
||||
oos_end: pd.Timestamp
|
||||
is_metrics: Dict[str, float]
|
||||
oos_metrics: Dict[str, float]
|
||||
|
||||
|
||||
def walk_forward(
|
||||
trades: pd.DataFrame,
|
||||
is_days: int = 60,
|
||||
oos_days: int = 30,
|
||||
step_days: Optional[int] = None,
|
||||
) -> List[WFWindow]:
|
||||
"""
|
||||
滚动切窗。
|
||||
trades: 含 open_time, net_profit 的 DataFrame
|
||||
is_days / oos_days: IS/OOS 窗口天数
|
||||
step_days: 滑动步长（默认 = oos_days，即不重叠的 OOS）
|
||||
"""
|
||||
if trades is None or trades.empty:
|
||||
return []
|
||||
t = trades.sort_values("open_time").reset_index(drop=True)
|
||||
t["net_profit"] = t["net_profit"].astype(float)
|
||||
|
||||
if step_days is None:
|
||||
step_days = oos_days
|
||||
|
||||
start = t["open_time"].min()
|
||||
end = t["open_time"].max()
|
||||
total_span = (end - start).days
|
||||
win_span = is_days + oos_days
|
||||
if total_span < win_span:
|
||||
return [] # 数据不足以切出一个窗口
|
||||
|
||||
windows: List[WFWindow] = []
|
||||
cur = start
|
||||
idx = 0
|
||||
while True:
|
||||
is_s = cur
|
||||
is_e = cur + pd.Timedelta(days=is_days)
|
||||
oos_s = is_e
|
||||
oos_e = oos_s + pd.Timedelta(days=oos_days)
|
||||
if oos_e > end + pd.Timedelta(days=1):
|
||||
break
|
||||
is_mask = (t["open_time"] >= is_s) & (t["open_time"] < is_e)
|
||||
oos_mask = (t["open_time"] >= oos_s) & (t["open_time"] < oos_e)
|
||||
is_net = t.loc[is_mask, "net_profit"]
|
||||
oos_net = t.loc[oos_mask, "net_profit"]
|
||||
if len(is_net) == 0 or len(oos_net) == 0:
|
||||
cur += pd.Timedelta(days=step_days)
|
||||
continue
|
||||
windows.append(WFWindow(
|
||||
idx=idx,
|
||||
is_start=is_s, is_end=is_e,
|
||||
oos_start=oos_s, oos_end=oos_e,
|
||||
is_metrics=_seg_metrics(is_net),
|
||||
oos_metrics=_seg_metrics(oos_net),
|
||||
))
|
||||
idx += 1
|
||||
cur += pd.Timedelta(days=step_days)
|
||||
return windows
|
||||
|
||||
|
||||
def wf_summary(windows: List[WFWindow]) -> Dict[str, Any]:
|
||||
if not windows:
|
||||
return {}
|
||||
is_nets = [w.is_metrics["net"] for w in windows]
|
||||
oos_nets = [w.oos_metrics["net"] for w in windows]
|
||||
is_pfs = [w.is_metrics["pf"] for w in windows if not np.isinf(w.is_metrics["pf"])]
|
||||
oos_pfs = [w.oos_metrics["pf"] for w in windows if not np.isinf(w.oos_metrics["pf"])]
|
||||
wfe = sum(oos_nets) / sum(is_nets) if sum(is_nets) != 0 else np.nan
|
||||
oos_profitable = sum(1 for n in oos_nets if n > 0)
|
||||
# IS-OOS PF 相关性
|
||||
if len(is_pfs) > 2 and len(oos_pfs) > 2:
|
||||
# 配对（取有限值的交集）
|
||||
pairs = [(w.is_metrics["pf"], w.oos_metrics["pf"]) for w in windows
|
||||
if not np.isinf(w.is_metrics["pf"]) and not np.isinf(w.oos_metrics["pf"])]
|
||||
if len(pairs) > 2:
|
||||
arr = np.array(pairs)
|
||||
corr = float(np.corrcoef(arr[:, 0], arr[:, 1])[0, 1])
|
||||
else:
|
||||
corr = np.nan
|
||||
else:
|
||||
corr = np.nan
|
||||
return {
|
||||
"n_windows": len(windows),
|
||||
"wfe": float(wfe),
|
||||
"oos_profitable": oos_profitable,
|
||||
"oos_profitable_pct": float(oos_profitable / len(windows) * 100),
|
||||
"sum_is_net": float(sum(is_nets)),
|
||||
"sum_oos_net": float(sum(oos_nets)),
|
||||
"avg_is_pf": float(np.mean(is_pfs)) if is_pfs else 0.0,
|
||||
"avg_oos_pf": float(np.mean(oos_pfs)) if oos_pfs else 0.0,
|
||||
"is_oos_pf_corr": corr,
|
||||
"max_oos_dd": float(min(w.oos_metrics["dd"] for w in windows)),
|
||||
}
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# SVG
|
||||
# --------------------------------------------------------------------------- #
|
||||
def svg_wf_curve(windows: List[WFWindow], w: int = 760, h: int = 260) -> str:
|
||||
if not windows:
|
||||
return ""
|
||||
top, bottom, left, right = 24, 36, 46, 16
|
||||
plot_h = h - top - bottom
|
||||
plot_w = w - left - right
|
||||
is_pfs = [w.is_metrics["pf"] for w in windows]
|
||||
oos_pfs = [w.oos_metrics["pf"] for w in windows]
|
||||
vals = is_pfs + oos_pfs + [1.0]
|
||||
vmin = min(vals)
|
||||
vmax = max(vals)
|
||||
if vmax == vmin:
|
||||
vmax = vmin + 1
|
||||
vmax = max(vmax, 1.0)
|
||||
vmin = min(vmin, 0.0)
|
||||
|
||||
def xy(i, v):
|
||||
x = left + i / max(1, len(windows) - 1) * plot_w
|
||||
y = top + plot_h - (v - vmin) / (vmax - vmin) * plot_h
|
||||
return x, y
|
||||
|
||||
def poly(vals_, color):
|
||||
pts = " ".join(f"{x:.1f},{y:.1f}" for x, y in [xy(i, v) for i, v in enumerate(vals_)])
|
||||
return f'<polyline points="{pts}" fill="none" stroke="{color}" stroke-width="1.6"><title></title></polyline>'
|
||||
|
||||
# 1.0 参考线
|
||||
one_y = top + plot_h - (1.0 - vmin) / (vmax - vmin) * plot_h
|
||||
grid = ""
|
||||
for frac in (0, 0.25, 0.5, 0.75, 1.0):
|
||||
yv = vmin + frac * (vmax - vmin)
|
||||
gy = top + plot_h - frac * plot_h
|
||||
grid += (f'<line x1="{left}" y1="{gy:.1f}" x2="{w-right}" y2="{gy:.1f}" stroke="#e5e7eb" stroke-width="0.5"/>'
|
||||
f'<text x="{left-6}" y="{gy+3:.1f}" font-size="9" fill="#6b7280" text-anchor="end">{yv:.2f}</text>')
|
||||
one_line = f'<line x1="{left}" y1="{one_y:.1f}" x2="{w-right}" y2="{one_y:.1f}" stroke="#9ca3af" stroke-dasharray="3,3" stroke-width="0.8"/>'
|
||||
|
||||
# 窗口标签（仅首/中/尾）
|
||||
xlabs = ""
|
||||
for i in [0, len(windows) // 2, len(windows) - 1]:
|
||||
x = left + i / max(1, len(windows) - 1) * plot_w
|
||||
xlabs += f'<text x="{x:.1f}" y="{h-bottom+14:.1f}" font-size="9" fill="#374151" text-anchor="middle">#{windows[i].idx}</text>'
|
||||
|
||||
return (
|
||||
f'<svg viewBox="0 0 {w} {h}" class="chart">'
|
||||
f'{grid}{one_line}'
|
||||
f'{poly(is_pfs, "#2563eb")}'
|
||||
f'{poly(oos_pfs, "#dc2626")}'
|
||||
f'{xlabs}'
|
||||
f'<text x="{w//2}" y="16" font-size="11" fill="#374151" text-anchor="middle">滚动窗口 IS vs OOS 盈利因子 (PF=1 为盈亏平衡线)</text>'
|
||||
f'<text x="{w-right}" y="32" font-size="10" fill="#2563eb" text-anchor="end">IS PF</text>'
|
||||
f'<text x="{w-right}" y="46" font-size="10" fill="#dc2626" text-anchor="end">OOS PF</text>'
|
||||
f'<text x="{w//2}" y="{h-4}" font-size="9" fill="#6b7280" text-anchor="middle">窗口序号 →</text>'
|
||||
f'</svg>'
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# HTML 片段（供 run_analysis.py 嵌入）
|
||||
# --------------------------------------------------------------------------- #
|
||||
def wf_html_fragment(windows: List[WFWindow]) -> str:
|
||||
if not windows:
|
||||
return "<p class='muted'>数据不足以切出 Walk-Forward 窗口。</p>"
|
||||
s = wf_summary(windows)
|
||||
rows = ""
|
||||
for w in windows:
|
||||
oos_cls = " class='pos'" if w.oos_metrics["net"] > 0 else " class='neg'"
|
||||
rows += (
|
||||
f"<tr><td>#{w.idx}</td>"
|
||||
f"<td>{w.is_start:%Y-%m-%d}~{w.is_end:%Y-%m-%d}</td>"
|
||||
f"<td>{w.oos_start:%Y-%m-%d}~{w.oos_end:%Y-%m-%d}</td>"
|
||||
f"<td>{w.is_metrics['n']}</td>"
|
||||
f"<td>{w.is_metrics['net']:+.2f}</td>"
|
||||
f"<td>{w.is_metrics['pf']:.2f}</td>"
|
||||
f"<td>{w.is_metrics['win']:.1f}%</td>"
|
||||
f"<td>{w.oos_metrics['n']}</td>"
|
||||
f"<td{oos_cls}>{w.oos_metrics['net']:+.2f}</td>"
|
||||
f"<td>{w.oos_metrics['pf']:.2f}</td>"
|
||||
f"<td>{w.oos_metrics['win']:.1f}%</td>"
|
||||
f"<td>{w.oos_metrics['dd']:.2f}</td></tr>"
|
||||
)
|
||||
table = (
|
||||
"<table class='data'><thead><tr>"
|
||||
"<th>窗口</th><th>IS区间</th><th>OOS区间</th>"
|
||||
"<th>IS笔</th><th>IS净</th><th>IS PF</th><th>IS胜率</th>"
|
||||
"<th>OOS笔</th><th>OOS净</th><th>OOS PF</th><th>OOS胜率</th><th>OOS回撤</th>"
|
||||
"</tr></thead><tbody>" + rows + "</tbody></table>"
|
||||
)
|
||||
|
||||
corr_str = f"{s['is_oos_pf_corr']:.2f}" if not np.isnan(s['is_oos_pf_corr']) else "—"
|
||||
cards = (
|
||||
f"<div class='cards'>"
|
||||
f"<div class='card'><div class='k'>窗口数</div><div class='v'>{s['n_windows']}</div></div>"
|
||||
f"<div class='card'><div class='k'>WFE (ΣOOS净/ΣIS净)</div><div class='v {'pos' if s['wfe']>0 else 'neg'}'>{s['wfe']:+.2f}</div></div>"
|
||||
f"<div class='card'><div class='k'>OOS 盈利窗口占比</div><div class='v'>{s['oos_profitable_pct']:.0f}%</div></div>"
|
||||
f"<div class='card'><div class='k'>IS-OOS PF 相关性</div><div class='v'>{corr_str}</div></div>"
|
||||
f"</div>"
|
||||
)
|
||||
|
||||
interp = ""
|
||||
if not np.isnan(s['is_oos_pf_corr']):
|
||||
if s['is_oos_pf_corr'] > 0.6:
|
||||
interp = "<p class='muted'>IS 与 OOS 的 PF 相关性较高 → 参数对行情段有一定泛化能力。</p>"
|
||||
elif s['is_oos_pf_corr'] < 0.2:
|
||||
interp = "<p class='muted'>IS 与 OOS 的 PF 相关性低 → IS 表现难以预测 OOS，参数不稳健，警惕过拟合单段行情。</p>"
|
||||
else:
|
||||
interp = "<p class='muted'>IS/OOS PF 中等相关性，泛化能力中等。</p>"
|
||||
if s['wfe'] < 0:
|
||||
interp += "<div class='warn'>WFE 为负：OOS 累计亏损。即使 IS 段盈利，策略也未能泛化。</div>"
|
||||
elif s['wfe'] > 0.5:
|
||||
interp += "<div class='note'>WFE > 0.5：OOS 能保留 IS 的一半以上盈利，泛化较好。</div>"
|
||||
|
||||
return cards + svg_wf_curve(windows) + table + interp
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# 独立 HTML 报告
|
||||
# --------------------------------------------------------------------------- #
|
||||
def build_standalone_html(name: str, windows: List[WFWindow]) -> str:
|
||||
frag = wf_html_fragment(windows)
|
||||
return f"""<!doctype html><html lang="zh-CN"><head><meta charset="utf-8">
|
||||
<title>Walk-Forward 报告 — {html.escape(name)}</title>
|
||||
<style>
|
||||
body {{ font-family: -apple-system, "Microsoft YaHei", sans-serif; background:#f8fafc; color:#111827; margin:0; padding:20px;}}
|
||||
.wrap {{ max-width:1100px; margin:0 auto;}}
|
||||
h1 {{ color:#1e3a8a; border-bottom:3px solid #1e3a8a; padding-bottom:8px;}}
|
||||
table.data {{ border-collapse:collapse; width:100%; background:#fff; margin:8px 0 16px; font-size:12px;}}
|
||||
table.data th, table.data td {{ border:1px solid #e2e8f0; padding:5px 7px; text-align:left;}}
|
||||
table.data th {{ background:#f1f5f9;}}
|
||||
td.pos, tr.pos td {{ color:#16a34a;}} td.neg, tr.neg td {{ color:#dc2626;}}
|
||||
.muted {{ color:#6b7280; font-size:12px;}}
|
||||
.cards {{ display:grid; grid-template-columns:repeat(auto-fit,minmax(200px,1fr)); gap:12px; margin:16px 0;}}
|
||||
.card {{ background:#fff; padding:14px; border-radius:8px; border:1px solid #e2e8f0;}}
|
||||
.card .k {{ color:#6b7280; font-size:12px;}} .card .v {{ font-size:20px; font-weight:600; margin-top:4px;}}
|
||||
.card .v.pos {{ color:#16a34a;}} .card .v.neg {{ color:#dc2626;}}
|
||||
.chart {{ width:100%; height:auto; background:#fff; border:1px solid #e2e8f0; border-radius:6px;}}
|
||||
.note {{ background:#fef3c7; border-left:4px solid #f59e0b; padding:10px 14px; margin:12px 0; border-radius:4px;}}
|
||||
.warn {{ background:#fee2e2; border-left:4px solid #dc2626; padding:10px 14px; margin:12px 0; border-radius:4px;}}
|
||||
</style></head><body><div class="wrap">
|
||||
<h1>Walk-Forward 滚动验证报告</h1>
|
||||
<p class='muted'>来源: {html.escape(name)}</p>
|
||||
{frag}
|
||||
</div></body></html>"""
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# demo
|
||||
# --------------------------------------------------------------------------- #
|
||||
def _demo_trades() -> pd.DataFrame:
|
||||
rng = np.random.default_rng(7)
|
||||
n = 1200
|
||||
start = pd.Timestamp("2024-01-01")
|
||||
times = start + pd.to_timedelta(np.arange(n) * 3, unit="h")
|
||||
# 让 PF 随时间漂移，模拟参数不稳健
|
||||
drift = np.sin(np.arange(n) / 200) * 1.5
|
||||
base = rng.normal(-0.1 + drift * 0.1, 3, n)
|
||||
return pd.DataFrame({"open_time": times, "net_profit": base})
|
||||
|
||||
|
||||
# --------------------------------------------------------------------------- #
|
||||
# CLI
|
||||
# --------------------------------------------------------------------------- #
|
||||
def main(argv: List[str]) -> int:
|
||||
ap = argparse.ArgumentParser(description="Walk-Forward 滚动验证")
|
||||
ap.add_argument("report", nargs="?", help="MT5 回测报告 xlsx 路径")
|
||||
ap.add_argument("--is-days", type=int, default=60)
|
||||
ap.add_argument("--oos-days", type=int, default=30)
|
||||
ap.add_argument("--step-days", type=int, default=None)
|
||||
ap.add_argument("--demo", action="store_true")
|
||||
args = ap.parse_args(argv)
|
||||
|
||||
os.makedirs(OUT_DIR, exist_ok=True)
|
||||
|
||||
if args.demo:
|
||||
trades = _demo_trades()
|
||||
name = "demo"
|
||||
else:
|
||||
if not args.report:
|
||||
ap.error("需要报告路径或 --demo")
|
||||
rep = mp.parse_report(args.report)
|
||||
trades = rep.trades
|
||||
name = os.path.basename(args.report)
|
||||
|
||||
windows = walk_forward(trades, args.is_days, args.oos_days, args.step_days)
|
||||
if not windows:
|
||||
print("数据不足以切出窗口，试试更小的 is-days/oos-days")
|
||||
return 1
|
||||
s = wf_summary(windows)
|
||||
print(f"窗口数: {s['n_windows']} WFE: {s['wfe']:+.3f} "
|
||||
f"OOS盈利窗口: {s['oos_profitable_pct']:.0f}% "
|
||||
f"IS-OOS PF 相关性: {s['is_oos_pf_corr']}")
|
||||
html_doc = build_standalone_html(name, windows)
|
||||
out = os.path.join(OUT_DIR, "walk_forward.html")
|
||||
with open(out, "w", encoding="utf-8") as f:
|
||||
f.write(html_doc)
|
||||
print(f"HTML 报告: {out}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main(sys.argv[1:]))
|
||||
Reference in New Issue
Block a user