# -*- 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'
| 排名 | 报告 | EA | 品种 | 周期 | " "净盈利 | PF | 胜率 | 回撤% | 夏普 | 笔数 | " "
|---|
| 类型 | 报告 | 详情 |
|---|
共解析 {n} 份报告 · {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M')}