516 lines
20 KiB
Python
516 lines
20 KiB
Python
# -*- 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'<rect x="{x:.1f}" y="{yt:.1f}" width="{bin_w:.1f}" height="{yb-yt:.1f}" fill="{color}" opacity="{alpha}"/>'
|
||
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'<line x1="{gx:.1f}" y1="{top_pad}" x2="{gx:.1f}" y2="{top_pad+plot_h}" stroke="#f1f5f9" stroke-width="0.5"/>'
|
||
xlabs += f'<text x="{gx:.1f}" y="{h-bottom_pad+14:.1f}" font-size="9" fill="#6b7280" text-anchor="middle">{xv:.2f}</text>'
|
||
|
||
# 图例
|
||
legend = ""
|
||
for i, (lbl, _) in enumerate(items[:4]):
|
||
color = colors[i % len(colors)]
|
||
legend += f'<rect x="{left_pad+i*100}" y="6" width="12" height="10" fill="{color}"/>'
|
||
legend += f'<text x="{left_pad+i*100+16}" y="14" font-size="9" fill="#374151">{html.escape(lbl)}</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">指标分布</text>'
|
||
f'{legend}{bars}{xlabs}</svg>')
|
||
|
||
|
||
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'<circle cx="{sx(xi):.1f}" cy="{sy(yi):.1f}" r="3" fill="{col}" opacity="0.65"/>'
|
||
|
||
# 趋势线
|
||
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'<polyline points="{d}" fill="none" stroke="#1e3a8a" stroke-width="1.2" stroke-dasharray="4,3" opacity="0.8"/>'
|
||
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'<line x1="{gx:.1f}" y1="{top_pad}" x2="{gx:.1f}" y2="{top_pad+plot_h}" stroke="#f1f5f9" stroke-width="0.5"/>'
|
||
xlabs += f'<text x="{gx:.1f}" y="{h-bottom_pad+14:.1f}" font-size="9" fill="#6b7280" text-anchor="middle">{xv:.2f}</text>'
|
||
yv = y_min + frac * (y_max - y_min)
|
||
gy = top_pad + plot_h - frac * plot_h
|
||
xlabs += f'<line x1="{left_pad}" y1="{gy:.1f}" x2="{w-right_pad}" y2="{gy:.1f}" stroke="#f1f5f9" stroke-width="0.5"/>'
|
||
xlabs += f'<text x="{left_pad-6:.1f}" y="{gy+3:.1f}" font-size="9" fill="#6b7280" text-anchor="end">{yv:.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">{html.escape(title)}</text>'
|
||
f'{xlabs}{trend_line}{pts}'
|
||
f'<text x="{w//2}" y="{h-4}" font-size="9" fill="#6b7280" text-anchor="middle">{html.escape(x)} →</text>'
|
||
f'<text x="14" y="{h//2}" font-size="9" fill="#6b7280" text-anchor="middle" transform="rotate(-90 14 {h//2})">{html.escape(y)} →</text>'
|
||
f'</svg>')
|
||
|
||
|
||
# =========================================================================== #
|
||
# 排名表
|
||
# =========================================================================== #
|
||
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"<tr{cls}><td>{rank}</td>"
|
||
f"<td>{html.escape(filename)}</td>"
|
||
f"<td>{html.escape(str(row.get('ea_name','—')))}</td>"
|
||
f"<td>{html.escape(str(row.get('symbol','—')))}</td>"
|
||
f"<td>{html.escape(str(row.get('period','—')))}</td>"
|
||
f"<td>{row['net_profit']:+.2f}</td>"
|
||
f"<td>{row['profit_factor']:.2f}</td>"
|
||
f"<td>{row['win_rate']:.1f}%</td>"
|
||
f"<td>{row['max_dd_pct']:.0f}%</td>"
|
||
f"<td>{row['sharpe']:.2f}</td>"
|
||
f"<td>{row['n_trades']}</td></tr>"
|
||
)
|
||
return (
|
||
"<table class='data'><thead><tr>"
|
||
"<th>排名</th><th>报告</th><th>EA</th><th>品种</th><th>周期</th>"
|
||
"<th>净盈利</th><th>PF</th><th>胜率</th><th>回撤%</th><th>夏普</th><th>笔数</th>"
|
||
"</tr></thead><tbody>" + rows + "</tbody></table>"
|
||
)
|
||
|
||
|
||
# =========================================================================== #
|
||
# 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"<tr><td style='background:{color}'>{html.escape(a['type'])}</td>"
|
||
f"<td>{html.escape(fname)}</td>"
|
||
f"<td>{html.escape(a['detail'])}</td></tr>")
|
||
anomaly_html = (
|
||
"<h3>异常检测报告</h3>"
|
||
"<table class='data'><thead><tr><th>类型</th><th>报告</th><th>详情</th></tr></thead><tbody>"
|
||
+ anomaly_rows + "</tbody></table>"
|
||
)
|
||
|
||
return f"""<!doctype html>
|
||
<html lang="zh-CN"><head><meta charset="utf-8">
|
||
<title>批量回测汇总报告</title>
|
||
<style>
|
||
body {{ font-family:-apple-system,"Microsoft YaHei","Segoe UI",sans-serif; background:#f8fafc; color:#111827; margin:0; padding:20px;}}
|
||
.wrap {{ max-width:1200px; 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:28px;}}
|
||
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;}}
|
||
table.data tr:nth-child(even) td {{ background:#fafbfc;}}
|
||
td.neg, tr.neg td {{ color:#dc2626;}} td.pos, tr.pos td {{ color:#16a34a;}}
|
||
.muted {{ color:#6b7280; font-size:11px;}}
|
||
.cards {{ display:grid; grid-template-columns:repeat(auto-fit,minmax(180px,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;}}
|
||
.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;}}
|
||
.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;}}
|
||
a {{ color:#2563eb; text-decoration:none;}} a:hover {{ text-decoration:underline;}}
|
||
footer {{ color:#6b7280; font-size:11px; margin-top:32px; text-align:center;}}
|
||
</style></head><body><div class="wrap">
|
||
<h1>批量回测汇总报告</h1>
|
||
<p class="muted">共解析 <b>{n}</b> 份报告 · {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M')}</p>
|
||
|
||
<div class="cards">
|
||
<div class="card"><div class="k">报告总数</div><div class="v">{n}</div></div>
|
||
<div class="card"><div class="k">PF 均值</div><div class="v">{pf_stats['mean']:.2f}</div></div>
|
||
<div class="card"><div class="k">PF 中位数</div><div class="v">{pf_stats['median']:.2f}</div></div>
|
||
<div class="card"><div class="k">PF 标准差</div><div class="v">{pf_stats['std']:.2f}</div></div>
|
||
<div class="card"><div class="k">PF 范围</div><div class="v">{pf_stats['min']:.2f}~{pf_stats['max']:.2f}</div></div>
|
||
<div class="card"><div class="k">净盈利均值</div><div class="v">{net_stats['mean']:+.2f}</div></div>
|
||
</div>
|
||
|
||
<h2>1. 按 PF 排名</h2>
|
||
{pf_rank}
|
||
|
||
<h2>2. 按净盈利排名</h2>
|
||
{net_rank}
|
||
|
||
<h2>3. 指标分布</h2>
|
||
<div class="grid2">
|
||
<div>{hist_svg}</div>
|
||
<div>{scatter_svg}</div>
|
||
</div>
|
||
|
||
{anomaly_html}
|
||
|
||
<div class="note">
|
||
<b>使用说明</b>：点击报告名称可跳转至单份详细分析（需配合 run_analysis.py）。
|
||
异常检测标记了可能过拟合或统计不显著的报告，请优先复核。
|
||
</div>
|
||
|
||
<footer>由 batch_report.py 生成 · {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M')}</footer>
|
||
</div></body></html>"""
|
||
|
||
|
||
# =========================================================================== #
|
||
# 单份报告 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:]))
|