SniperGold_ML/ml/parity/parity_p2.py

# -*- coding: utf-8 -*-
"""P2.5: FULL 19-FEATURE PARITY — EA corrected (BTTFBias fixed) vs build_features_p2.

Join timestamp (EA dump) -> build_features_p2 utk bar yg sama.
Per-feature: abs(runtime - training) <= eps.
  eps diskrit (f0-f5,f7-f12,f18): 1e-9 ; kontinu (f6,f13-f17): 1e-6
"""
import os
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"))
EA_FIXED = os.path.join(HERE, "AlgoForge_bt_features_fixed_XAUUSD_M15.csv")
SRC_AGENT = r"D:\TradingTerminal\HFM Metatrader 5\Tester\Agent-127.0.0.1-3000\MQL5\Files\AlgoForge_bt_features_XAUUSD_M15.csv"

FEAT = ["f0_htf1", "f1_htf2", "f2_htf3", "f3_swing", "f4_internal", "f5_bias",
        "f6_eqpos", "f7_sweep", "f8_choch", "f9_chochok", "f10_eqh", "f11_eql",
        "f12_dsign", "f13_dmag", "f14_dhigh", "f15_dlow", "f16_mom20",
        "f17_range", "f18_conf"]
EPS = [1e-9, 1e-9, 1e-9, 1e-9, 1e-9, 1e-9,
       1e-6, 1e-9, 1e-9, 1e-9, 1e-9, 1e-9,
       1e-9, 1e-6, 1e-6, 1e-6, 1e-6, 1e-6, 1e-6]


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 main():
    if not os.path.exists(EA_FIXED):
        import shutil
        shutil.copy2(SRC_AGENT, EA_FIXED)
        log(f"copied corrected dump -> {EA_FIXED}")

    log("Load data...")
    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_)

    log("Load EA corrected dump...")
    rows = []
    with open(EA_FIXED, encoding="utf-8-sig") as f:
        rdr = csv.reader(f, delimiter="\t")
        next(rdr, None)
        for r in rdr:
            if len(r) < 22:
                continue
            try:
                tt = parse_ea_time(r[0].strip())
                fea = [float(x) for x in r[3:22]]
                close_ea = float(r[1])
            except ValueError:
                continue
            rows.append((tt, fea, close_ea))
    log(f"  EA rows={len(rows)}")

    py_idx = {int(tt): i for i, tt in enumerate(t)}
    joined = [(py_idx[tt], fea, close_ea, tt) for (tt, fea, close_ea) in rows if tt in py_idx]
    missing = [tt for (tt, fea, close_ea) in rows if tt not in py_idx]
    log(f"  joined={len(joined)} missing={len(missing)}")
    if missing:
        for tt in missing[:5]:
            log(f"    missing ts: {dt.datetime.fromtimestamp(tt, dt.timezone.utc)}")

    idxs = np.array([j[0] for j in joined], dtype=np.int64)
    log("Compute build_features_p2 over parity rows...")
    F = BFP.build_features_p2(t, o, h, l, c, v, htf, idxs=idxs)
    log("  done")

    # close sanity
    close_py = c[idxs]
    d_close = np.abs(close_py - np.array([j[2] for j in joined]))
    log(f"feed sanity close: max|d|={d_close.max():.6f} mean|d|={d_close.mean():.6f}")

    log("")
    log("=== PER-FEATURE PARITY (EA fixed vs build_features_p2) ===")
    log(f"{'feature':12s} {'mismatch':>10s} {'rate':>9s} {'max|d|':>12s} {'eps':>8s}")
    all_ok = True
    n = len(joined)
    for j in range(19):
        d = np.abs(F[:, j] - np.array([jf[1][j] for jf in joined]))
        m = int((d > EPS[j]).sum())
        rate = m / n
        status = "PASS" if m == 0 else "FAIL"
        if m > 0:
            all_ok = False
        log(f"{FEAT[j]:12s} {m:>5d}/{n:<5d} {rate:9.4f} {d.max():12.6e} {EPS[j]:8.0e} {status}")

    # tanpa row pertama (artefak pre-test, computed late)
    log("")
    log("=== TANPA ROW PERTAMA (artefak 2025.12.31 20:00) ===")
    F2 = F[1:]
    for j in range(19):
        d = np.abs(F2[:, j] - np.array([jf[1][j] for jf in joined[1:]]))
        m = int((d > EPS[j]).sum())
        if m:
            log(f"  {FEAT[j]:12s} mismatch={m}/{n-1} rate={m/(n-1):.4f} max|d|={d.max():.6e} FAIL")
        else:
            log(f"  {FEAT[j]:12s} mismatch=0/{n-1} PASS")
    return 0 if all_ok else 2


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