# -*- coding: utf-8 -*- """ MT5 EA 回测报告通用分析器 ======================== 读取任意两份 MT5 Strategy Tester 导出的 xlsx 报告（通常一份 IS / 一份 OSS， 但也支持任意两份对比），输出自包含 HTML 报告： - 元信息与实际交易区间 - 核心指标对比（PF / 胜率 / 回撤 / 夏普 / Sortino / Calmar / 滚动PF 等） - 资金曲线、回撤曲线 - 方向性诊断（多/空各自的胜率、PF、盈亏比、期望） - What-If 假设分析（sell-only / buy-only / 信号反向 / 过滤双负时段 / 仓位缩放 / 盈利单放大 / 亏损截断 等通用场景，结论由数据推导） - 蒙特卡洛回撤模拟（顺序无关性检验） - 时段 / 星期 / 持仓时间分桶热力表 - 数据驱动的优化方向建议（不预写任何策略特定结论） 设计原则： 1. 不依赖任何具体 EA 的参数名、阈值或结论 2. 所有"建议"由当前两份报告的数据特征触发，而非硬编码 3. 输入仅依赖 mt5_report_parser 解析出的结构化数据 用法: python run_analysis.py python run_analysis.py # 自动找当前目录 IS-*.xlsx 和 OSS-*.xlsx """ from __future__ import annotations import glob import html import os import sys from typing import Any, Dict, List, Optional, Tuple import numpy as np import pandas as pd import mt5_report_parser as mp import walk_forward as wf OUT_DIR = "output" HTML_PATH = os.path.join(OUT_DIR, "report.html") WEEKDAY_NAMES = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"] # =========================================================================== # # 加载 # =========================================================================== # def load_reports(argv: List[str]) -> List[Tuple[str, mp.MT5Report]]: """加载两份报告。优先用命令行参数，否则自动找 IS-/OSS- 前缀文件。""" if len(argv) >= 3: files = argv[1:3] else: is_files = sorted(glob.glob("IS-*.xlsx")) oss_files = sorted(glob.glob("OSS-*.xlsx")) if not is_files or not oss_files: raise SystemExit("未找到 IS-*.xlsx / OSS-*.xlsx，请显式传两个文件名") files = [is_files[0], oss_files[0]] return [(os.path.basename(f), mp.parse_report(f)) for f in files] def actual_range(rep: mp.MT5Report) -> Tuple[Optional[str], Optional[str]]: t = rep.trades if t is None or t.empty: return None, None return str(t["open_time"].min()), str(t["open_time"].max()) # =========================================================================== # # 扩展指标 # =========================================================================== # def extended_metrics(trades: pd.DataFrame) -> Dict[str, Any]: """计算单份报告的扩展指标。纯数据驱动。""" 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] gp = wins.sum() gl = -losses.sum() pf = gp / gl if gl > 0 else np.inf equity = net.cumsum() running_max = equity.cummax() dd = equity - running_max max_dd = float(dd.min()) # Sortino（逐笔，下行波动） downside = losses downside_std = downside.std(ddof=0) if len(downside) > 1 else 0.0 sortino = float(net.mean() / downside_std) if downside_std > 0 else np.nan # Calmar = 总净盈利 / |最大回撤| calmar = float(net.sum() / abs(max_dd)) if max_dd != 0 else np.nan # 滚动 PF：每 100 笔窗口 win = 100 roll_pf = [] for i in range(0, n - win + 1, max(1, win // 4)): seg = net.iloc[i : i + win] w = seg[seg > 0].sum() l = -seg[seg <= 0].sum() roll_pf.append(w / l if l > 0 else np.inf) roll_pf = [x for x in roll_pf if not np.isinf(x)] roll_pf_min = float(min(roll_pf)) if roll_pf else np.nan roll_pf_max = float(max(roll_pf)) if roll_pf else np.nan pct_profitable_window = ( sum(1 for x in roll_pf if x > 1) / len(roll_pf) * 100 if roll_pf else 0.0 ) # 连续盈亏 cur_w = cur_l = 0 max_sw = max_sl = 0 for v in net: if v > 0: cur_w += 1 cur_l = 0 max_sw = max(max_sw, cur_w) else: cur_l += 1 cur_w = 0 max_sl = max(max_sl, cur_l) # 按方向 by_dir: Dict[str, Dict[str, Any]] = {} for direction, g in t.groupby("direction"): gn = g["net_profit"].astype(float) w = gn[gn > 0] l = gn[gn <= 0] gp_d = w.sum() gl_d = -l.sum() by_dir[direction] = { "n": int(len(g)), "win_rate": float(len(w) / len(g) * 100) if len(g) else 0, "net_profit": float(gn.sum()), "profit_factor": float(gp_d / gl_d) if gl_d > 0 else np.inf, "avg_win": float(w.mean()) if len(w) else 0.0, "avg_loss": float(l.mean()) if len(l) else 0.0, "expectancy": float(gn.mean()), } t2 = t.copy() t2["hour"] = t2["open_time"].dt.hour t2["weekday"] = t2["open_time"].dt.dayofweek 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") # 持仓时间分桶 bins = [0, 5, 15, 30, 60, 120, 1e9] labels = ["<5m", "5-15m", "15-30m", "30-60m", "1-2h", ">2h"] t2["dur_bin"] = pd.cut(t2["duration_min"], bins=bins, labels=labels, right=False) by_dur = ( t2.groupby("dur_bin", observed=True)["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_dd": max_dd, "sortino": sortino, "calmar": calmar, "max_streak_win": int(max_sw), "max_streak_loss": int(max_sl), "avg_duration_min": float(t["duration_min"].mean()), "median_duration_min": float(t["duration_min"].median()), "roll_pf_min": roll_pf_min, "roll_pf_max": roll_pf_max, "pct_profitable_window": float(pct_profitable_window), "by_direction": by_dir, "by_hour": by_hour, "by_weekday": by_weekday, "by_month": by_month, "by_duration": by_dur, "_equity": equity, "_dd": dd, "_net": net, } # =========================================================================== # # 单段通用指标（从 mt5_report_parser 复用） # =========================================================================== # from mt5_report_parser import compute_segment_metrics as _stats_from_net def whatif_scenarios(rep: mp.MT5Report) -> List[Dict[str, Any]]: """ 生成通用 What-If 场景。不依赖任何具体 EA 的参数。 场景选择基于 trades DataFrame 的通用字段（direction / open_time / net_profit）。 """ t = rep.trades.copy() t["hour"] = t["open_time"].dt.hour t["weekday"] = t["open_time"].dt.dayofweek net = t["net_profit"].astype(float) scenarios: List[Dict[str, Any]] = [] # 基线 scenarios.append({"name": "基线（现状）", "desc": "不做任何修改", **_stats_from_net(net)}) # 仅做空（若存在 short 方向） short_net = t.loc[t.direction == "short", "net_profit"] if len(short_net) > 0: scenarios.append({ "name": "仅做空 (sell-only)", "desc": "禁用多头，验证空头方向是否有正期望", **_stats_from_net(short_net), }) # 仅做多（若存在 long 方向） long_net = t.loc[t.direction == "long", "net_profit"] if len(long_net) > 0: scenarios.append({ "name": "仅做多 (buy-only)", "desc": "禁用空头，验证多头方向是否有正期望", **_stats_from_net(long_net), }) # 信号反向 —— 检测信号方向是否设反 scenarios.append({ "name": "信号反向 (net × -1)", "desc": "盈亏整体取反。若反向后净盈利为正且 PF>1，需警惕信号方向逻辑写反", **_stats_from_net(-net), }) # 过滤"双负时段"——这里只能基于单份报告自身， # 但因为本分析器同时持有两份报告，跨报告双负过滤在 build_html 里组合 # 此处提供单份报告的"过滤最差星期"作为通用场景 wd_sums = t.groupby("weekday")["net_profit"].sum() if len(wd_sums) > 0: worst_wd = int(wd_sums.idxmin()) scenarios.append({ "name": f"过滤最差星期({WEEKDAY_NAMES[worst_wd]})", "desc": f"剔除净盈亏最差的星期 {WEEKDAY_NAMES[worst_wd]}", **_stats_from_net(t.loc[t.weekday != worst_wd, "net_profit"]), }) # 过滤最差小时 hr_sums = t.groupby("hour")["net_profit"].sum() if len(hr_sums) > 0: worst_hrs = hr_sums.nsmallest(max(1, len(hr_sums) // 6)).index.tolist() scenarios.append({ "name": f"过滤最差 {len(worst_hrs)} 个小时", "desc": f"剔除净盈亏最差的 {len(worst_hrs)} 个小时窗口", **_stats_from_net(t.loc[~t.hour.isin(worst_hrs), "net_profit"]), }) # 仓位减半 scenarios.append({ "name": "仓位减半 (net × 0.5)", "desc": "缩小仓位，回撤与盈利同步减半", **_stats_from_net(net * 0.5), }) # 盈利单放大（模拟让盈利单跑更远） net_tp = net.copy() net_tp[net_tp > 0] = net_tp[net_tp > 0] * 1.5 scenarios.append({ "name": "盈利单放大 ×1.5", "desc": "模拟改进入场/离场让盈利单兑现更多利润", **_stats_from_net(net_tp), }) # 亏损截断（模拟更紧止损） net_cut = net.copy() med_loss = float(abs(losses).median()) if (losses := net[net <= 0]).size else 3.0 cap = med_loss # 以中位亏损为截断阈值，避免硬编码 net_cut[net_cut < -cap] = -cap scenarios.append({ "name": f"亏损截断 ≤ ${cap:.2f}", "desc": "模拟更紧止损，把单笔亏损封顶在中位亏损水平", **_stats_from_net(net_cut), }) # 盈亏平衡（盈利单部分回吐，模拟 BE 触发） net_be = net.copy() net_be[net_be > 0] = net_be[net_be > 0] * 0.5 scenarios.append({ "name": "启用 BE (盈利单 ×0.5)", "desc": "模拟盈亏平移触发后盈利单部分回吐，但被保护", **_stats_from_net(net_be), }) return scenarios # =========================================================================== # # 蒙特卡洛回撤模拟 # =========================================================================== # def monte_carlo_dd(net: pd.Series, n_sim: int = 1000, seed: int = 42) -> Dict[str, float]: """打乱交易顺序，看最大回撤分布。检验顺序自相关性。""" rng = np.random.default_rng(seed) arr = net.to_numpy() n = len(arr) if n == 0: return {"p5": 0, "p50": 0, "p95": 0, "actual": 0, "mean": 0} # 向量化：一次性生成 (n_sim, n) 置换矩阵，在 C 层完成 # 每行是一个随机打乱的交易顺序 perms = rng.integers(n, size=(n_sim, n)) # 每行按该行的索引排序得到 (n_sim, n) 的排列索引 idx = np.argsort(perms, axis=1) dds = np.empty(n_sim) for i in range(n_sim): perm = arr[idx[i]] eq = np.cumsum(perm) dds[i] = (eq - np.maximum.accumulate(eq)).min() return { "p5": float(np.percentile(dds, 5)), "p50": float(np.percentile(dds, 50)), "p95": float(np.percentile(dds, 95)), "actual": float((net.cumsum() - net.cumsum().cummax()).min()), "mean": float(dds.mean()), } # =========================================================================== # # 内联 SVG 图表 # =========================================================================== # def svg_equity(reports: List[Tuple[str, mp.MT5Report]], w: int = 760, h: int = 240) -> str: figs = [] for name, rep in reports: m = extended_metrics(rep.trades) if not m: continue figs.append((name, m["_equity"].to_numpy())) if not figs: return "" all_vals = np.concatenate([f[1] for f in figs]) y_min, y_max = float(all_vals.min()), float(all_vals.max()) if y_min == y_max: y_max = y_min + 1 pad = 40 colors = ["#2563eb", "#dc2626"] paths = [] for i, (name, eq) in enumerate(figs): n = len(eq) xs = np.linspace(pad, w - 10, n) ys = h - pad - (eq - y_min) / (y_max - y_min) * (h - pad - 10) d = " ".join(f"{x:.1f},{y:.1f}" for x, y in zip(xs, ys)) paths.append( f'{html.escape(name)}' ) zero_y = h - pad - (0 - y_min) / (y_max - y_min) * (h - pad - 10) grid = [] for frac in (0, 0.25, 0.5, 0.75, 1.0): yv = y_min + frac * (y_max - y_min) gy = h - pad - frac * (h - pad - 10) grid.append(f'' f'{yv:.0f}') zero_line = ( f'' if y_min < 0 < y_max else "" ) return ( f'' f'{"".join(grid)}{zero_line}{"".join(paths)}' f'交易序号 →' f'累计盈亏' f'' ) def svg_dd_hist(mc: Dict[str, float], w: int = 460, h: int = 220) -> str: """蒙特卡洛回撤分布条形图。""" top_pad = 20 bottom_pad = 40 left_pad = 50 right_pad = 16 plot_h = h - top_pad - bottom_pad plot_w = w - left_pad - right_pad items = [ ("p5(更糟)", mc["p5"]), ("均值", mc["mean"]), ("p50", mc["p50"]), ("p95(较好)", mc["p95"]), ("实际", mc["actual"]), ] vmin = min(v for _, v in items) vmax = max(v for _, v in items) if vmin == vmax: vmax = vmin + 1 pad_range = (vmax - vmin) * 0.12 vmin -= pad_range * 0.3 vmax += pad_range n = len(items) slot = plot_w / n bw = slot * 0.6 bars = [] for i, (lbl, v) in enumerate(items): x = left_pad + i * slot + (slot - bw) / 2 frac = (v - vmin) / (vmax - vmin) y1 = top_pad + plot_h - frac * plot_h color = "#dc2626" if lbl == "实际" else "#3b82f6" bars.append( f'' f'{v:.0f}' f'{lbl}' ) yticks = "" for frac in (0, 0.5, 1.0): yv = vmin + frac * (vmax - vmin) gy = top_pad + plot_h - frac * plot_h yticks += (f'' f'{yv:.0f}') return f'{yticks}{"".join(bars)}' def svg_bars(items: List[Tuple[str, float]], w: int = 520, h: int = 240, title: str = "") -> str: """柱状图。画布留足标题区(顶 24)与标签区(底 30)，柱子不再与文字重叠。""" top_pad = 24 bottom_pad = 40 left_pad = 50 right_pad = 16 plot_h = h - top_pad - bottom_pad plot_w = w - left_pad - right_pad vals = [v for _, v in items] vmin = min(vals + [0]) vmax = max(vals + [0]) if vmin == vmax: vmax = vmin + 1 # 留 10% 顶部余量，避免数值标签贴边 pad_range = (vmax - vmin) * 0.12 vmin -= pad_range * 0.3 vmax += pad_range zero_y = top_pad + plot_h - (0 - vmin) / (vmax - vmin) * plot_h n = len(items) slot = plot_w / n bw = slot * 0.6 bars = [] for i, (lbl, v) in enumerate(items): x = left_pad + i * slot + (slot - bw) / 2 y1 = top_pad + plot_h - (v - vmin) / (vmax - vmin) * plot_h yt, yb = min(zero_y, y1), max(zero_y, y1) col = "#16a34a" if v >= 0 else "#dc2626" # 数值标签放在柱子顶端外侧 val_y = yt - 4 if v >= 0 else yb + 12 bars.append( f'' f'{v:.0f}' f'{html.escape(lbl)}' ) ttl = f'{html.escape(title)}' if title else "" zero_line = f'' # Y 轴刻度 yticks = "" for frac in (0, 0.5, 1.0): yv = vmin + frac * (vmax - vmin) gy = top_pad + plot_h - frac * plot_h yticks += (f'' f'{yv:.0f}') return f'{yticks}{zero_line}{ttl}{"".join(bars)}' # =========================================================================== # # HTML 工具 # =========================================================================== # def fmt(v: Any, suffix: str = "") -> str: if v is None: return "—" if isinstance(v, float): if np.isnan(v): return "—" return f"{v:.2f}{suffix}" return f"{v}{suffix}" def heatmap_color(v: float, vmax: float) -> str: if vmax == 0: return "transparent" ratio = max(-1, min(1, v / vmax)) if ratio >= 0: return f"rgba(22,163,74,{0.15 + 0.5*abs(ratio)})" else: return f"rgba(220,38,38,{0.15 + 0.5*abs(ratio)})" def whatif_table(scenarios: List[Dict[str, Any]]) -> str: rows = [] for s in scenarios: cls = " class='pos'" if s["net"] > 0 else " class='neg'" rows.append( f"" f"{html.escape(s['name'])}
{html.escape(s['desc'])}" f"{s['n']}" f"{s['net']:+.2f}" f"{s['pf']:.2f}" f"{s['win']:.1f}%" f"{s['dd']:.2f}" f"{s['exp']:+.3f}" f"" ) return ( "" "" "" "" + "".join(rows) + "
场景笔数净盈利PF胜率最大回撤期望/笔
" ) # =========================================================================== # # 数据驱动的优化方向建议 # =========================================================================== # def build_advice(reports: List[Tuple[str, mp.MT5Report]]) -> str: """ 完全由数据特征触发建议，不预写任何策略特定结论。 每条建议都附"触发条件 → 数据 → 动作"，便于使用者核对。 """ parts: List[str] = [] (is_name, is_rep), (oss_name, oss_rep) = reports is_m = extended_metrics(is_rep.trades) oss_m = extended_metrics(oss_rep.trades) is_wif = whatif_scenarios(is_rep) oss_wif = whatif_scenarios(oss_rep) def find_wif(wif: List[Dict[str, Any]], name_contains: str) -> Optional[Dict[str, Any]]: for s in wif: if name_contains in s["name"]: return s return None # ---- 规则1：信号反向自检 ---- rev_is = find_wif(is_wif, "信号反向") rev_oss = find_wif(oss_wif, "信号反向") if rev_is and rev_oss: cond = rev_is["net"] > 0 and rev_oss["net"] > 0 and rev_is["pf"] > 1 and rev_oss["pf"] > 1 flag = "⚠️ 触发" if cond else "未触发" parts.append(f"""

