gavindiaz
ed8e8afb44
通用可复用的 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，无图片依赖)。
767 lines
32 KiB
Python
767 lines
32 KiB
Python
# -*- 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:]))
|