SniperGold_ML/ml/parity/parity_prediction.py

# -*- coding: utf-8 -*-
"""P2.5/13: PREDICTION PARITY — Python model vs MQL5 model (freeze, no retrain).

- Runtime : mode-0 dump EA FIXED (prob pd fitur corrected) = AlgoForge_bt_prob_...
- Python  : forward pass identik dgn SGML_Logit/Prob dari arrays freeze .mqh
            diterapkan pd fitur corrected (build_features_p2).
Ukur: max/mean |dp|, mismatch count/rate (eps 1e-9).
"""
import os
import re
import sys
import csv
import datetime as dt

import numpy as np

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import build_features_p2 as BFP                        # noqa: E402

DATA = os.path.normpath(os.path.join(HERE, "..", "..", "..", "..",
                                     "Files", "AlgoForge", "Data"))
MQH = os.path.normpath(os.path.join(HERE, "..", "..", "..", "..",
                                    "Include", "SniperGold_ML.mqh"))
PROB_AGENT = r"D:\TradingTerminal\HFM Metatrader 5\Tester\Agent-127.0.0.1-3000\MQL5\Files\AlgoForge_bt_prob_XAUUSD_M15.csv"
PROB = os.path.join(HERE, "AlgoForge_bt_prob_fixed_XAUUSD_M15.csv")


def log(msg):
    print(msg, flush=True)


def load_npz(name):
    z = np.load(os.path.join(DATA, name + ".npz"))
    return (z["time"].astype(np.int64), z["open"].astype(np.float64),
            z["high"].astype(np.float64), z["low"].astype(np.float64),
            z["close"].astype(np.float64), z["tick_volume"].astype(np.float64))


def parse_ea_time(s):
    return int(dt.datetime.strptime(s, "%Y.%m.%d %H:%M")
               .replace(tzinfo=dt.timezone.utc).timestamp())


def parse_arr(text, name, n):
    pat = re.compile(re.escape(name) + r"\[[^\]]*\](\[[^\]]*\])?=\{(.*)\};", re.S)
    m = pat.search(text)
    vals = [float(x) for x in re.findall(r"[-+]?[0-9]*\.?[0-9]+(?:[eE][-+]?[0-9]+)?", m.group(2))]
    return np.array(vals[:n])