① 信号方向自检 {flag}

触发条件：两份报告"信号反向"场景净盈利均为正且 PF>1。

数据：IS 反向后 净=${rev_is['net']:+.2f} / PF={rev_is['pf']:.2f}； OSS 反向后 净=${rev_oss['net']:+.2f} / PF={rev_oss['pf']:.2f}。

动作：{'强烈建议人工复核 EA 源码里 buy/sell 信号的方向判定，怀疑方向逻辑写反。' if cond else '方向逻辑未见反向特征，正常。'}

""") # ---- 规则2：方向不对称 ---- is_bd = is_m["by_direction"] oss_bd = oss_m["by_direction"] if "short" in is_bd and "long" in is_bd and "short" in oss_bd and "long" in oss_bd: is_wr_diff = abs(is_bd["short"]["win_rate"] - is_bd["long"]["win_rate"]) oss_wr_diff = abs(oss_bd["short"]["win_rate"] - oss_bd["long"]["win_rate"]) cond = is_wr_diff > 20 and oss_wr_diff > 20 flag = "⚠️ 触发" if cond else "未触发" worse_dir = "long" if is_bd["long"]["win_rate"] < is_bd["short"]["win_rate"] else "short" parts.append(f"""

② 多空方向不对称 {flag}

触发条件：两份报告多空胜率差均 > 20 个百分点。

