1062 lines
47 KiB
Python
1062 lines
47 KiB
Python
# -*- 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 <report_a.xlsx> <report_b.xlsx>
|
||
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'<polyline points="{d}" fill="none" stroke="{colors[i % len(colors)]}" '
|
||
f'stroke-width="1.4" opacity="0.9"><title>{html.escape(name)}</title></polyline>'
|
||
)
|
||
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'<line x1="{pad}" y1="{gy:.1f}" x2="{w-10}" y2="{gy:.1f}" stroke="#e5e7eb" stroke-width="0.5"/>'
|
||
f'<text x="4" y="{gy+3:.1f}" font-size="9" fill="#6b7280">{yv:.0f}</text>')
|
||
zero_line = (
|
||
f'<line x1="{pad}" y1="{zero_y:.1f}" x2="{w-10}" y2="{zero_y:.1f}" '
|
||
f'stroke="#9ca3af" stroke-dasharray="3,3" stroke-width="0.7"/>'
|
||
if y_min < 0 < y_max else ""
|
||
)
|
||
return (
|
||
f'<svg viewBox="0 0 {w} {h}" class="chart">'
|
||
f'{"".join(grid)}{zero_line}{"".join(paths)}'
|
||
f'<text x="{w//2}" y="{h-4}" font-size="10" fill="#374151" text-anchor="middle">交易序号 →</text>'
|
||
f'<text x="10" y="14" font-size="10" fill="#374151">累计盈亏</text>'
|
||
f'</svg>'
|
||
)
|
||
|
||
|
||
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'<rect x="{x:.1f}" y="{y1:.1f}" width="{bw:.1f}" height="{top_pad+plot_h-y1:.1f}" fill="{color}"/>'
|
||
f'<text x="{x+bw/2:.1f}" y="{y1-4:.1f}" font-size="9" fill="#111827" text-anchor="middle">{v:.0f}</text>'
|
||
f'<text x="{x+bw/2:.1f}" y="{h-bottom_pad+15:.1f}" font-size="9" fill="#374151" text-anchor="middle">{lbl}</text>'
|
||
)
|
||
yticks = ""
|
||
for frac in (0, 0.5, 1.0):
|
||
yv = vmin + frac * (vmax - vmin)
|
||
gy = top_pad + plot_h - frac * plot_h
|
||
yticks += (f'<line x1="{left_pad}" y1="{gy:.1f}" x2="{w-right_pad}" y2="{gy:.1f}" stroke="#e5e7eb" stroke-width="0.5"/>'
|
||
f'<text x="{left_pad-6}" 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">{yticks}{"".join(bars)}</svg>'
|
||
|
||
|
||
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'<rect x="{x:.1f}" y="{yt:.1f}" width="{bw:.1f}" height="{yb-yt:.1f}" fill="{col}"/>'
|
||
f'<text x="{x+bw/2:.1f}" y="{val_y:.1f}" font-size="10" fill="#111827" text-anchor="middle">{v:.0f}</text>'
|
||
f'<text x="{x+bw/2:.1f}" y="{h-bottom_pad+15:.1f}" font-size="10" fill="#374151" text-anchor="middle">{html.escape(lbl)}</text>'
|
||
)
|
||
ttl = f'<text x="{w//2}" y="16" font-size="11" fill="#374151" text-anchor="middle">{html.escape(title)}</text>' if title else ""
|
||
zero_line = f'<line x1="{left_pad}" y1="{zero_y:.1f}" x2="{w-right_pad}" y2="{zero_y:.1f}" stroke="#9ca3af" stroke-dasharray="3,3" stroke-width="0.7"/>'
|
||
# 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'<line x1="{left_pad}" y1="{gy:.1f}" x2="{w-right_pad}" y2="{gy:.1f}" stroke="#e5e7eb" stroke-width="0.5"/>'
|
||
f'<text x="{left_pad-6}" 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">{yticks}{zero_line}{ttl}{"".join(bars)}</svg>'
|
||
|
||
|
||
# =========================================================================== #
|
||
# 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"<tr{cls}>"
|
||
f"<td><b>{html.escape(s['name'])}</b><br><span class='muted'>{html.escape(s['desc'])}</span></td>"
|
||
f"<td>{s['n']}</td>"
|
||
f"<td>{s['net']:+.2f}</td>"
|
||
f"<td>{s['pf']:.2f}</td>"
|
||
f"<td>{s['win']:.1f}%</td>"
|
||
f"<td>{s['dd']:.2f}</td>"
|
||
f"<td>{s['exp']:+.3f}</td>"
|
||
f"</tr>"
|
||
)
|
||
return (
|
||
"<table class='data'><thead><tr>"
|
||
"<th>场景</th><th>笔数</th><th>净盈利</th><th>PF</th>"
|
||
"<th>胜率</th><th>最大回撤</th><th>期望/笔</th>"
|
||
"</tr></thead><tbody>" + "".join(rows) + "</tbody></table>"
|
||
)
|
||
|
||
|
||
# =========================================================================== #
|
||
# 数据驱动的优化方向建议
|
||
# =========================================================================== #
|
||
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"""
|
||
<h3>① 信号方向自检 <span class='tag {'' if cond else 'ok'}'>{flag}</span></h3>
|
||
<p><b>触发条件</b>：两份报告"信号反向"场景净盈利均为正且 PF>1。</p>
|
||
<p><b>数据</b>：IS 反向后 净=${rev_is['net']:+.2f} / PF={rev_is['pf']:.2f}；
|
||
OSS 反向后 净=${rev_oss['net']:+.2f} / PF={rev_oss['pf']:.2f}。</p>
|
||
<p><b>动作</b>：{'强烈建议人工复核 EA 源码里 buy/sell 信号的方向判定，怀疑方向逻辑写反。' if cond else '方向逻辑未见反向特征，正常。'}</p>
|
||
""")
|
||
|
||
# ---- 规则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"""
|
||
<h3>② 多空方向不对称 <span class='tag {'' if cond else 'ok'}'>{flag}</span></h3>
|
||
<p><b>触发条件</b>：两份报告多空胜率差均 > 20 个百分点。</p>
|
||
<p><b>数据</b>：IS 多空胜率差 {is_wr_diff:.1f}，OSS 多空胜率差 {oss_wr_diff:.1f}。
|
||
较弱方向：{worse_dir}（IS 胜率 {is_bd[worse_dir]['win_rate']:.1f}%）。</p>
|
||
<p><b>动作</b>：{'审查较弱方向的入场信号；可先单独跑该方向 What-If（sell-only/buy-only）确认其期望。' if cond else '多空较均衡，无需特殊处理。'}</p>
|
||
""")
|
||
|
||
# ---- 规则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"""
|
||
<h3>③ {d} 方向：高盈亏比低胜率</h3>
|
||
<p><b>触发条件</b>：胜率 < 40% 且盈亏比 > 1.5（入场时机差但单笔盈亏结构尚可）。</p>
|
||
<p><b>数据</b>：IS {d} 胜率 {di['win_rate']:.1f}%，盈亏比 {rr_i:.2f}。</p>
|
||
<p><b>动作</b>：改进入场过滤条件以提升胜率；或确认止损/止盈设置是否合理。</p>
|
||
""")
|
||
cond2 = di["win_rate"] > 55 and rr_i < 0.8
|
||
if cond2:
|
||
parts.append(f"""
|
||
<h3>③ {d} 方向：高胜率低盈亏比</h3>
|
||
<p><b>触发条件</b>：胜率 > 55% 且盈亏比 < 0.8（赢的太小输的太大，离场管理问题）。</p>
|
||
<p><b>数据</b>：IS {d} 胜率 {di['win_rate']:.1f}%，盈亏比 {rr_i:.2f}。</p>
|
||
<p><b>动作</b>：检查提前离场逻辑是否砍断盈利单；考虑启用盈亏平衡或让盈利单跑到目标价。</p>
|
||
""")
|
||
|
||
# ---- 规则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"""
|
||
<h3>④ 时段/星期过滤 <span class='tag warn'>触发</span></h3>
|
||
<p><b>触发条件</b>：某时段在两份报告同时为负（系统性劣势）。</p>
|
||
<p><b>数据</b>：双负小时 {len(both_neg_hours)} 个 {[f'{h:02d}' for h in both_neg_hours]}；
|
||
双负星期：{wd_str}。</p>
|
||
<p><b>动作</b>：若 EA 有时段过滤参数，开启并剔除上述窗口；否则在信号逻辑中加入时间过滤条件。</p>
|
||
""")
|
||
|
||
# ---- 规则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"""
|
||
<h3>⑤ 回撤控制 <span class='tag warn'>触发</span></h3>
|
||
<p><b>触发条件</b>：任一报告最大回撤 > 40%。</p>
|
||
<p><b>数据</b>：IS 回撤 {is_dd_pct}%，OSS 回撤 {oss_dd_pct}%，最大连败 IS {is_m['max_streak_loss']} 笔 / OSS {oss_m['max_streak_loss']} 笔。</p>
|
||
<p><b>动作</b>：缩小单笔风险（止损/仓位）；考虑在连亏达 N 笔时暂停或减仓；
|
||
参考 What-If 表"仓位减半"和"亏损截断"场景的回撤改善效果。</p>
|
||
""")
|
||
|
||
# ---- 规则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"""
|
||
<h3>⑥ 连败聚集检测 <span class='tag warn'>触发</span></h3>
|
||
<p><b>触发条件</b>：实际回撤显著差于蒙特卡洛中位回撤（>1.3倍）。</p>
|
||
<p><b>数据</b>：IS 实际 {is_mc['actual']:.0f} vs 中位 {is_mc['p50']:.0f}；
|
||
OSS 实际 {oss_mc['actual']:.0f} vs 中位 {oss_mc['p50']:.0f}。</p>
|
||
<p><b>动作</b>：交易顺序存在自我相关（亏损倾向连发），建议加连亏保护机制。</p>
|
||
""")
|
||
|
||
# ---- 规则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"""
|
||
<h3>⑦ 离场优化潜力</h3>
|
||
<p><b>触发条件</b>："盈利单放大×1.5"场景 PF 站上 1 且优于基线。</p>
|
||
<p><b>数据</b>：IS 基线 PF {is_m['profit_factor']:.2f} → 场景 PF {tp_is['pf']:.2f}；
|
||
OSS 基线 PF {oss_m['profit_factor']:.2f} → 场景 PF {tp_oss['pf']:.2f}。</p>
|
||
<p><b>动作</b>：离场逻辑有改进空间，考虑让盈利单兑现更多利润（放宽止盈/调整提前离场条件）。</p>
|
||
""")
|
||
|
||
# ---- 规则8：参数稳健性 ----
|
||
pf_diff = abs(is_m["profit_factor"] - oss_m["profit_factor"])
|
||
if pf_diff > 0.2:
|
||
parts.append(f"""
|
||
<h3>⑧ 参数稳健性</h3>
|
||
<p><b>触发条件</b>：两份报告 PF 差异 > 0.2。</p>
|
||
<p><b>数据</b>：IS PF {is_m['profit_factor']:.2f}，OSS PF {oss_m['profit_factor']:.2f}，差 {pf_diff:.2f}。</p>
|
||
<p><b>动作</b>：策略对行情敏感，建议做参数敏感度扫描找稳健高原（而非单点峰值），并用 Walk-Forward 验证。</p>
|
||
""")
|
||
|
||
if not parts:
|
||
parts.append("<p>当前数据未触发任何预设告警规则，策略各维度表现均在阈值内。</p>")
|
||
|
||
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 = (
|
||
"<table class='data'><thead><tr>"
|
||
f"<th>指标</th><th>{html.escape(a_name)}</th><th>{html.escape(b_name)}</th><th>单位</th></tr></thead><tbody>"
|
||
+ "".join(
|
||
f"<tr><td>{n}</td><td>{fmt(iv)}</td><td>{fmt(ov)}</td><td>{u}</td></tr>"
|
||
for n, iv, ov, u in core_rows
|
||
)
|
||
+ "</tbody></table>"
|
||
)
|
||
|
||
# ---------- 方向诊断 ----------
|
||
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"<tr{cls}><td>{d}</td><td>{lbl}</td><td>{fmt(iv)}</td><td>{fmt(ov)}</td></tr>")
|
||
dir_html = (
|
||
"<table class='data'><thead><tr>"
|
||
f"<th>方向</th><th>指标</th><th>{html.escape(a_name)}</th><th>{html.escape(b_name)}</th></tr></thead><tbody>"
|
||
+ "".join(dir_rows) + "</tbody></table>"
|
||
)
|
||
|
||
# ---------- What-If ----------
|
||
wif_html = (
|
||
f"<h3>{html.escape(a_name)}</h3>" + whatif_table(a_wif)
|
||
+ f"<h3>{html.escape(b_name)}</h3>" + whatif_table(b_wif)
|
||
)
|
||
|
||
# ---------- 蒙特卡洛 ----------
|
||
mc_html = (
|
||
"<table class='data'><thead><tr>"
|
||
f"<th>分位</th><th>{html.escape(a_name)}</th><th>{html.escape(b_name)}</th></tr></thead><tbody>"
|
||
f"<tr><td>实际最大回撤</td><td class='neg'>{a_mc['actual']:.2f}</td><td class='neg'>{b_mc['actual']:.2f}</td></tr>"
|
||
f"<tr><td>蒙特卡洛 5% 分位（更糟）</td><td>{a_mc['p5']:.2f}</td><td>{b_mc['p5']:.2f}</td></tr>"
|
||
f"<tr><td>蒙特卡洛 中位</td><td>{a_mc['p50']:.2f}</td><td>{b_mc['p50']:.2f}</td></tr>"
|
||
f"<tr><td>蒙特卡洛 95% 分位（较好）</td><td>{a_mc['p95']:.2f}</td><td>{b_mc['p95']:.2f}</td></tr>"
|
||
f"<tr><td>蒙特卡洛 均值</td><td>{a_mc['mean']:.2f}</td><td>{b_mc['mean']:.2f}</td></tr>"
|
||
+ "</tbody></table>"
|
||
)
|
||
|
||
# ---------- 时段表 ----------
|
||
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"<tr><td>{h:02d}</td>"
|
||
f"<td style='background:{ci}'>{a_sum:+.2f}</td><td>{ih.get('count',0)}</td>"
|
||
f"<td style='background:{co}'>{b_sum:+.2f}</td><td>{oh.get('count',0)}</td>"
|
||
f"<td>{both_neg}</td></tr>"
|
||
)
|
||
hour_html = (
|
||
"<table class='data'><thead><tr><th>小时</th>"
|
||
f"<th>{html.escape(a_name)}净</th><th>笔</th>"
|
||
f"<th>{html.escape(b_name)}净</th><th>笔</th><th>双负</th></tr></thead><tbody>"
|
||
+ "".join(hour_rows) + "</tbody></table>"
|
||
)
|
||
|
||
# ---------- 星期表 ----------
|
||
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"<tr><td>{WEEKDAY_NAMES[d]}</td>"
|
||
f"<td style='background:{ci}'>{a_sum:+.2f}</td><td>{iw.get('count',0)}</td>"
|
||
f"<td style='background:{co}'>{b_sum:+.2f}</td><td>{ow.get('count',0)}</td>"
|
||
f"<td>{both_neg}</td></tr>"
|
||
)
|
||
wd_html = (
|
||
"<table class='data'><thead><tr><th>星期</th>"
|
||
f"<th>{html.escape(a_name)}净</th><th>笔</th>"
|
||
f"<th>{html.escape(b_name)}净</th><th>笔</th><th>双负</th></tr></thead><tbody>"
|
||
+ "".join(wd_rows) + "</tbody></table>"
|
||
)
|
||
|
||
# ---------- 持仓时间分桶 ----------
|
||
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"<tr><td>{lbl}</td>"
|
||
f"<td>{ad.get('count',0)}</td><td>{ad.get('sum',0):+.2f}</td><td>{ad.get('mean',0):+.3f}</td>"
|
||
f"<td>{bd.get('count',0)}</td><td>{bd.get('sum',0):+.2f}</td><td>{bd.get('mean',0):+.3f}</td></tr>"
|
||
)
|
||
dur_html = (
|
||
"<table class='data'><thead><tr><th>持仓区间</th>"
|
||
f"<th>{html.escape(a_name)}笔</th><th>净</th><th>均值</th>"
|
||
f"<th>{html.escape(b_name)}笔</th><th>净</th><th>均值</th></tr></thead><tbody>"
|
||
+ "".join(dur_rows) + "</tbody></table>"
|
||
)
|
||
|
||
# ---------- 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"<tr><td><code>{html.escape(k)}</code></td><td>{html.escape(v)}</td>"
|
||
f"<td>{'✓ 一致' if inputs_match else html.escape(b_rep.meta.get('inputs',{}).get(k,'—'))}</td></tr>"
|
||
for k, v in inputs.items()
|
||
)
|
||
inputs_html = (
|
||
"<h3>输入参数</h3>"
|
||
"<table class='data'><thead><tr><th>参数</th><th>报告A</th><th>报告B</th></tr></thead><tbody>"
|
||
+ rows + "</tbody></table>"
|
||
)
|
||
|
||
# ---------- 拼装 ----------
|
||
doc = f"""<!doctype html>
|
||
<html lang="zh-CN"><head><meta charset="utf-8">
|
||
<title>EA 回测分析报告</title>
|
||
<style>
|
||
:root {{--green:#16a34a; --red:#dc2626; --blue:#2563eb; --amber:#f59e0b;}}
|
||
body {{ font-family: -apple-system, "Microsoft YaHei", "Segoe UI", sans-serif;
|
||
background:#f8fafc; color:#111827; margin:0; padding:20px; line-height:1.5;}}
|
||
.wrap {{ max-width: 1100px; margin: 0 auto;}}
|
||
h1,h2,h3 {{ color:#1e3a8a;}}
|
||
h1 {{ border-bottom:3px solid #1e3a8a; padding-bottom:8px;}}
|
||
h2 {{ border-bottom:1px solid #cbd5e1; padding-bottom:4px; margin-top:32px;}}
|
||
.cards {{ display:grid; grid-template-columns:repeat(auto-fit,minmax(220px,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:var(--green);}} .card .v.neg {{ color:var(--red);}}
|
||
table.data {{ border-collapse:collapse; width:100%; background:#fff; margin:8px 0 16px; font-size:13px;}}
|
||
table.data th, table.data td {{ border:1px solid #e2e8f0; padding:6px 8px; text-align:left;}}
|
||
table.data th {{ background:#f1f5f9;}}
|
||
table.data tr:nth-child(even) td {{ background:#fafbfc;}}
|
||
td.neg, tr.neg td {{ color:var(--red);}}
|
||
td.pos, tr.pos td {{ color:var(--green);}}
|
||
.muted {{ color:#6b7280; font-size:11px;}}
|
||
code {{ background:#f1f5f9; padding:1px 5px; border-radius:3px; font-size:12px;}}
|
||
.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 var(--amber); padding:10px 14px; margin:12px 0; border-radius:4px;}}
|
||
.warn {{ background:#fee2e2; border-left:4px solid var(--red); padding:10px 14px; margin:12px 0; border-radius:4px;}}
|
||
.tag {{ display:inline-block; padding:2px 8px; border-radius:10px; font-size:11px; font-weight:600; margin-left:6px;}}
|
||
.tag {{ background:#fee2e2; color:var(--red);}}
|
||
.tag.ok {{ background:#dcfce7; color:var(--green);}}
|
||
.tag.warn {{ background:#fef3c7; color:var(--amber);}}
|
||
footer {{ color:#6b7280; font-size:11px; margin-top:32px; text-align:center;}}
|
||
.rules > div {{ background:#fff; border:1px solid #e2e8f0; border-radius:6px; padding:12px 14px; margin:10px 0;}}
|
||
</style></head><body><div class="wrap">
|
||
<h1>EA 回测分析报告</h1>
|
||
<p class="muted">EA: <code>{html.escape(str(ea_name))}</code> 品种: {html.escape(str(symbol))}
|
||
报告A = <code>{html.escape(a_name)}</code> 报告B = <code>{html.escape(b_name)}</code></p>
|
||
|
||
<div class="cards">
|
||
<div class="card"><div class="k">报告A 净盈利</div><div class="v {'neg' if a_m['net_profit']<0 else 'pos'}">{a_m['net_profit']:+.2f}</div></div>
|
||
<div class="card"><div class="k">报告B 净盈利</div><div class="v {'neg' if b_m['net_profit']<0 else 'pos'}">{b_m['net_profit']:+.2f}</div></div>
|
||
<div class="card"><div class="k">PF (A / B)</div><div class="v">{a_m['profit_factor']:.2f} / {b_m['profit_factor']:.2f}</div></div>
|
||
<div class="card"><div class="k">回撤% (A / B)</div><div class="v neg">{a_s.get('max_balance_dd_pct')}% / {b_s.get('max_balance_dd_pct')}%</div></div>
|
||
</div>
|
||
|
||
<div class="note">
|
||
<b>实际交易区间</b>：报告A = {a_start} ~ {a_end}；报告B = {b_start} ~ {b_end}。<br>
|
||
<b>输入参数</b>：{'完全一致' if inputs_match else '存在差异（见下表）'}。
|
||
若两份报告参数相同而结果差异大，说明策略对行情敏感；若参数不同，需先确认对比是否公平。
|
||
</div>
|
||
|
||
{inputs_html}
|
||
|
||
<h2>1. 核心指标对比</h2>
|
||
{core_html}
|
||
<p class="muted">Sortino/Calmar 为逐笔口径估算；滚动 PF 以 100 笔窗口滑动，"滚动窗口盈利比例"=窗口 PF>1 的占比（越高越稳定）。</p>
|
||
|
||
<h2>2. 资金曲线</h2>
|
||
{equity_svg}
|
||
|
||
<h2>3. 方向性诊断</h2>
|
||
{dir_html}
|
||
{dir_svg}
|
||
<p class="muted">若多空胜率/盈亏比显著不对称，是结构性偏差信号，需结合下方 What-If 与建议规则分析。</p>
|
||
|
||
<h2>4. What-If 假设分析</h2>
|
||
<p>对每份报告模拟多种改造场景，重算指标。场景为<b>通用逻辑</b>，不依赖具体 EA 参数。
|
||
<b>重点看 PF 是否站上 1、回撤是否收敛、期望是否转正</b>。</p>
|
||
{wif_html}
|
||
<div class="warn">
|
||
<b>必看</b>："信号反向"场景若两份报告净盈利均转正且 PF>1，需警惕信号方向逻辑写反，应人工复核源码。
|
||
</div>
|
||
|
||
<h2>5. 蒙特卡洛回撤模拟</h2>
|
||
<p>将逐笔盈亏随机打乱 1000 次，看<b>顺序无关</b>下的回撤分布。
|
||
若"实际回撤"显著差于中位，说明存在<b>连败聚集</b>（顺序有自我相关），建议加连亏保护。</p>
|
||
<div class="grid2">
|
||
<div><h4>报告A 蒙特卡洛回撤</h4>{a_mc_svg}</div>
|
||
<div><h4>报告B 蒙特卡洛回撤</h4>{b_mc_svg}</div>
|
||
</div>
|
||
{mc_html}
|
||
|
||
<h2>6. 时段诊断</h2>
|
||
<h3>6.1 按小时</h3>
|
||
{hour_html}
|
||
<h3>6.2 按星期</h3>
|
||
{wd_html}
|
||
<p class="muted">⚠️ = 两份报告同时为负，属系统性劣势时段，应优先过滤。单元格颜色：红=负、绿=正。</p>
|
||
|
||
<h2>7. 持仓时间分桶</h2>
|
||
{dur_html}
|
||
<p class="muted">观察哪个持仓区间贡献正/负盈亏，可指导止盈止损时间维度调整。</p>
|
||
|
||
<h2>8. 数据驱动的优化方向建议</h2>
|
||
<p>以下建议<b>由当前数据的特征触发</b>（每条附触发条件、数据、动作），不预设任何策略特定结论。
|
||
未触发的规则不显示，未列出的维度表示数据正常。</p>
|
||
<div class="rules">
|
||
{advice_html}
|
||
</div>
|
||
|
||
<h2>9. Walk-Forward 滚动验证</h2>
|
||
<p>将每份报告的逐笔交易按时间切成滚动窗口（IS 45天 + OOS 30天，步长 30天），
|
||
计算每个窗口的 IS/OOS 指标。WFE = ΣOOS净 / ΣIS净，越高越稳健；
|
||
IS-OOS PF 相关性高=泛化好，低=过拟合风险。数据太短无法切窗会提示。</p>
|
||
<h3>报告A: {html.escape(a_name)}</h3>
|
||
{wf_a}
|
||
<h3>报告B: {html.escape(b_name)}</h3>
|
||
{wf_b}
|
||
|
||
<h2>10. 可选扩展分析</h2>
|
||
<p>以下三项已实现为独立可复用工具，按需调用：</p>
|
||
<ul>
|
||
<li><b>参数敏感度扫描</b>：<code>python param_scan.py gen-set ...</code> 生成网格 .set + 批处理脚本，
|
||
在 MT5 跑完后 <code>python param_scan.py analyze ...</code> 出响应面热力图 + 3D 曲面 + 高原检测。
|
||
<code>python param_scan.py demo</code> 可先看效果。</li>
|
||
<li><b>MAE/MFE 分析</b>：MT5 标准 xlsx 不含逐笔 MAE/MFE，需先用 <code>python mae_mfe.py --show-snippet</code>
|
||
取 MQL5 代码粘进 EA 导出 CSV，再 <code>python mae_mfe.py mae_mfe.csv</code> 出散点 + TP/SL 扫描热力图。</li>
|
||
<li><b>Walk-Forward 独立报告</b>：<code>python walk_forward.py <report.xlsx> --is-days 60 --oos-days 30</code>
|
||
出滚动窗口 IS/OOS 验证（本报告第 8 节已内嵌简化版）。</li>
|
||
</ul>
|
||
<p>其它建议（未实现，需自行扩展）：</p>
|
||
<ul>
|
||
<li><b>多品种/多周期</b>：同策略测多品种多周期，看泛化能力。</li>
|
||
<li><b>交易成本敏感度</b>：点差/手续费放大 1.5x、2x 看 PF 退化曲线，评估实盘可行性。</li>
|
||
<li><b>信号因子分解</b>：复合信号拆单因子分别回测，剔除拖累项。</li>
|
||
</ul>
|
||
|
||
<footer>由 run_analysis.py 自动生成 · 数据来自 MT5 Strategy Tester 导出的 xlsx · {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M')}</footer>
|
||
</div></body></html>
|
||
"""
|
||
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))
|