gavindiaz
ed8e8afb44
通用可复用的 MT5 回测分析工具，不针对任何具体 EA： - mt5_report_parser.py: 解析 MT5 中文 xlsx 报告，重建逐笔交易 - run_analysis.py: 主分析器，IS/OSS 对比 + What-If + 蒙特卡洛 + 数据驱动建议 - walk_forward.py: Walk-Forward 滚动 IS→OOS 验证 (WFE/泛化相关性) - param_scan.py: 参数敏感度扫描，高原检测 + 输出优化后 .set - mae_mfe.py: MAE/MFE 分析 + TP/SL 扫描，附 MQL5 导出代码 所有输出为自包含 HTML (内联 SVG，无图片依赖)。
415 lines
14 KiB
Python
415 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
MT5 策略测试报告解析器（中文版 xlsx）
|
||
=====================================
|
||
读取 MT5 Strategy Tester 导出的 .xlsx 报告，结构化为：
|
||
- meta : 元信息（EA、品种、期间、经纪商、初始资金、杠杆、输入参数）
|
||
- summary : “结果”区的汇总指标字典
|
||
- orders : 订单明细 DataFrame
|
||
- deals : 成交明细 DataFrame
|
||
- trades : 由 in/out 成交对重建的“逐笔交易”DataFrame
|
||
|
||
设计为可复用：后续新的回测报告可直接喂给本解析器。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Dict, List, Optional, Tuple
|
||
|
||
import numpy as np
|
||
import openpyxl
|
||
import pandas as pd
|
||
|
||
# 报告中区段标题（A 列出现的精确字符串）
|
||
SEC_SETTINGS = "设置"
|
||
SEC_RESULTS = "结果"
|
||
SEC_ORDERS = "订单"
|
||
SEC_DEALS = "成交"
|
||
|
||
|
||
@dataclass
|
||
class MT5Report:
|
||
meta: Dict[str, Any] = field(default_factory=dict)
|
||
summary: Dict[str, Any] = field(default_factory=dict)
|
||
orders: Optional[pd.DataFrame] = None
|
||
deals: Optional[pd.DataFrame] = None
|
||
trades: Optional[pd.DataFrame] = None
|
||
source_file: str = ""
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 工具函数
|
||
# --------------------------------------------------------------------------- #
|
||
def _find_section_rows(ws) -> Dict[str, int]:
|
||
"""扫描 A 列，定位各段标题所在行。"""
|
||
rows: Dict[str, int] = {}
|
||
for r in range(1, ws.max_row + 1):
|
||
v = ws.cell(r, 1).value
|
||
if v in (SEC_SETTINGS, SEC_RESULTS, SEC_ORDERS, SEC_DEALS):
|
||
rows[v] = r
|
||
return rows
|
||
|
||
|
||
def _kv_in_row(ws, r: int, max_col: int = 14) -> Dict[str, Any]:
|
||
"""
|
||
按行解析 “标签: 值” 对。
|
||
MT5 报告里标签以全角冒号 ':' 结尾，值位于其右侧下一个非空单元格。
|
||
一行可能含多组 标签/值。
|
||
"""
|
||
out: Dict[str, Any] = {}
|
||
cells = [ws.cell(r, c).value for c in range(1, max_col + 1)]
|
||
i = 0
|
||
while i < len(cells):
|
||
v = cells[i]
|
||
if isinstance(v, str) and v.endswith(":"):
|
||
label = v[:-1].strip()
|
||
# 向右找第一个非空值
|
||
j = i + 1
|
||
while j < len(cells) and cells[j] is None:
|
||
j += 1
|
||
if j < len(cells):
|
||
out[label] = cells[j]
|
||
i = j + 1
|
||
continue
|
||
i += 1
|
||
return out
|
||
|
||
|
||
def _parse_table(
|
||
ws, header_row: int, end_row: int, max_col: int = 14
|
||
) -> pd.DataFrame:
|
||
"""从 header_row 读取列名，读取其后到 end_row 的数据，返回 DataFrame。"""
|
||
headers = [ws.cell(header_row, c).value for c in range(1, max_col + 1)]
|
||
# 去掉末尾连续的 None 列
|
||
while headers and headers[-1] is None:
|
||
headers.pop()
|
||
ncol = len(headers)
|
||
|
||
records: List[List[Any]] = []
|
||
for r in range(header_row + 1, end_row + 1):
|
||
row = [ws.cell(r, c).value for c in range(1, ncol + 1)]
|
||
# 跳过整行空
|
||
if all(v is None for v in row):
|
||
continue
|
||
records.append(row)
|
||
|
||
df = pd.DataFrame(records, columns=headers)
|
||
return df
|
||
|
||
|
||
def _to_float(x: Any) -> Optional[float]:
|
||
"""从形如 '971.54 (96.32%)' 或 '0.69415' 的字符串中提取首个数值。"""
|
||
if x is None:
|
||
return None
|
||
if isinstance(x, (int, float)):
|
||
return float(x)
|
||
if isinstance(x, str):
|
||
m = re.search(r"-?\d+\.?\d*", x)
|
||
return float(m.group()) if m else None
|
||
return None
|
||
|
||
|
||
def _pct_to_float(x: Any) -> Optional[float]:
|
||
"""从 '96.32%' 提取 96.32。"""
|
||
if x is None:
|
||
return None
|
||
if isinstance(x, (int, float)):
|
||
return float(x)
|
||
m = re.search(r"(-?\d+\.?\d*)\s*%", str(x))
|
||
return float(m.group(1)) if m else None
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 主入口
|
||
# --------------------------------------------------------------------------- #
|
||
def parse_report(path: str) -> MT5Report:
|
||
"""解析一份 MT5 xlsx 报告。"""
|
||
wb = openpyxl.load_workbook(path, data_only=True)
|
||
ws = wb[wb.sheetnames[0]]
|
||
rep = MT5Report(source_file=path)
|
||
|
||
sec = _find_section_rows(ws)
|
||
r_settings = sec.get(SEC_SETTINGS, 1)
|
||
r_results = sec.get(SEC_RESULTS, r_settings)
|
||
r_orders = sec.get(SEC_ORDERS, r_results)
|
||
r_deals = sec.get(SEC_DEALS, r_orders)
|
||
|
||
# ---- 元信息 / 设置 ----
|
||
# 整个设置块（设置标题行之后到结果段之前）逐行解析 “标签: 值” 对
|
||
meta: Dict[str, Any] = {}
|
||
for r in range(r_settings + 1, r_results):
|
||
kv = _kv_in_row(ws, r)
|
||
for k, v in kv.items():
|
||
if k == "输入": # “输入:” 是子段标题，其值为分节字符串，跳过
|
||
continue
|
||
meta[k] = v
|
||
|
||
# 输入参数：在 “输入:” 段落里，列 D 形如 Key=Value
|
||
inputs: Dict[str, str] = {}
|
||
for r in range(r_settings + 1, r_results):
|
||
for c in (4,): # 经验上输入参数在 D 列
|
||
v = ws.cell(r, c).value
|
||
if isinstance(v, str) and "=" in v and not v.endswith(":"):
|
||
# 跳过分节标题（如 "=== PAINEL ===="）
|
||
if v.strip().startswith("==="):
|
||
continue
|
||
k, _, val = v.partition("=")
|
||
inputs[k.strip()] = val.strip()
|
||
meta["inputs"] = inputs
|
||
|
||
# ---- 结果汇总 ----
|
||
summary: Dict[str, Any] = {}
|
||
for r in range(r_results, r_orders):
|
||
summary.update(_kv_in_row(ws, r))
|
||
|
||
# ---- 订单表 ----
|
||
rep.orders = _parse_table(ws, header_row=r_orders + 1, end_row=r_deals - 1)
|
||
|
||
# ---- 成交表 ----
|
||
rep.deals = _parse_table(ws, header_row=r_deals + 1, end_row=ws.max_row)
|
||
|
||
rep.meta = meta
|
||
rep.summary = summary
|
||
rep.trades = _reconstruct_trades(rep.deals)
|
||
_normalize_summary_numbers(rep)
|
||
return rep
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 逐笔交易重建
|
||
# --------------------------------------------------------------------------- #
|
||
DEALS_COLS = {
|
||
"time": "时间",
|
||
"deal_id": "成交",
|
||
"symbol": "交易品种",
|
||
"type": "类型", # buy / sell / balance
|
||
"entry": "趋势", # in / out / None
|
||
"volume": "交易量",
|
||
"price": "价位",
|
||
"order": "订单",
|
||
"commission": "手续费",
|
||
"swap": "库存费",
|
||
"profit": "盈利",
|
||
"balance": "结余",
|
||
"comment": "注释",
|
||
}
|
||
|
||
|
||
def _reconstruct_trades(deals: pd.DataFrame) -> pd.DataFrame:
|
||
"""
|
||
由成交明细重建逐笔交易。
|
||
规则：每笔 'in' 成交后紧跟其 'out' 平仓成交，两两配对。
|
||
返回字段：open_time, close_time, direction, volume, open_price,
|
||
close_price, profit, swap, commission, net_profit, duration_min
|
||
"""
|
||
if deals is None or deals.empty:
|
||
return pd.DataFrame()
|
||
|
||
d = deals.copy()
|
||
# 统一列名
|
||
d = d.rename(columns={v: k for k, v in DEALS_COLS.items() if v in d.columns})
|
||
|
||
# 类型转换
|
||
d["time"] = pd.to_datetime(d["time"], errors="coerce")
|
||
for col in ("price", "commission", "swap", "profit", "balance"):
|
||
if col in d.columns:
|
||
d[col] = pd.to_numeric(d[col], errors="coerce")
|
||
# 成交量形如 '0.01 / 0.01'，取前半
|
||
if "volume" in d.columns:
|
||
d["volume"] = d["volume"].astype(str).str.split("/").str[0].str.strip()
|
||
d["volume"] = pd.to_numeric(d["volume"], errors="coerce")
|
||
|
||
# 仅保留 in/out 行（balance 行无 entry）
|
||
d = d[d["entry"].isin(["in", "out"])].reset_index(drop=True)
|
||
|
||
trades: List[Dict[str, Any]] = []
|
||
i = 0
|
||
while i < len(d) - 1:
|
||
in_row = d.iloc[i]
|
||
out_row = d.iloc[i + 1]
|
||
if in_row["entry"] != "in" or out_row["entry"] != "out":
|
||
i += 1
|
||
continue
|
||
direction = "short" if in_row["type"] == "sell" else "long"
|
||
profit = float(out_row.get("profit", 0.0) or 0.0)
|
||
swap = float(out_row.get("swap", 0.0) or 0.0)
|
||
commission = float(out_row.get("commission", 0.0) or 0.0)
|
||
dur = (out_row["time"] - in_row["time"]).total_seconds() / 60.0
|
||
trades.append(
|
||
{
|
||
"open_time": in_row["time"],
|
||
"close_time": out_row["time"],
|
||
"direction": direction,
|
||
"volume": in_row.get("volume"),
|
||
"open_price": in_row.get("price"),
|
||
"close_price": out_row.get("price"),
|
||
"profit": profit,
|
||
"swap": swap,
|
||
"commission": commission,
|
||
"net_profit": profit + swap + commission,
|
||
"duration_min": dur,
|
||
}
|
||
)
|
||
i += 2
|
||
|
||
return pd.DataFrame(trades)
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 汇总指标数值化（便于程序对比）
|
||
# --------------------------------------------------------------------------- #
|
||
def _normalize_summary_numbers(rep: MT5Report) -> None:
|
||
s = rep.summary
|
||
norm: Dict[str, Optional[float]] = {}
|
||
|
||
# 直接数值字段
|
||
direct = {
|
||
"总净盈利": "total_net_profit",
|
||
"毛利": "gross_profit",
|
||
"毛损": "gross_loss",
|
||
"盈利因子": "profit_factor",
|
||
"预期收益": "expected_payoff",
|
||
"夏普比率": "sharpe",
|
||
"采收率": "recovery",
|
||
"LR 相关性": "lr_correlation",
|
||
"LR 标准误差": "lr_std_error",
|
||
"总成交": "total_deals",
|
||
"交易总计": "total_trades",
|
||
}
|
||
for cn, en in direct.items():
|
||
norm[en] = _to_float(s.get(cn))
|
||
|
||
# 回撤类形如 "971.54 (96.32%)" —— 同时拆金额与百分比
|
||
dd_pairs = {
|
||
"最大结余亏损": "max_balance_dd",
|
||
"最大净值亏损": "max_equity_dd",
|
||
"相对结余亏损": "rel_balance_dd",
|
||
"相对净值亏损": "rel_equity_dd",
|
||
}
|
||
for cn, en in dd_pairs.items():
|
||
val = _to_float(s.get(cn)) # 取金额
|
||
# 单独找百分比
|
||
pct = _pct_to_float(s.get(cn))
|
||
norm[en] = val
|
||
norm[en + "_pct"] = pct
|
||
|
||
# 胜率类形如 "732 (38.28%)" —— 取百分比
|
||
wr_pairs = {
|
||
"盈利交易 (% 全部)": "win_rate_pct",
|
||
"亏损交易 (% 全部)": "loss_rate_pct",
|
||
"卖出交易 (赢得 %)": "sell_win_rate_pct",
|
||
"买入交易 (赢得 %)": "buy_win_rate_pct",
|
||
}
|
||
for cn, en in wr_pairs.items():
|
||
norm[en] = _pct_to_float(s.get(cn))
|
||
|
||
# 平均盈亏
|
||
norm["avg_win"] = _to_float(s.get("平均 获利交易"))
|
||
norm["avg_loss"] = _to_float(s.get("平均 亏损交易"))
|
||
norm["max_win"] = _to_float(s.get("最大 获利交易"))
|
||
norm["max_loss"] = _to_float(s.get("最大 亏损交易"))
|
||
|
||
rep.summary_norm = norm # type: ignore[attr-defined]
|
||
|
||
|
||
# --------------------------------------------------------------------------- #
|
||
# 便捷：从 trades 直接计算扩展指标
|
||
# --------------------------------------------------------------------------- #
|
||
def compute_trade_metrics(trades: pd.DataFrame) -> Dict[str, Any]:
|
||
"""由逐笔交易 DataFrame 计算扩展指标。"""
|
||
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]
|
||
gross_profit = wins.sum()
|
||
gross_loss = -losses.sum()
|
||
pf = gross_profit / gross_loss if gross_loss > 0 else np.inf
|
||
|
||
# 连续盈亏
|
||
streak_win = streak_loss = cur_w = cur_l = 0
|
||
max_streak_win = max_streak_loss = 0
|
||
for v in net:
|
||
if v > 0:
|
||
cur_w += 1
|
||
cur_l = 0
|
||
max_streak_win = max(max_streak_win, cur_w)
|
||
else:
|
||
cur_l += 1
|
||
cur_w = 0
|
||
max_streak_loss = max(max_streak_loss, cur_l)
|
||
|
||
# 资金曲线与回撤
|
||
equity = net.cumsum()
|
||
running_max = equity.cummax()
|
||
dd = equity - running_max
|
||
max_dd = dd.min()
|
||
|
||
# 按方向
|
||
by_dir = {}
|
||
for direction, g in t.groupby("direction"):
|
||
gn = g["net_profit"].astype(float)
|
||
w = gn[gn > 0]
|
||
l = gn[gn <= 0]
|
||
gp = w.sum()
|
||
gl = -l.sum()
|
||
by_dir[direction] = {
|
||
"n": len(g),
|
||
"win_rate": len(w) / len(g) * 100 if len(g) else 0,
|
||
"net_profit": gn.sum(),
|
||
"profit_factor": gp / gl if gl > 0 else np.inf,
|
||
"avg_win": w.mean() if len(w) else 0,
|
||
"avg_loss": l.mean() if len(l) else 0,
|
||
"expectancy": gn.mean(),
|
||
}
|
||
|
||
# 按小时 / 星期 / 月份
|
||
t2 = t.copy()
|
||
t2["hour"] = t2["open_time"].dt.hour
|
||
t2["weekday"] = t2["open_time"].dt.dayofweek # 0=Mon
|
||
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")
|
||
|
||
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_streak_win": int(max_streak_win),
|
||
"max_streak_loss": int(max_streak_loss),
|
||
"max_dd": float(max_dd),
|
||
"avg_duration_min": float(t["duration_min"].mean()),
|
||
"median_duration_min": float(t["duration_min"].median()),
|
||
"by_direction": by_dir,
|
||
"by_hour": by_hour,
|
||
"by_weekday": by_weekday,
|
||
"by_month": by_month,
|
||
"_equity": equity,
|
||
"_dd": dd,
|
||
}
|
||
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
|
||
for f in sys.argv[1:]:
|
||
rep = parse_report(f)
|
||
print(f"== {f} ==")
|
||
print("EA:", rep.meta.get("专家"))
|
||
print("symbol:", rep.meta.get("交易品种"))
|
||
print("period:", rep.meta.get("期间"))
|
||
print("inputs:", rep.meta.get("inputs"))
|
||
print("trades rows:", 0 if rep.trades is None else len(rep.trades))
|
||
print("deals rows:", 0 if rep.deals is None else len(rep.deals))
|
||
print("summary_norm sample:", rep.summary_norm) # type: ignore[attr-defined]
|
||
print()
|