数据：IS 多空胜率差 {is_wr_diff:.1f}，OSS 多空胜率差 {oss_wr_diff:.1f}。 较弱方向：{worse_dir}（IS 胜率 {is_bd[worse_dir]['win_rate']:.1f}%）。

动作：{'审查较弱方向的入场信号；可先单独跑该方向 What-If（sell-only/buy-only）确认其期望。' if cond else '多空较均衡，无需特殊处理。'}

""") # ---- 规则3：盈亏比 vs 胜率失衡 ---- for d in ("short", "long"): if d not in is_bd or d not in oss_bd: continue di = is_bd[d] rr_i = di["avg_win"] / abs(di["avg_loss"]) if di["avg_loss"] != 0 else 0 cond = di["win_rate"] < 40 and rr_i > 1.5 if cond: parts.append(f"""

③ {d} 方向：高盈亏比低胜率

触发条件：胜率 < 40% 且盈亏比 > 1.5（入场时机差但单笔盈亏结构尚可）。

数据：IS {d} 胜率 {di['win_rate']:.1f}%，盈亏比 {rr_i:.2f}。

动作：改进入场过滤条件以提升胜率；或确认止损/止盈设置是否合理。

""") cond2 = di["win_rate"] > 55 and rr_i < 0.8 if cond2: parts.append(f"""

