# -*- coding: utf-8 -*- """ 大批量报告汇总分析 ================== 扫描目录下所有 MT5 xlsx 报告,一次性解析并汇总成一张排名表 + 指标分布图 + 异常检测。适合参数扫描或批量回测后的综合比较。 用法: python batch_report.py reports_dir python batch_report.py reports_dir --pattern "ReportTester-*.xlsx" python batch_report.py . --glob # 用 glob 模式匹配(通配符支持多目录) """ from __future__ import annotations import argparse 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 from mt5_report_parser import parse_report, MT5Report import run_analysis as ra OUT_DIR = "output" BATCH_HTML_PATH = os.path.join(OUT_DIR, "batch_report.html") # 汇总指标列定义 SUMMARY_COLS = [ ("ea_name", "EA"), ("symbol", "品种"), ("period", "周期"), ("net_profit", "净盈利"), ("profit_factor", "PF"), ("win_rate", "胜率(%)"), ("expectancy", "期望/笔"), ("max_dd", "最大回撤"), ("max_dd_pct", "回撤(%)"), ("sharpe", "夏普"), ("sortino", "Sortino"), ("n_trades", "笔数"), ] # =========================================================================== # # 加载 + 解析 # =========================================================================== # def scan_reports( paths: List[str], pattern: str = "*.xlsx", ) -> List[str]: """ 从路径列表展开为完整文件路径列表。 每个路径可以是目录、单个文件、或 glob 模式。 支持 .xlsx 和 .htm/.html 格式。 """ result: List[str] = [] for p in paths: if os.path.isfile(p): result.append(p) elif os.path.isdir(p): # 扫描目录下的所有 xlsx 和 htm/html 文件 for ext in ("*.xlsx", "*.htm", "*.html"): result.extend(sorted(glob.glob(os.path.join(p, ext)))) elif "*" in p or "?" in p: result.extend(sorted(glob.glob(p))) else: # 尝试当作目录 for ext in ("*.xlsx", "*.htm", "*.html"): result.extend(sorted(glob.glob(os.path.join(p, ext)))) return sorted(set(result)) def parse_batch( files: List[str], *, report_pair_mode: bool = False, pair_prefix: str = "IS-", ) -> List[Tuple[str, MT5Report]]: """ 批量解析所有文件。 如果 report_pair_mode=True 且文件以 pair_prefix 开头成对出现, 则只取第二份(OSS)的报告以避免重复。 """ parsed: List[Tuple[str, MT5Report]] = [] for f in files: try: rep = parse_report(f) parsed.append((os.path.basename(f), rep)) except Exception as e: print(f" 跳过 {f}: {e}", file=sys.stderr) return parsed # =========================================================================== # # 汇总数据 # ========================================================================= # def build_summary_df( reports: List[Tuple[str, MT5Report]], ) -> pd.DataFrame: """ 汇总所有报告的关键指标为 DataFrame。 行 = 报告,列 = 指标。 """ rows: List[Dict[str, Any]] = [] for name, rep in reports: m = ra.extended_metrics(rep.trades) s = rep.summary_norm # type: ignore[attr-defined] meta = rep.meta row = { "filename": name, "ea_name": meta.get("专家", "—"), "symbol": meta.get("交易品种", "—"), "period": meta.get("期间", "—"), "net_profit": m.get("net_profit", 0.0), "profit_factor": m.get("profit_factor", 0.0), "win_rate": m.get("win_rate", 0.0), "expectancy": m.get("expectancy", 0.0), "max_dd": m.get("max_dd", 0.0), "max_dd_pct": m.get("max_dd_pct", 0.0), "sharpe": s.get("sharpe", 0.0), "sortino": m.get("sortino", 0.0), "n_trades": m.get("n_trades", 0), "avg_win": m.get("avg_win", 0.0), "avg_loss": m.get("avg_loss", 0.0), "max_streak_loss": m.get("max_streak_loss", 0), "roll_pf_min": m.get("roll_pf_min", 0.0), "roll_pf_max": m.get("roll_pf_max", 0.0), "pct_profitable_window": m.get("pct_profitable_window", 0.0), } # 参数信息(前几个输入参数) inputs = meta.get("inputs", {}) for k, v in inputs.items(): row[f"param_{k}"] = v rows.append(row) df = pd.DataFrame(rows) # 数值列类型统一(跳过字符串列) _String_Cols = {"filename", "ea_name", "symbol", "period"} for col in df.columns: if col not in _String_Cols and not col.startswith("param_"): df[col] = pd.to_numeric(df[col], errors="coerce") return df # =========================================================================== # # 异常检测 # =========================================================================== # def detect_anomalies(df: pd.DataFrame) -> List[Dict[str, Any]]: """ 标记异常报告: - PF 异常高(Top 3% 且偏离均值 > 2σ)→ 过拟合嫌疑 - 胜率异常高 → 可能数据太少 - 笔数异常少 → 统计不显著 """ flags: List[Dict[str, Any]] = [] n = len(df) if n < 3: return flags pf_mean = float(df["profit_factor"].mean()) pf_std = float(df["profit_factor"].std()) top_pct = max(1, n // 30) # Top 3% # PF 异常高 pf_sorted = df.sort_values("profit_factor", ascending=False) for _, row in pf_sorted.head(top_pct).iterrows(): if pf_std > 0 and row["profit_factor"] > pf_mean + 2 * pf_std: flags.append({ "filename": row["filename"], "type": "过拟合嫌疑", "detail": f"PF={row['profit_factor']:.2f} 远高于均值 {pf_mean:.2f} (偏离 {((row['profit_factor']-pf_mean)/pf_std):.1f}σ),且仅基于 {row['n_trades']} 笔,可能过拟合单段行情", }) # 笔数过少 for _, row in df.iterrows(): if row["n_trades"] < 30: flags.append({ "filename": row["filename"], "type": "样本不足", "detail": f"仅 {row['n_trades']} 笔交易,统计指标不可靠", }) # 回撤异常大 dd_mean = float(df["max_dd_pct"].mean()) dd_std = float(df["max_dd_pct"].std()) for _, row in df.iterrows(): if dd_std > 0 and row["max_dd_pct"] > dd_mean + 2 * dd_std and row["max_dd_pct"] > 50: flags.append({ "filename": row["filename"], "type": "回撤过大", "detail": f"回撤 {row['max_dd_pct']:.0f}% 远高于均值 {dd_mean:.0f}%", }) return flags # =========================================================================== # # SVG 分布图 # =========================================================================== # def svg_hist(items: List[Tuple[str, np.ndarray]], w: int = 520, h: int = 300) -> str: """多变量分布直方图叠加。""" top_pad = 30 bottom_pad = 45 left_pad = 55 right_pad = 16 plot_h = h - top_pad - bottom_pad plot_w = w - left_pad - right_pad colors = ["#2563eb", "#dc2626", "#16a34a", "#f59e0b", "#7c3aed"] all_min = float(np.concatenate([x for _, x in items]).min()) all_max = float(np.concatenate([x for _, x in items]).max()) if all_max == all_min: all_max = all_min + 1 def _hist(vals: np.ndarray, color: str, label: str, alpha: float = 0.7) -> str: # 过滤 NaN 和 inf clean = vals[(~np.isnan(vals)) & (np.isfinite(vals))] if len(clean) == 0: return "" n_bins = 30 counts, edges = np.histogram(clean, bins=n_bins) bin_w = (all_max - all_min) / n_bins bars = "" for i, c in enumerate(counts): if c == 0: continue x = left_pad + i * bin_w yb = top_pad + plot_h yt = top_pad + plot_h - c / max(1, max(counts)) * plot_h bars += f'' return bars bars = "" for i, (lbl, vals) in enumerate(items): color = colors[i % len(colors)] bars += _hist(vals, color, lbl) # 轴 xlabs = "" for frac in (0, 0.25, 0.5, 0.75, 1.0): xv = all_min + frac * (all_max - all_min) gx = left_pad + frac * plot_w xlabs += f'' xlabs += f'{xv:.2f}' # 图例 legend = "" for i, (lbl, _) in enumerate(items[:4]): color = colors[i % len(colors)] legend += f'' legend += f'{html.escape(lbl)}' return (f'' f'指标分布' f'{legend}{bars}{xlabs}') def svg_scatter(df: pd.DataFrame, x: str, y: str, title: str, w: int = 520, h: int = 340) -> str: """参数-指标散点图(带趋势线)。""" top_pad = 30 bottom_pad = 45 left_pad = 55 right_pad = 16 plot_h = h - top_pad - bottom_pad plot_w = w - left_pad - right_pad x_arr = pd.to_numeric(df[x], errors="coerce").to_numpy() y_arr = pd.to_numeric(df[y], errors="coerce").to_numpy() mask = ~np.isnan(x_arr) & ~np.isnan(y_arr) & np.isfinite(x_arr) & np.isfinite(y_arr) x_arr = x_arr[mask] y_arr = y_arr[mask] if len(x_arr) < 3: return "" x_min, x_max = float(x_arr.min()), float(x_arr.max()) y_min, y_max = float(y_arr.min()), float(y_arr.max()) if x_max == x_min: x_max = x_min + 1 if y_max == y_min: y_max = y_min + 1 def sx(v): return left_pad + (v - x_min) / (x_max - x_min) * plot_w def sy(v): return top_pad + plot_h - (v - y_min) / (y_max - y_min) * plot_h pts = "" for xi, yi in zip(x_arr, y_arr): col = "#16a34a" if yi > 0 else "#dc2626" pts += f'' # 趋势线 trend_line = "" try: if len(x_arr) > 2: coeffs = np.polyfit(x_arr, y_arr, 1) trend_xs = np.linspace(x_min, x_max, 50) trend_ys = np.polyval(coeffs, trend_xs) d = " ".join(f"{sx(x):.1f},{sy(y):.1f}" for x, y in zip(trend_xs, trend_ys)) trend_line = f'' except np.linalg.LinAlgError: pass # 趋势线拟合失败时跳过 xlabs = "" for frac in (0, 0.5, 1.0): xv = x_min + frac * (x_max - x_min) gx = left_pad + frac * plot_w xlabs += f'' xlabs += f'{xv:.2f}' yv = y_min + frac * (y_max - y_min) gy = top_pad + plot_h - frac * plot_h xlabs += f'' xlabs += f'{yv:.2f}' return (f'' f'{html.escape(title)}' f'{xlabs}{trend_line}{pts}' f'{html.escape(x)} →' f'{html.escape(y)} →' f'') # =========================================================================== # # 排名表 # =========================================================================== # def build_ranking_table(df: pd.DataFrame, sort_col: str, ascending: bool = False) -> str: """按指定指标排序的排名表。""" sorted_df = df.sort_values(sort_col, ascending=ascending) rows = "" for rank, (_, row) in enumerate(sorted_df.iterrows(), 1): cls = " class='pos'" if row[sort_col] > 0 else " class='neg'" if row[sort_col] < 0 else "" filename = str(row.get("filename", "")) rows += ( f"{rank}" f"{html.escape(filename)}" f"{html.escape(str(row.get('ea_name','—')))}" f"{html.escape(str(row.get('symbol','—')))}" f"{html.escape(str(row.get('period','—')))}" f"{row['net_profit']:+.2f}" f"{row['profit_factor']:.2f}" f"{row['win_rate']:.1f}%" f"{row['max_dd_pct']:.0f}%" f"{row['sharpe']:.2f}" f"{row['n_trades']}" ) return ( "" "" "" "" + rows + "
排名报告EA品种周期净盈利PF胜率回撤%夏普笔数
" ) # =========================================================================== # # HTML 报告 # =========================================================================== # def build_html(df: pd.DataFrame, anomalies: List[Dict[str, Any]]) -> str: n = len(df) pf_stats = { "mean": float(df["profit_factor"].mean()), "median": float(df["profit_factor"].median()), "std": float(df["profit_factor"].std()), "min": float(df["profit_factor"].min()), "max": float(df["profit_factor"].max()), } net_stats = { "mean": float(df["net_profit"].mean()), "std": float(df["net_profit"].std()), } # 排序选择:PF 降序(默认)、净盈利降序、胜率降序 pf_rank = build_ranking_table(df, "profit_factor", ascending=False) net_rank = build_ranking_table(df, "net_profit", ascending=False) # 分布图 hist_svg = svg_hist([ ("PF", df["profit_factor"].to_numpy()), ("净盈利", df["net_profit"].to_numpy()), ("胜率", df["win_rate"].to_numpy()), ]) scatter_svg = svg_scatter(df, "profit_factor", "win_rate", "PF vs 胜率") # 异常告警 anomaly_html = "" if anomalies: anomaly_rows = "" for a in anomalies: color = "#fee2e2" if a["type"] in ("过拟合嫌疑", "回撤过大") else "#fef3c7" fname = str(a.get("filename", "")) anomaly_rows += (f"{html.escape(a['type'])}" f"{html.escape(fname)}" f"{html.escape(a['detail'])}") anomaly_html = ( "

