mt5_report_parser/mae_mfe.py

# -*- coding: utf-8 -*-
"""
MAE / MFE 分析
==============
MT5 标准回测 xlsx 不含逐笔 MAE/MFE（最大不利/有利偏移），需要由 EA 在
OnDeinit 里把每笔交易的 MAE/MFE 写到 CSV。本模块：

  1. mql5_snippet()   : 返回可粘贴进 EA 的 MQL5 代码，自动导出 mae_mfe.csv
  2. analyze(df)      : 由 CSV 计算最优 TP/SL 区间
  3. build_html(df)   : 散点图 + 推荐区间

CSV 列约定（首行表头）:
    ticket,direction,open_time,close_time,open_price,close_price,
    mae_points,mfe_points,profit_money

direction: long / short
mae_points: 持仓期间最大不利偏移（点，正数表示逆向走了多少点）
mfe_points: 持仓期间最大有利偏移（点，正数）
profit_money: 该笔最终盈亏（金额）

分析原理:
  - MFE vs 盈亏散点：能涨到 X 点的笔里，最终盈利占比？→ 若大量高 MFE 笔最终
    亏损，说明 TP 太贪；把 TP 放到 MFE 高胜率区间可改善。
  - MAE vs 盈亏散点：逆向超过 Y 点的笔几乎全亏 → SL 设在 Y 内可避免大损。
  - 用扫描法找使"理论净盈利"最大的 TP/SL 组合（基于历史 MAE/MFE 重算）。

用法:
    python mae_mfe.py <mae_mfe.csv>
    python mae_mfe.py --show-snippet   # 打印 MQL5 代码
    python mae_mfe.py --demo           # 合成数据演示
"""
from __future__ import annotations

import argparse
import html
import os
import sys
from typing import Any, Dict, List, Optional, Tuple

import numpy as np
import pandas as pd

OUT_DIR = "output"


# =========================================================================== #
# 1. MQL5 代码片段
# =========================================================================== #
def mql5_snippet() -> str:
    """返回 MQL5 代码：在 OnDeinit 里把每笔历史仓位的 MAE/MFE 写入 CSV。"""
    return r'''// ====== MAE/MFE 日志片段（粘贴进 EA） ======
// 在文件顶部加:
#include <Trade\DealInfo.mqh>
#include <Trade\PositionInfo.mqh>

// 在 OnDeinit(const int reason) 末尾加:
void ExportMAEMFE()
{
    string fname = "mae_mfe_" + TimeToString(TimeCurrent(), TIME_DATE) + ".csv";
    int h = FileOpen(fname, FILE_WRITE | FILE_CSV | FILE_ANSI, ',');
    if(h == INVALID_HANDLE) { Print("MAE/MFE: 文件打开失败"); return; }
    FileWrite(h, "ticket","direction","open_time","close_time","open_price","close_price",
                 "mae_points","mfe_points","profit_money");

    HistorySelect(0, TimeCurrent() + 86400);
    int total = HistoryDealsTotal();
    // 按订单(order)配对 in/out
    for(int i = 0; i < total; i++)
    {
        ulong dealIn = HistoryDealGetTicket(i);
        if(HistoryDealGetInteger(dealIn, DEAL_ENTRY) != DEAL_ENTRY_IN) continue;

        long order = HistoryDealGetInteger(dealIn, DEAL_ORDER);
        ulong dealOut = 0;
        // 找同 order 的 out 成交
        for(int j = i + 1; j < total; j++)
        {
            ulong d = HistoryDealGetTicket(j);
            if(HistoryDealGetInteger(d, DEAL_ORDER) == order &&
               HistoryDealGetInteger(d, DEAL_ENTRY) == DEAL_ENTRY_OUT)
            { dealOut = d; break; }
        }
        if(dealOut == 0) continue;

        datetime tIn  = (datetime)HistoryDealGetInteger(dealIn,  DEAL_TIME);
        datetime tOut = (datetime)HistoryDealGetInteger(dealOut, DEAL_TIME);
        double pIn  = HistoryDealGetDouble(dealIn,  DEAL_PRICE);
        double pOut = HistoryDealGetDouble(dealOut, DEAL_PRICE);
        long   type = HistoryDealGetInteger(dealIn, DEAL_TYPE);   // DEAL_TYPE_BUY / SELL
        double profit = HistoryDealGetDouble(dealOut, DEAL_PROFIT)
                      + HistoryDealGetDouble(dealOut, DEAL_SWAP)
                      + HistoryDealGetDouble(dealOut, DEAL_COMMISSION);
        string dir = (type == DEAL_TYPE_BUY) ? "long" : "short";

        // 扫描持仓期间最高/最低价算 MAE/MFE（点）
        double mfe = 0, mae = 0;
        double pt  = SymbolInfoDouble(_Symbol, SYMBOL_POINT);
        for(datetime t = tIn; t <= tOut; t += PeriodSeconds())
        {
            double hi = iHigh(_Symbol, PERIOD_M1, iBarShift(_Symbol, PERIOD_M1, t, false));
            double lo = iLow (_Symbol, PERIOD_M1, iBarShift(_Symbol, PERIOD_M1, t, false));
            if(hi <= 0 || lo <= 0) continue;
            if(type == DEAL_TYPE_BUY)
            {
                double fav = (hi - pIn) / pt;   // 有利
                double adv = (pIn - lo) / pt;   // 不利
                if(fav > mfe) mfe = fav;
                if(adv > mae) mae = adv;
            }
            else
            {
                double fav = (pIn - lo) / pt;
                double adv = (hi - pIn) / pt;
                if(fav > mfe) mfe = fav;
                if(adv > mae) mae = adv;
            }
        }
        FileWrite(h, order, dir, TimeToString(tIn), TimeToString(tOut),
                     DoubleToString(pIn, _Digits), DoubleToString(pOut, _Digits),
                     DoubleToString(mae, 1), DoubleToString(mfe, 1),
                     DoubleToString(profit, 2));
    }
    FileClose(h);
    Print("MAE/MFE 已导出: ", fname);
}
// 在 OnDeinit 里调用: ExportMAEMFE();
// ====== 片段结束 ======
'''