③ {d} 方向：高胜率低盈亏比

触发条件：胜率 > 55% 且盈亏比 < 0.8（赢的太小输的太大，离场管理问题）。

数据：IS {d} 胜率 {di['win_rate']:.1f}%，盈亏比 {rr_i:.2f}。

动作：检查提前离场逻辑是否砍断盈利单；考虑启用盈亏平衡或让盈利单跑到目标价。

""") # ---- 规则4：双负时段过滤 ---- is_hr = is_m["by_hour"] oss_hr = oss_m["by_hour"] both_neg_hours = [ h for h in range(24) if is_hr.get(h, {}).get("sum", 0) < 0 and oss_hr.get(h, {}).get("sum", 0) < 0 ] is_wd = is_m["by_weekday"] oss_wd = oss_m["by_weekday"] both_neg_wds = [ d for d in range(7) if is_wd.get(d, {}).get("sum", 0) < 0 and oss_wd.get(d, {}).get("sum", 0) < 0 ] if both_neg_hours or both_neg_wds: wd_str = "、".join(WEEKDAY_NAMES[d] for d in both_neg_wds) or "无" parts.append(f"""

④ 时段/星期过滤 触发

触发条件：某时段在两份报告同时为负（系统性劣势）。

数据：双负小时 {len(both_neg_hours)} 个 {[f'{h:02d}' for h in both_neg_hours]}； 双负星期：{wd_str}。

