Warrior_EA/research/reconcile.py

"""
Trade-by-trade reconciliation: WarriorDipZ.ex5 (MT5 tester) vs research/backtest.py.

A research result is a claim about the research code until the EA that will
actually trade reproduces it. This matches each EA trade to the backtest trade
on the same symbol whose fill bar contains the EA's entry time, then reports
what agrees, what does not, and why.

It also reads the tester report's EQUITY drawdown -- marked to market, with
concurrent positions counted together -- which the backtest's exit-based
curve can only bound from below.
"""

from __future__ import annotations

import re
import sys
import numpy as np

sys.path.insert(0, __file__.rsplit("\\", 1)[0] if "\\" in __file__ else ".")

from vol_filter_test import collect_filtered  # noqa: E402

COMMON = r"C:\Users\admin\AppData\Roaming\MetaQuotes\Terminal\Common\Files"
TERM = r"C:\Users\admin\AppData\Roaming\MetaQuotes\Terminal\10CE948A1DFC9A8C27E56E827008EBD4"
BAR = np.timedelta64(4 * 3600, "s")


def load_ea(path=rf"{COMMON}\dipz_trades.csv"):
    raw = np.genfromtxt(path, delimiter=",", skip_header=1, dtype=str, encoding="ansi")
    if raw.ndim == 1:
        raw = raw[None, :]
    to_ts = lambda s: np.datetime64(s[:10].replace(".", "-") + "T" + s[11:] + ":00")  # noqa: E731
    return [dict(pos=r[0], symbol=r[1], t=to_ts(r[2]), entry=float(r[3]), vol=float(r[4]),
                 exit_t=to_ts(r[5]), exit=float(r[6]), net=float(r[7]), reason=r[8])
            for r in raw]


def report_stats(name):
    """Pull headline numbers out of the tester's HTML report (UTF-16)."""
    path = rf"{TERM}\{name}.htm"
    try:
        txt = open(path, encoding="utf-16").read()
    except (OSError, UnicodeError):
        try:
            txt = open(path, encoding="utf-8", errors="ignore").read()
        except OSError:
            return {}
    txt = re.sub(r"<[^>]+>", "|", txt)
    txt = re.sub(r"\|+", "|", txt)
    out = {}
    for key in ("Total Net Profit", "Profit Factor", "Equity Drawdown Maximal",
                "Balance Drawdown Maximal", "Total Trades", "Expected Payoff",
                "Equity Drawdown Relative"):
        m = re.search(re.escape(key) + r":\|([^|]+)\|", txt)
        if m:
            out[key] = m.group(1).strip()
    return out


def main(symbols, report, lo="2022-01-01", deposit=100000.0, risk=0.0025, gated=True):
    ea = load_ea()
    gate = (lambda p: np.nan_to_num(p, nan=-1) >= 0.50) if gated else (lambda p: np.ones(len(p), bool))
    py = [t for t in collect_filtered(symbols, "PERIOD_H4", gate, lo, None)]

    # match: same symbol, EA entry inside the python fill bar
    used = set()
    pairs, ea_only = [], []
    for e in ea:
        hit = None
        for k, p in enumerate(py):
            if k in used or p["symbol"] != e["symbol"]:
                continue
            if p["t"] <= e["t"] < p["t"] + BAR:
                hit = k
                break
        if hit is None:
            ea_only.append(e)
        else:
            used.add(hit)
            pairs.append((e, py[hit]))
    py_only = [p for k, p in enumerate(py) if k not in used]

    print(f"EA trades {len(ea)}   backtest trades {len(py)}   matched {len(pairs)} "
          f"({len(pairs) / max(len(py), 1):.0%} of backtest)   EA-only {len(ea_only)}   "
          f"backtest-only {len(py_only)}\n")

    if pairs:
        er = np.array([(e["exit"] - e["entry"]) / e["entry"] for e, _ in pairs])
        pr = np.array([p["gross"] for _, p in pairs])
        same_exit_bar = np.mean([abs(e["exit_t"] - p["exit_t"]) <= BAR for e, p in pairs])
        rmap = {"stop": "stop", "expert": None}
        agree = np.mean([(e["reason"] == "stop") == (p["reason"] == "stop") for e, p in pairs])
        print(f"matched trades: gross return corr {np.corrcoef(er, pr)[0, 1]:.3f}   "
              f"mean |diff| {np.mean(np.abs(er - pr)) * 1e4:.1f} bp   "
              f"exit within one bar {same_exit_bar:.0%}   stop/non-stop agreement {agree:.0%}")
        print(f"                mean gross: EA {er.mean() * 1e4:+.1f} bp   backtest {pr.mean() * 1e4:+.1f} bp\n")

    for tag, rows in (("EA-only", ea_only[:8]), ("backtest-only", py_only[:8])):
        if rows:
            print(f"first {tag}:")
            for r in rows:
                print(f"   {r['symbol']:<7} {str(r['t'])[:16]}  -> {str(r['exit_t'])[:16]}")
            print()

    # EA portfolio stats from its own P&L (exit-based, comparable to backtest)
    ea_sorted = sorted(ea, key=lambda e: e["exit_t"])
    eq = deposit + np.cumsum([e["net"] for e in ea_sorted])
    eq = np.concatenate([[deposit], eq])
    dd = ((np.maximum.accumulate(eq) - eq) / np.maximum.accumulate(eq)).max()
    years = (ea_sorted[-1]["exit_t"] - ea_sorted[0]["t"]) / np.timedelta64(365, "D")
    tot = eq[-1] / deposit - 1
    print(f"EA portfolio: {len(ea)} trades, {len(ea) / (years * 12):.1f}/mo, total {tot:+.1%}, "
          f"CAGR {(1 + tot) ** (1 / years) - 1:+.1%}, exit-based maxDD {dd:.1%}, ret/DD {tot / dd:.2f}")
    for split in ("2024-01-01",):
        for nm, sel in (("IS ", [e for e in ea_sorted if e["t"] < np.datetime64(split)]),
                        ("OOS", [e for e in ea_sorted if e["t"] >= np.datetime64(split)])):
            if not sel:
                continue
            q = deposit + np.concatenate([[0], np.cumsum([e["net"] for e in sel])])
            d_ = ((np.maximum.accumulate(q) - q) / np.maximum.accumulate(q)).max()
            y_ = (sel[-1]["exit_t"] - sel[0]["t"]) / np.timedelta64(365, "D")
            t_ = q[-1] / deposit - 1
            print(f"   {nm} {len(sel):>4} trades  {len(sel) / (y_ * 12):>5.1f}/mo  total {t_:+.1%}  "
                  f"maxDD {d_:.1%}  ret/DD {t_ / d_ if d_ > 0 else float('nan'):.2f}")
    rep = report_stats(report)
    if rep:
        print("\ntester report (EQUITY drawdown is marked to market - the real one):")
        for k, v in rep.items():
            print(f"   {k:<28} {v}")


if __name__ == "__main__":
    syms = sys.argv[1].split(",") if len(sys.argv) > 1 else ["SP500", "NAS100", "US30", "DAX40"]
    rep = sys.argv[2] if len(sys.argv) > 2 else "claude_dipz_4"
    main(syms, rep)