# =========================================================================== #
# 2. 加载
# =========================================================================== #
def load_csv(path: str) -> pd.DataFrame:
    df = pd.read_csv(path)
    # 列名容错
    rename = {
        "mae_points": "mae", "mfe_points": "mfe",
        "profit_money": "profit",
        "open_price": "open", "close_price": "close",
    }
    df = df.rename(columns={k: v for k, v in rename.items() if k in df.columns})
    for c in ("mae", "mfe", "profit"):
        if c in df.columns:
            df[c] = pd.to_numeric(df[c], errors="coerce")
    return df


# =========================================================================== #
# 3. 分析：最优 TP/SL
# =========================================================================== #
def analyze(df: pd.DataFrame, tp_grid: Optional[List[float]] = None,
            sl_grid: Optional[List[float]] = None) -> Dict[str, Any]:
    """
    扫描 TP/SL 组合，找使理论净盈利最大的参数。
    理论模型：对每笔，若 mfe >= TP 则按 TP 离场（盈利 = TP×点值）；
              若 mae >= SL 则按 SL 离场（亏损 = -SL×点值）；
              否则按原 profit。
    这里用"点数口径"近似：盈利/亏损点数替代金额（点值统一假设）。
    """
    if df.empty:
        return {}
    if tp_grid is None:
        mfe_max = float(df["mfe"].max())
        tp_grid = list(np.linspace(0, mfe_max, 30))
    if sl_grid is None:
        mae_max = float(df["mae"].max())
        sl_grid = list(np.linspace(0, mae_max, 30))

    mae = df["mae"].to_numpy()
    mfe = df["mfe"].to_numpy()
    n = len(df)
    n_sim = len(sl_grid) * len(tp_grid)

    # 向量化：构造 (n_sim, n) 矩阵，避免 Python 嵌套循环
    mae_arr = np.broadcast_to(mae, (len(sl_grid), len(tp_grid), n)).reshape(-1, n)
    mfe_arr = np.broadcast_to(mfe, (len(sl_grid), len(tp_grid), n)).reshape(-1, n)
    sl_arr = np.tile(np.array(sl_grid)[:, np.newaxis], (1, n)).reshape(-1, n)
    tp_arr = np.tile(np.array(tp_grid)[np.newaxis, :], (len(sl_grid), n)).reshape(-1, n)

    # 优先 SL 触发（更保守）
    result_pts = np.where(mae_arr >= sl_arr, -sl_arr,
                          np.where(mfe_arr >= tp_arr, tp_arr,
                                   (mfe_arr - mae_arr) * 0.5))

    net_arr = result_pts.sum(axis=1)
    wins_arr = (result_pts > 0).sum(axis=1)
    losses_arr = -(result_pts[result_pts <= 0]).sum(axis=1)
    pf_arr = np.where(losses_arr > 0, result_pts[result_pts > 0].sum(axis=1) / losses_arr, np.inf)
    wr_arr = wins_arr / n * 100

    # 找最佳
    best_i = int(np.nanargmax(net_arr))
    grid_df = pd.DataFrame({
        "tp": np.repeat(np.array(tp_grid), len(sl_grid)),
        "sl": np.tile(np.array(sl_grid), len(tp_grid)),
        "net": net_arr,
        "win_rate": wr_arr,
        "pf": pf_arr,
    })

    best = {"tp": float(tp_grid[best_i % len(tp_grid)]),
            "sl": float(sl_grid[best_i // len(tp_grid)]),
            "net": float(net_arr[best_i]),
            "win_rate": float(wr_arr[best_i]),
            "pf": float(pf_arr[best_i])}

    # MAE/MFE 分布统计
    mae_threshold = float(df.loc[df["profit"] <= 0, "mae"].quantile(0.75)) if (df["profit"] <= 0).any() else 0
    mfe_threshold = float(df.loc[df["profit"] > 0, "mfe"].quantile(0.5)) if (df["profit"] > 0).any() else 0

    return {
        "best": best,
        "grid": grid_df,
        "mae_q3_of_losers": mae_threshold,   # 75% 亏损笔的 MAE 上限 → SL 候选
        "mfe_median_of_winners": mfe_threshold,  # 盈利笔 MFE 中位 → TP 候选
        "n": n,
        "profit_corr_mae": float(df["profit"].corr(df["mae"])),
        "profit_corr_mfe": float(df["profit"].corr(df["mfe"])),
    }


# =========================================================================== #
# 4. SVG 散点
# =========================================================================== #
def svg_scatter(x: np.ndarray, y: np.ndarray, profit: np.ndarray,
                xlabel: str, ylabel: str, title: str,
                threshold: Optional[float] = None, threshold_label: str = "",
                w: int = 460, h: int = 320) -> str:
    top, bottom, left, right = 30, 40, 50, 16
    plot_w = w - left - right
    plot_h = h - top - bottom
    x_min, x_max = float(x.min()), float(x.max())
    y_min, y_max = float(y.min()), float(y.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 + (v - x_min) / (x_max - x_min) * plot_w
    def sy(v): return top + plot_h - (v - y_min) / (y_max - y_min) * plot_h

    pts = ""
    for xi, yi, pi in zip(x, y, profit):
        col = "#16a34a" if pi > 0 else "#dc2626"
        pts += f'<circle cx="{sx(xi):.1f}" cy="{sy(yi):.1f}" r="2.2" fill="{col}" opacity="0.45"><title>{xlabel}={xi:.1f}, {ylabel}={yi:.2f}, profit={pi:.2f}</title></circle>'

    thr = ""
    if threshold is not None and x_min <= threshold <= x_max:
        tx = sx(threshold)
        thr = (f'<line x1="{tx:.1f}" y1="{top}" x2="{tx:.1f}" y2="{top+plot_h}" stroke="#1e3a8a" stroke-width="1.5" stroke-dasharray="4,3"/>'
               f'<text x="{tx+4:.1f}" y="{top+12:.1f}" font-size="9" fill="#1e3a8a">{html.escape(threshold_label)}={threshold:.1f}</text>')

    grid = ""
    for frac in (0, 0.5, 1.0):
        xv = x_min + frac * (x_max - x_min)
        yv = y_min + frac * (y_max - y_min)
        gx = left + frac * plot_w
        gy = top + plot_h - frac * plot_h
        grid += (f'<line x1="{gx:.1f}" y1="{top}" x2="{gx:.1f}" y2="{top+plot_h}" stroke="#f1f5f9" stroke-width="0.5"/>'
                 f'<text x="{gx:.1f}" y="{h-bottom+14:.1f}" font-size="9" fill="#6b7280" text-anchor="middle">{xv:.0f}</text>'
                 f'<line x1="{left}" y1="{gy:.1f}" x2="{w-right}" y2="{gy:.1f}" stroke="#f1f5f9" stroke-width="0.5"/>'
                 f'<text x="{left-6:.1f}" 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">'
            f'<text x="{w//2}" y="16" font-size="11" fill="#374151" text-anchor="middle">{html.escape(title)}</text>'
            f'{grid}{pts}{thr}'
            f'<text x="{w//2}" y="{h-4}" font-size="9" fill="#6b7280" text-anchor="middle">{html.escape(xlabel)} →</text>'
            f'<text x="14" y="{h//2}" font-size="9" fill="#6b7280" text-anchor="middle" transform="rotate(-90 14 {h//2})">{html.escape(ylabel)} →</text>'
            f'</svg>')


def svg_tp_sl_heatmap(grid_df: pd.DataFrame, w: int = 460, h: int = 360) -> str:
    """TP×SL 净盈利热力图。"""
    p = grid_df.pivot_table(index="sl", columns="tp", values="net", aggfunc="first")
    p = p.sort_index().sort_index(axis=1)
    Z = p.to_numpy()
    xs = list(p.columns)
    ys = list(p.index)
    top, bottom, left, right = 30, 40, 60, 16
    plot_w = w - left - right
    plot_h = h - top - bottom
    ny, nx = Z.shape
    cw = plot_w / nx
    ch = plot_h / ny
    valid = Z[~np.isnan(Z)]
    vmin, vmax = float(valid.min()), float(valid.max())
    if vmax == vmin: vmax = vmin + 1

    def col(v):
        if np.isnan(v): return "#f3f4f6"
        r = (v - vmin) / (vmax - vmin)
        if r < 0.5:
            t = r * 2
            return f"rgb({int(220-60*t)},{int(60+160*t)},{int(60+40*t)})"
        t = (r - 0.5) * 2
        return f"rgb({int(160-100*t)},{int(220-40*t)},{int(100-40*t)})"

    cells = ""
    for j in range(ny):
        for i in range(nx):
            v = Z[j, i]
            x = left + i * cw
            y = top + (ny - 1 - j) * ch
            cells += f'<rect x="{x:.1f}" y="{y:.1f}" width="{cw:.1f}" height="{ch:.1f}" fill="{col(v)}" stroke="#fff" stroke-width="0.4"/>'
    # 最优点
    bi = np.unravel_index(np.nanargmax(Z), Z.shape)
    bx = left + bi[1] * cw + cw / 2
    by = top + (ny - 1 - bi[0]) * ch + ch / 2
    marker = f'<circle cx="{bx:.1f}" cy="{by:.1f}" r="6" fill="none" stroke="#1e3a8a" stroke-width="2"/>'

    xlabs = "".join(f'<text x="{left+(i+0.5)*cw:.1f}" y="{h-bottom+14:.1f}" font-size="8" fill="#374151" text-anchor="middle">{xs[i]:.0f}</text>' for i in range(0, nx, 2))
    ylabs = "".join(f'<text x="{left-6:.1f}" y="{top+(ny-0.5-j)*ch+3:.1f}" font-size="8" fill="#374151" text-anchor="end">{ys[j]:.0f}</text>' for j in range(0, ny, 2))
    return (f'<svg viewBox="0 0 {w} {h}" class="chart">'
            f'<text x="{w//2}" y="16" font-size="11" fill="#374151" text-anchor="middle">理论净盈利 (TP × SL 扫描) — 圈=最优</text>'
            f'{cells}{marker}{xlabs}{ylabs}'
            f'<text x="{w//2}" y="{h-4}" font-size="9" fill="#6b7280" text-anchor="middle">TP(点) →</text>'
            f'<text x="14" y="{h//2}" font-size="9" fill="#6b7280" text-anchor="middle" transform="rotate(-90 14 {h//2})">SL(点) →</text>'
            f'</svg>')


# =========================================================================== #
# 5. HTML
# =========================================================================== #
def build_html(df: pd.DataFrame, res: Dict[str, Any]) -> str:
    mae_svg = svg_scatter(
        df["mae"].to_numpy(), df["profit"].to_numpy(), df["profit"].to_numpy(),
        "MAE(点)", "盈利($)", "MAE vs 盈亏（绿=赢/红=输）",
        threshold=res["mae_q3_of_losers"], threshold_label="SL候选",
    )
    mfe_svg = svg_scatter(
        df["mfe"].to_numpy(), df["profit"].to_numpy(), df["profit"].to_numpy(),
        "MFE(点)", "盈利($)", "MFE vs 盈亏（绿=赢/红=输）",
        threshold=res["mfe_median_of_winners"], threshold_label="TP候选",
    )
    tpsl_svg = svg_tp_sl_heatmap(res["grid"])
    b = res["best"]

    cards = (
        "<div class='cards'>"
        f"<div class='card'><div class='k'>样本笔数</div><div class='v'>{res['n']}</div></div>"
        f"<div class='card'><div class='k'>最优 TP(点)</div><div class='v'>{b['tp']:.0f}</div></div>"
        f"<div class='card'><div class='k'>最优 SL(点)</div><div class='v'>{b['sl']:.0f}</div></div>"
        f"<div class='card'><div class='k'>理论净盈利(点)</div><div class='v pos'>{b['net']:.0f}</div></div>"
        f"<div class='card'><div class='k'>理论胜率</div><div class='v'>{b['win_rate']:.1f}%</div></div>"
        f"<div class='card'><div class='k'>理论 PF</div><div class='v'>{b['pf']:.2f}</div></div>"
        "</div>"
    )

    interp = ""
    if res["profit_corr_mae"] < -0.3:
        interp += f"<p class='muted'>盈亏与 MAE 负相关 ({res['profit_corr_mae']:.2f})：逆向偏移越大越易亏，设 SL 合理。SL 候选 = 亏损笔 MAE 75 分位 ≈ <b>{res['mae_q3_of_losers']:.0f} 点</b>。</p>"
    if res["profit_corr_mfe"] > 0.3:
        interp += f"<p class='muted'>盈亏与 MFE 正相关 ({res['profit_corr_mfe']:.2f})：能跑到 X 点的笔更易盈利，TP 候选 = 盈利笔 MFE 中位 ≈ <b>{res['mfe_median_of_winners']:.0f} 点</b>。</p>"
    if not interp:
        interp = "<p class='muted'>MAE/MFE 与盈亏相关性弱，TP/SL 优化空间有限，建议关注入场逻辑。</p>"

    interp += (f"<div class='note'><b>理论最优 TP/SL</b>：TP={b['tp']:.0f}点 / SL={b['sl']:.0f}点，"
               f"在该组合下重算净盈利 {b['net']:.0f}点、胜率 {b['win_rate']:.1f}%、PF {b['pf']:.2f}。"
               f"此为基于历史 MAE/MFE 的理论值，需回测验证。</div>")

    return f"""<!doctype html><html lang="zh-CN"><head><meta charset="utf-8">
<title>MAE/MFE 分析</title>
<style>
  body {{ font-family:-apple-system,"Microsoft YaHei",sans-serif; background:#f8fafc; color:#111827; margin:0; padding:20px;}}
  .wrap {{ max-width:1100px; margin:0 auto;}}
  h1,h2 {{ color:#1e3a8a;}} h1 {{ border-bottom:3px solid #1e3a8a; padding-bottom:8px;}}
  .muted {{ color:#6b7280; font-size:12px;}}
  .cards {{ display:grid; grid-template-columns:repeat(auto-fit,minmax(150px,1fr)); gap:12px; margin:16px 0;}}
  .card {{ background:#fff; padding:12px; border-radius:8px; border:1px solid #e2e8f0;}}
  .card .k {{ color:#6b7280; font-size:11px;}} .card .v {{ font-size:18px; font-weight:600; margin-top:4px;}}
  .card .v.pos {{ color:#16a34a;}}
  .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 #f59e0b; padding:10px 14px; margin:12px 0; border-radius:4px;}}
  pre {{ background:#1e293b; color:#e2e8f0; padding:14px; border-radius:6px; overflow:auto; font-size:11px;}}
  footer {{ color:#6b7280; font-size:11px; margin-top:32px; text-align:center;}}
</style></head><body><div class="wrap">
<h1>MAE / MFE 分析报告</h1>
<p class='muted'>样本: {res['n']} 笔 · 盈亏-MAE 相关 {res['profit_corr_mae']:.2f} · 盈亏-MFE 相关 {res['profit_corr_mfe']:.2f}</p>
{cards}
{interp}
<h2>1. MAE / MFE 散点</h2>
<div class="grid2"><div>{mae_svg}</div><div>{mfe_svg}</div></div>
<p class='muted'>蓝虚线 = 推荐的 SL/TP 候选阈值。MAE 图：右侧红点=逆向走很远还亏=该早止损；MFE 图：右侧绿点=涨得高最终赚=TP 可放到此处。</p>
<h2>2. TP × SL 扫描热力图</h2>
{tpsl_svg}
<p class='muted'>圈=理论最优组合。色越绿=净盈利越高。注意"宽绿区"比"单点深绿"更稳健。</p>
<h2>3. 如何获取 MAE/MFE 数据</h2>
<p class='muted'>MT5 标准 xlsx 报告不含逐笔 MAE/MFE。需在 EA 的 OnDeinit 里加如下代码导出 CSV，再喂给本工具：</p>
<pre>{html.escape(mql5_snippet())}</pre>
<footer>由 mae_mfe.py 生成 · {pd.Timestamp.now().strftime('%Y-%m-%d %H:%M')}</footer>
</div></body></html>"""


# =========================================================================== #
# 6. demo
# =========================================================================== #
def _demo_df() -> pd.DataFrame:
    rng = np.random.default_rng(3)
    n = 400
    direction = rng.choice(["long", "short"], n)
    mae = np.abs(rng.normal(20, 15, n))
    mfe = np.abs(rng.normal(40, 20, n))
    # 盈亏与 mfe 正相关、与 mae 负相关
    base = (mfe - mae) * 0.3 + rng.normal(0, 5, n)
    profit = np.where(base > 0, mfe * 0.2, -mae * 0.25) + rng.normal(0, 2, n)
    return pd.DataFrame({
        "ticket": np.arange(n),
        "direction": direction,
        "mae": np.round(mae, 1),
        "mfe": np.round(mfe, 1),
        "profit": np.round(profit, 2),
    })


# =========================================================================== #
# CLI
# =========================================================================== #
def main(argv: List[str]) -> int:
    ap = argparse.ArgumentParser(description="MAE/MFE 分析")
    ap.add_argument("csv", nargs="?", help="mae_mfe.csv 路径")
    ap.add_argument("--show-snippet", action="store_true", help="打印 MQL5 代码片段")
    ap.add_argument("--demo", action="store_true")
    args = ap.parse_args(argv)

    os.makedirs(OUT_DIR, exist_ok=True)

    if args.show_snippet:
        print(mql5_snippet())
        return 0

    if args.demo:
        df = _demo_df()
    else:
        if not args.csv:
            ap.error("需要 csv 路径，或 --demo，或 --show-snippet")
        df = load_csv(args.csv)

    res = analyze(df)
    html_doc = build_html(df, res)
    out = os.path.join(OUT_DIR, "mae_mfe.html")
    with open(out, "w", encoding="utf-8") as f:
        f.write(html_doc)
    print(f"HTML 报告: {out}")
    print(f"最优 TP={res['best']['tp']:.0f}点  SL={res['best']['sl']:.0f}点  "
          f"理论净盈利={res['best']['net']:.0f}点  PF={res['best']['pf']:.2f}")
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))