动作：若 EA 有时段过滤参数，开启并剔除上述窗口；否则在信号逻辑中加入时间过滤条件。

""") # ---- 规则5：回撤过大 ---- is_dd_pct = is_rep.summary_norm.get("max_balance_dd_pct") oss_dd_pct = oss_rep.summary_norm.get("max_balance_dd_pct") if is_dd_pct and oss_dd_pct and (is_dd_pct > 40 or oss_dd_pct > 40): parts.append(f"""

⑤ 回撤控制 触发

触发条件：任一报告最大回撤 > 40%。

数据：IS 回撤 {is_dd_pct}%，OSS 回撤 {oss_dd_pct}%，最大连败 IS {is_m['max_streak_loss']} 笔 / OSS {oss_m['max_streak_loss']} 笔。

动作：缩小单笔风险（止损/仓位）；考虑在连亏达 N 笔时暂停或减仓； 参考 What-If 表"仓位减半"和"亏损截断"场景的回撤改善效果。

""") # ---- 规则6：蒙特卡洛顺序自相关 ---- is_mc = monte_carlo_dd(is_m["_net"]) oss_mc = monte_carlo_dd(oss_m["_net"]) # 实际回撤显著差于中位 → 顺序有自相关（连败聚集） is_cluster = is_mc["actual"] < is_mc["p50"] * 1.3 oss_cluster = oss_mc["actual"] < oss_mc["p50"] * 1.3 if is_cluster or oss_cluster: parts.append(f"""

⑥ 连败聚集检测 触发

触发条件：实际回撤显著差于蒙特卡洛中位回撤（>1.3倍）。

数据：IS 实际 {is_mc['actual']:.0f} vs 中位 {is_mc['p50']:.0f}； OSS 实际 {oss_mc['actual']:.0f} vs 中位 {oss_mc['p50']:.0f}。

动作：交易顺序存在自我相关（亏损倾向连发），建议加连亏保护机制。

""") # ---- 规则7：盈利单放大场景效果 ---- tp_is = find_wif(is_wif, "盈利单放大") tp_oss = find_wif(oss_wif, "盈利单放大") if tp_is and tp_oss: cond = tp_is["pf"] > 1 and tp_oss["pf"] > 1 and \ tp_is["pf"] > is_m["profit_factor"] and tp_oss["pf"] > oss_m["profit_factor"] if cond: parts.append(f"""

⑦ 离场优化潜力

触发条件："盈利单放大×1.5"场景 PF 站上 1 且优于基线。

数据：IS 基线 PF {is_m['profit_factor']:.2f} → 场景 PF {tp_is['pf']:.2f}； OSS 基线 PF {oss_m['profit_factor']:.2f} → 场景 PF {tp_oss['pf']:.2f}。

动作：离场逻辑有改进空间，考虑让盈利单兑现更多利润（放宽止盈/调整提前离场条件）。

""") # ---- 规则8：参数稳健性 ---- pf_diff = abs(is_m["profit_factor"] - oss_m["profit_factor"]) if pf_diff > 0.2: parts.append(f"""

⑧ 参数稳健性

触发条件：两份报告 PF 差异 > 0.2。

数据：IS PF {is_m['profit_factor']:.2f}，OSS PF {oss_m['profit_factor']:.2f}，差 {pf_diff:.2f}。

动作：策略对行情敏感，建议做参数敏感度扫描找稳健高原（而非单点峰值），并用 Walk-Forward 验证。