异常检测报告

" "" + anomaly_rows + "
类型报告详情
" ) return f""" 批量回测汇总报告

批量回测汇总报告

共解析 {n} 份报告 · {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M')}

报告总数
{n}
PF 均值
{pf_stats['mean']:.2f}
PF 中位数
{pf_stats['median']:.2f}
PF 标准差
{pf_stats['std']:.2f}
PF 范围
{pf_stats['min']:.2f}~{pf_stats['max']:.2f}
净盈利均值
{net_stats['mean']:+.2f}

1. 按 PF 排名

{pf_rank}

2. 按净盈利排名

{net_rank}

3. 指标分布

{hist_svg}
{scatter_svg}
{anomaly_html}
使用说明:点击报告名称可跳转至单份详细分析(需配合 run_analysis.py)。 异常检测标记了可能过拟合或统计不显著的报告,请优先复核。
""" # =========================================================================== # # 单份报告 HTML(辅助跳转) # =========================================================================== # def write_single_htmls( reports: List[Tuple[str, MT5Report]], base_dir: str = OUT_DIR, ) -> None: """ 为每份报告生成单份 mini 报告(可被主报告链接),避免生成 1MB+ 大文件。 仅在批量报告中提供索引跳转,不重复生成完整 HTML。 此处仅输出文件名索引信息供参考。 """ pass # =========================================================================== # # CLI # =========================================================================== # def main(argv: List[str]) -> int: ap = argparse.ArgumentParser(description="大批量报告汇总分析") ap.add_argument("paths", nargs="+", help="报告目录或 glob 路径") ap.add_argument("--pattern", default="*.xlsx", help="glob 模式(默认 *.xlsx)") args = ap.parse_args(argv) os.makedirs(OUT_DIR, exist_ok=True) files = scan_reports(args.paths, args.pattern) if not files: print(f"未找到任何报告文件,路径: {args.paths}") return 1 print(f"找到 {len(files)} 份报告,开始解析...") reports = parse_batch(files) if not reports: print("所有文件解析失败,退出。") return 1 df = build_summary_df(reports) anomalies = detect_anomalies(df) print(f"\n汇总统计:") print(f" PF 均值={df['profit_factor'].mean():.2f} 中位数={df['profit_factor'].median():.2f}") print(f" 净盈利均值={df['net_profit'].mean():+.2f}") print(f" 异常报告: {len(anomalies)} 份") html_doc = build_html(df, anomalies) with open(BATCH_HTML_PATH, "w", encoding="utf-8") as f: f.write(html_doc) print(f"\n批量汇总报告已生成: {BATCH_HTML_PATH}") return 0 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))