def main():
    if not os.path.exists(PROB):
        import shutil
        shutil.copy2(PROB_AGENT, PROB)
        log(f"copied mode-0 dump -> {PROB}")

    txt = open(MQH, encoding="utf-8").read()
    MEAN = parse_arr(txt, "SGML_MEAN", 19)
    STD = parse_arr(txt, "SGML_STD", 19)
    W1 = parse_arr(txt, "SGML_W1", 19 * 12).reshape(19, 12)
    B1 = parse_arr(txt, "SGML_B1", 12)
    W2L = parse_arr(txt, "SGML_W2L", 12)
    B2L = float(re.search(r"SGML_B2L=([-0-9.eE+]+)", txt).group(1))
    W2S = parse_arr(txt, "SGML_W2S", 12)
    B2S = float(re.search(r"SGML_B2S=([-0-9.eE+]+)", txt).group(1))
    m = re.search(r"MathExp\(-\(o\*([-0-9.eE+]+)\+([-0-9.eE+]+)\)\)", txt)
    calL = (float(m.group(1)), float(m.group(2)))
    m2 = re.search(r"MathExp\(-\(o\*([-0-9.eE+]+)\+\-?([-0-9.eE+]+)\)\)", txt)
    # short calibration ada di baris kedua; parse manual
    sh = re.findall(r"MathExp\(-\(o\*([-0-9.eE+]+)\+([-0-9.eE+]+)\)\)", txt)
    calS = (float(sh[1][0]), float(sh[1][1]))
    log(f"calL={calL} calS={calS}")

    def prob_long(f):
        z = np.where(STD > 0, (f - MEAN) / STD, 0.0)
        h = np.maximum(0.0, B1 + z @ W1)
        o = B2L + h @ W2L
        return 1.0 / (1.0 + np.exp(-(o * calL[0] + calL[1])))

    def prob_short(f):
        z = np.where(STD > 0, (f - MEAN) / STD, 0.0)
        h = np.maximum(0.0, B1 + z @ W1)
        o = B2S + h @ W2S
        return 1.0 / (1.0 + np.exp(-(o * calS[0] + calS[1])))

    # load data + features corrected
    t, o, h, l, c, v = load_npz("XAUUSD_M15")
    keep = t >= int(dt.datetime(2017, 1, 1, tzinfo=dt.timezone.utc).timestamp())
    t, o, h, l, c, v = t[keep], o[keep], h[keep], l[keep], c[keep], v[keep]
    htf = {}
    for key in ("D1", "H4", "H1"):
        ht_, ho_, hh_, hl_, hc_, hv_ = load_npz("XAUUSD_" + key)
        htf[key] = (hh_, hl_, hc_, ht_)

    # load mode-0 dump
    rows = []
    with open(PROB, encoding="utf-8-sig") as f:
        rdr = csv.reader(f, delimiter="\t")
        next(rdr, None)
        for r in rdr:
            if len(r) < 5:
                continue
            try:
                tt = parse_ea_time(r[0].strip())
                pl = float(r[3])
                ps = float(r[4])
            except ValueError:
                continue
            rows.append((tt, pl, ps))
    log(f"mode-0 rows={len(rows)}")

    py_idx = {int(tt): i for i, tt in enumerate(t)}
    joined = [(py_idx[tt], pl, ps, tt) for (tt, pl, ps) in rows if tt in py_idx]
    log(f"joined={len(joined)}")
    idxs = np.array([j[0] for j in joined], dtype=np.int64)
    log("compute features corrected...")
    F = BFP.build_features_p2(t, o, h, l, c, v, htf, idxs=idxs)
    log("forward python model...")
    pl_py = prob_long(F)
    ps_py = prob_short(F)
    pl_ea = np.array([j[1] for j in joined])
    ps_ea = np.array([j[2] for j in joined])

    dl = np.abs(pl_py - pl_ea)
    ds = np.abs(ps_py - ps_ea)
    eps = 6e-5          # toleransi rounding CSV mode-0 (5 desimal)
    eps_strict = 1e-9
    log("")
    log("=== PREDICTION PARITY (Python vs MQL5 runtime, freeze model) ===")
    log(f"  LONG : max|dp|={dl.max():.3e} mean|dp|={dl.mean():.3e}")
    log(f"  SHORT: max|dp|={ds.max():.3e} mean|dp|={ds.mean():.3e}")
    log(f"  LONG strict(1e-9) mismatch={int((dl > eps_strict).sum())}/{len(dl)} "
        f"(mayoritas = rounding CSV 5 desimal)")
    log(f"  LONG  tol(6e-5)  mismatch={int((dl > eps).sum())}/{len(dl)} rate={float((dl > eps).mean()):.6f}")
    log(f"  SHORT tol(6e-5)  mismatch={int((ds > eps).sum())}/{len(ds)} rate={float((ds > eps).mean()):.6f}")
    for lab, d in (("LONG", dl), ("SHORT", ds)):
        over = np.nonzero(d > eps)[0]
        if len(over):
            log(f"  {lab} rows dgn |dp|>6e-5:")
            for k in over[:12]:
                log(f"    {dt.datetime.fromtimestamp(joined[k][3], dt.timezone.utc)} "
                    f"py={pl_py[k] if lab=='LONG' else ps_py[k]:.6f} "
                    f"ea={pl_ea[k] if lab=='LONG' else ps_ea[k]:.6f} d={d[k]:.2e}")
    return 0


if __name__ == "__main__":
    sys.exit(main())