""") if not parts: parts.append("

当前数据未触发任何预设告警规则，策略各维度表现均在阈值内。

") return "".join(parts) # =========================================================================== # # HTML 主构建 # =========================================================================== # def build_html(reports: List[Tuple[str, mp.MT5Report]]) -> str: a_name, a_rep = reports[0] b_name, b_rep = reports[1] a_m = extended_metrics(a_rep.trades) b_m = extended_metrics(b_rep.trades) a_s = a_rep.summary_norm b_s = b_rep.summary_norm a_start, a_end = actual_range(a_rep) b_start, b_end = actual_range(b_rep) a_wif = whatif_scenarios(a_rep) b_wif = whatif_scenarios(b_rep) a_mc = monte_carlo_dd(a_m["_net"]) b_mc = monte_carlo_dd(b_m["_net"]) ea_name = a_rep.meta.get("专家", "—") symbol = a_rep.meta.get("交易品种", "—") # ---------- 核心指标对比 ---------- core_rows = [ ("交易笔数", a_m["n_trades"], b_m["n_trades"], ""), ("净盈利", a_m["net_profit"], b_m["net_profit"], ""), ("盈利因子 PF", a_m["profit_factor"], b_m["profit_factor"], ""), ("胜率 (%)", a_m["win_rate"], b_m["win_rate"], ""), ("平均盈利", a_m["avg_win"], b_m["avg_win"], ""), ("平均亏损", a_m["avg_loss"], b_m["avg_loss"], ""), ("盈亏比 (avgW/|avgL|)", a_m["avg_win"]/abs(a_m["avg_loss"]) if a_m["avg_loss"] else 0, b_m["avg_win"]/abs(b_m["avg_loss"]) if b_m["avg_loss"] else 0, ""), ("期望收益/笔", a_m["expectancy"], b_m["expectancy"], ""), ("最大回撤", a_m["max_dd"], b_m["max_dd"], ""), ("最大回撤 (%)", a_s.get("max_balance_dd_pct"), b_s.get("max_balance_dd_pct"), "%"), ("夏普比率", a_s.get("sharpe"), b_s.get("sharpe"), ""), ("Sortino (逐笔)", a_m["sortino"], b_m["sortino"], ""), ("Calmar (净利/|回撤|)", a_m["calmar"], b_m["calmar"], ""), ("最大连败", a_m["max_streak_loss"], b_m["max_streak_loss"], "笔"), ("平均持仓 (分钟)", a_m["avg_duration_min"], b_m["avg_duration_min"], ""), ("滚动PF 区间", f"{a_m['roll_pf_min']:.2f}~{a_m['roll_pf_max']:.2f}", f"{b_m['roll_pf_min']:.2f}~{b_m['roll_pf_max']:.2f}", ""), ("滚动窗口盈利比例", a_m["pct_profitable_window"], b_m["pct_profitable_window"], "%"), ] core_html = ( "" f"" + "".join( f"" for n, iv, ov, u in core_rows ) + "
指标{html.escape(a_name)}{html.escape(b_name)}单位
{n}{fmt(iv)}{fmt(ov)}{u}
" ) # ---------- 方向诊断 ---------- all_dirs = sorted(set(list(a_m["by_direction"].keys()) + list(b_m["by_direction"].keys()))) dir_rows = [] for d in all_dirs: di = a_m["by_direction"].get(d, {}) do = b_m["by_direction"].get(d, {}) rr_i = di.get('avg_win',0)/abs(di.get('avg_loss',1)) if di.get('avg_loss') else 0 rr_o = do.get('avg_win',0)/abs(do.get('avg_loss',1)) if do.get('avg_loss') else 0 for lbl, iv, ov in [ ("笔数", di.get('n',0), do.get('n',0)), ("胜率(%)", di.get('win_rate',0), do.get('win_rate',0)), ("净盈利", di.get('net_profit',0), do.get('net_profit',0)), ("PF", di.get('profit_factor',0), do.get('profit_factor',0)), ("平均盈利", di.get('avg_win',0), do.get('avg_win',0)), ("平均亏损", di.get('avg_loss',0), do.get('avg_loss',0)), ("盈亏比", rr_i, rr_o), ("期望/笔", di.get('expectancy',0), do.get('expectancy',0)), ]: cls = " class='neg'" if (isinstance(iv,(int,float)) and iv < 0) else "" dir_rows.append(f"{d}{lbl}{fmt(iv)}{fmt(ov)}") dir_html = ( "" f"" + "".join(dir_rows) + "
方向指标{html.escape(a_name)}{html.escape(b_name)}
" ) # ---------- What-If ---------- wif_html = ( f"

{html.escape(a_name)}

" + whatif_table(a_wif) + f"

{html.escape(b_name)}

" + whatif_table(b_wif) ) # ---------- 蒙特卡洛 ---------- mc_html = ( "" f"" f"" f"" f"" f"" f"" + "
分位{html.escape(a_name)}{html.escape(b_name)}
实际最大回撤{a_mc['actual']:.2f}{b_mc['actual']:.2f}
蒙特卡洛 5% 分位（更糟）{a_mc['p5']:.2f}{b_mc['p5']:.2f}
蒙特卡洛 中位{a_mc['p50']:.2f}{b_mc['p50']:.2f}
蒙特卡洛 95% 分位（较好）{a_mc['p95']:.2f}{b_mc['p95']:.2f}
蒙特卡洛 均值{a_mc['mean']:.2f}{b_mc['mean']:.2f}
" ) # ---------- 时段表 ---------- hour_rows = [] max_abs = max( [abs(a_m["by_hour"].get(h, {}).get("sum", 0)) for h in range(24)] + [abs(b_m["by_hour"].get(h, {}).get("sum", 0)) for h in range(24)] ) or 1 for h in range(24): ih = a_m["by_hour"].get(h, {}) oh = b_m["by_hour"].get(h, {}) if not ih and not oh: continue a_sum = ih.get('sum', 0) b_sum = oh.get('sum', 0) ci = heatmap_color(a_sum, max_abs) co = heatmap_color(b_sum, max_abs) both_neg = "⚠️" if (a_sum < 0 and b_sum < 0) else "" hour_rows.append( f"{h:02d}" f"{a_sum:+.2f}{ih.get('count',0)}" f"{b_sum:+.2f}{oh.get('count',0)}" f"{both_neg}" ) hour_html = ( "" f"" f"" + "".join(hour_rows) + "
小时{html.escape(a_name)}净笔{html.escape(b_name)}净笔双负
" ) # ---------- 星期表 ---------- wd_rows = [] max_abs_w = max( [abs(a_m["by_weekday"].get(d, {}).get("sum", 0)) for d in range(7)] + [abs(b_m["by_weekday"].get(d, {}).get("sum", 0)) for d in range(7)] ) or 1 for d in range(7): iw = a_m["by_weekday"].get(d, {}) ow = b_m["by_weekday"].get(d, {}) if not iw and not ow: continue a_sum = iw.get('sum', 0) b_sum = ow.get('sum', 0) ci = heatmap_color(a_sum, max_abs_w) co = heatmap_color(b_sum, max_abs_w) both_neg = "⚠️" if (a_sum < 0 and b_sum < 0) else "" wd_rows.append( f"{WEEKDAY_NAMES[d]}" f"{a_sum:+.2f}{iw.get('count',0)}" f"{b_sum:+.2f}{ow.get('count',0)}" f"{both_neg}" ) wd_html = ( "" f"" f"" + "".join(wd_rows) + "
星期{html.escape(a_name)}净笔{html.escape(b_name)}净笔双负
" ) # ---------- 持仓时间分桶 ---------- dur_labels = ["<5m", "5-15m", "15-30m", "30-60m", "1-2h", ">2h"] dur_rows = [] for lbl in dur_labels: ad = a_m["by_duration"].get(lbl, {}) bd = b_m["by_duration"].get(lbl, {}) if not ad and not bd: continue dur_rows.append( f"{lbl}" f"{ad.get('count',0)}{ad.get('sum',0):+.2f}{ad.get('mean',0):+.3f}" f"{bd.get('count',0)}{bd.get('sum',0):+.2f}{bd.get('mean',0):+.3f}" ) dur_html = ( "" f"" f"" + "".join(dur_rows) + "
持仓区间{html.escape(a_name)}笔净均值{html.escape(b_name)}笔净均值
" ) # ---------- SVG ---------- equity_svg = svg_equity(reports) dir_bar_items = [] for d in all_dirs: di = a_m["by_direction"].get(d, {}) do = b_m["by_direction"].get(d, {}) dir_bar_items.append((f"{d}-A", di.get('net_profit', 0))) dir_bar_items.append((f"{d}-B", do.get('net_profit', 0))) dir_svg = svg_bars(dir_bar_items, title="各方向净盈亏（A=报告A, B=报告B）") a_mc_svg = svg_dd_hist(a_mc) b_mc_svg = svg_dd_hist(b_mc) # ---------- 数据驱动建议 ---------- advice_html = build_advice(reports) # ---------- Walk-Forward ---------- # 对两份报告分别做滚动 IS→OOS 验证（窗口太短会自动返回空片段） wf_a = wf.wf_html_fragment(wf.walk_forward(a_rep.trades, is_days=45, oos_days=30)) wf_b = wf.wf_html_fragment(wf.walk_forward(b_rep.trades, is_days=45, oos_days=30)) # ---------- 元信息 ---------- inputs = a_rep.meta.get("inputs", {}) inputs_match = (inputs == b_rep.meta.get("inputs", {})) inputs_html = "" if inputs: rows = "".join( f"{html.escape(k)}{html.escape(v)}" f"{'✓ 一致' if inputs_match else html.escape(b_rep.meta.get('inputs',{}).get(k,'—'))}" for k, v in inputs.items() ) inputs_html = ( "

输入参数

" "" + rows + "
参数报告A报告B
" ) # ---------- 拼装 ---------- doc = f""" EA 回测分析报告

EA 回测分析报告

EA: {html.escape(str(ea_name))}　品种: {html.escape(str(symbol))}　 报告A = {html.escape(a_name)}　报告B = {html.escape(b_name)}

报告A 净盈利
{a_m['net_profit']:+.2f}
报告B 净盈利
{b_m['net_profit']:+.2f}
PF (A / B)
{a_m['profit_factor']:.2f} / {b_m['profit_factor']:.2f}
回撤% (A / B)
{a_s.get('max_balance_dd_pct')}% / {b_s.get('max_balance_dd_pct')}%
实际交易区间：报告A = {a_start} ~ {a_end}；报告B = {b_start} ~ {b_end}。
输入参数：{'完全一致' if inputs_match else '存在差异（见下表）'}。 若两份报告参数相同而结果差异大，说明策略对行情敏感；若参数不同，需先确认对比是否公平。
{inputs_html}

1. 核心指标对比

{core_html}

Sortino/Calmar 为逐笔口径估算；滚动 PF 以 100 笔窗口滑动，"滚动窗口盈利比例"=窗口 PF>1 的占比（越高越稳定）。

2. 资金曲线

{equity_svg}

3. 方向性诊断

{dir_html} {dir_svg}

若多空胜率/盈亏比显著不对称，是结构性偏差信号，需结合下方 What-If 与建议规则分析。

4. What-If 假设分析

对每份报告模拟多种改造场景，重算指标。场景为通用逻辑，不依赖具体 EA 参数。 重点看 PF 是否站上 1、回撤是否收敛、期望是否转正。

{wif_html}
必看："信号反向"场景若两份报告净盈利均转正且 PF>1，需警惕信号方向逻辑写反，应人工复核源码。

5. 蒙特卡洛回撤模拟

将逐笔盈亏随机打乱 1000 次，看顺序无关下的回撤分布。 若"实际回撤"显著差于中位，说明存在连败聚集（顺序有自我相关），建议加连亏保护。

报告A 蒙特卡洛回撤

{a_mc_svg}

报告B 蒙特卡洛回撤

{b_mc_svg}
{mc_html}

6. 时段诊断

6.1 按小时

{hour_html}

6.2 按星期

{wd_html}

⚠️ = 两份报告同时为负，属系统性劣势时段，应优先过滤。单元格颜色：红=负、绿=正。

7. 持仓时间分桶

{dur_html}

观察哪个持仓区间贡献正/负盈亏，可指导止盈止损时间维度调整。

8. 数据驱动的优化方向建议

以下建议由当前数据的特征触发（每条附触发条件、数据、动作），不预设任何策略特定结论。 未触发的规则不显示，未列出的维度表示数据正常。

{advice_html}

9. Walk-Forward 滚动验证

将每份报告的逐笔交易按时间切成滚动窗口（IS 45天 + OOS 30天，步长 30天）， 计算每个窗口的 IS/OOS 指标。WFE = ΣOOS净 / ΣIS净，越高越稳健； IS-OOS PF 相关性高=泛化好，低=过拟合风险。数据太短无法切窗会提示。

报告A: {html.escape(a_name)}

{wf_a}

报告B: {html.escape(b_name)}

{wf_b}

10. 可选扩展分析

以下三项已实现为独立可复用工具，按需调用：

  • 参数敏感度扫描：python param_scan.py gen-set ... 生成网格 .set + 批处理脚本， 在 MT5 跑完后 python param_scan.py analyze ... 出响应面热力图 + 3D 曲面 + 高原检测。 python param_scan.py demo 可先看效果。
  • MAE/MFE 分析：MT5 标准 xlsx 不含逐笔 MAE/MFE，需先用 python mae_mfe.py --show-snippet 取 MQL5 代码粘进 EA 导出 CSV，再 python mae_mfe.py mae_mfe.csv 出散点 + TP/SL 扫描热力图。
  • Walk-Forward 独立报告：python walk_forward.py <report.xlsx> --is-days 60 --oos-days 30 出滚动窗口 IS/OOS 验证（本报告第 8 节已内嵌简化版）。

其它建议（未实现，需自行扩展）：

  • 多品种/多周期：同策略测多品种多周期，看泛化能力。
  • 交易成本敏感度：点差/手续费放大 1.5x、2x 看 PF 退化曲线，评估实盘可行性。
  • 信号因子分解：复合信号拆单因子分别回测，剔除拖累项。
""" return doc # =========================================================================== # # 控制台摘要 # =========================================================================== # def print_console(reports: List[Tuple[str, mp.MT5Report]]) -> None: print("=" * 70) for name, rep in reports: m = extended_metrics(rep.trades) s = rep.summary_norm bd = m["by_direction"] print(f"[{name}]") print(f" 区间: {actual_range(rep)[0]} ~ {actual_range(rep)[1]}") print(f" 笔数={m['n_trades']} 净={m['net_profit']:+.2f} PF={m['profit_factor']:.2f} " f"胜率={m['win_rate']:.1f}% 回撤={s.get('max_balance_dd_pct')}% 夏普={s.get('sharpe')}") for d, g in bd.items(): print(f" {d}: 胜率={g['win_rate']:.1f}% 净={g['net_profit']:+.2f} PF={g['profit_factor']:.2f}") wif = whatif_scenarios(rep) print(f" --- What-If 关键 ---") for sc_name in ["信号反向 (net × -1)", "盈利单放大 ×1.5"]: for x in wif: if x["name"] == sc_name: print(f" {sc_name}: 净={x['net']:+.2f} PF={x['pf']:.2f} 回撤={x['dd']:.2f}") break print() print("=" * 70) def main(argv: List[str]) -> int: os.makedirs(OUT_DIR, exist_ok=True) reports = load_reports(argv) print_console(reports) html_doc = build_html(reports) with open(HTML_PATH, "w", encoding="utf-8") as f: f.write(html_doc) print(f"\nHTML 报告已生成: {HTML_PATH}") print("（自包含，无图片依赖，可直接浏览器打开 / AI 解析表格）") return 0 if __name__ == "__main__": sys.exit(main(sys.argv))