SniperGold_ML/ml/parity/verify_htf_fixed.py

# -*- coding: utf-8 -*-
"""P2.5: reproduksi HTF bias EA FIXED — E1 (last closed at tc) vs E2 (cache-aware).

E2: cache Engine1 hanya di-rebuild saat Bars(HTF) berubah (bar baru muncul).
newest closed dlm cache pd tc:
  E = bar terbaru dgn time <= tc
  jika E.time + period <= tc  (E tertutup):
      jika bar setelah E sudah muncul (ht[E+1] <= tc) -> newest = E
      else (gap setelah E)                            -> newest = E-1
  jika E forming (E.time+period > tc)                  -> newest = E-1
"""
import os
import sys
import csv
import datetime as dt
import numpy as np

sys.path.insert(0, r"D:\TradingTerminal\HFM Metatrader 5\MQL5\Shared Projects\SniperGold_ML")
import train_model as TM

HERE = os.path.dirname(os.path.abspath(__file__))
DATA = os.path.normpath(os.path.join(HERE, "..", "..", "..", "..",
                                     "Files", "AlgoForge", "Data"))
EA = os.path.join(HERE, "AlgoForge_bt_features_fixed_XAUUSD_M15.csv")
PERIOD = {"D1": 86400, "H4": 14400, "H1": 3600}


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["high"].astype(np.float64),
            z["low"].astype(np.float64), z["close"].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 bias_series(hh, hl, hc):
    n = len(hc)
    out = np.zeros(n, dtype=int)
    for k in range(n):
        out[k] = TM.tf_bias_asof(hh, hl, hc, k)
    return out


def main():
    htf = {}
    for key in ("D1", "H4", "H1"):
        ht_, hh_, hl_, hc_ = load_npz("XAUUSD_" + key)
        htf[key] = (ht_, hh_, hl_, hc_)
    B = {k: bias_series(htf[k][1], htf[k][2], htf[k][3]) for k in ("D1", "H4", "H1")}

    rows = []
    with open(EA, 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]]
            except ValueError:
                continue
            rows.append((tt, fea))
    log(f"EA rows={len(rows)}")

    mis = {"E1": {k: 0 for k in ("D1", "H4", "H1")},
           "E2": {k: 0 for k in ("D1", "H4", "H1")}}
    tot = {k: 0 for k in ("D1", "H4", "H1")}
    for (tt, fea) in rows:
        tc = tt + 900
        for fi, key in enumerate(("D1", "H4", "H1")):
            ht_, hh_, hl_, hc_ = htf[key]
            per = PERIOD[key]
            ea_val = int(fea[fi])
            tot[key] += 1
            e1 = int(np.searchsorted(ht_, tc - per, side="right")) - 1
            e2 = e1
            E = int(np.searchsorted(ht_, tc, side="right")) - 1
            if 0 <= E < len(ht_):
                if ht_[E] + per <= tc:            # E tertutup
                    if E + 1 < len(ht_) and ht_[E + 1] <= tc:
                        e2 = E
                    else:
                        e2 = E - 1
                else:                             # E forming
                    e2 = E - 1
            v1 = int(B[key][e1]) if 0 <= e1 < len(hc_) else 0
            v2 = int(B[key][e2]) if 0 <= e2 < len(hc_) else 0
            if v1 != ea_val:
                mis["E1"][key] += 1
            if v2 != ea_val:
                mis["E2"][key] += 1
    log("")
    log("=== MISMATCH EA fixed vs tf_bias_asof(E) ===")
    for hyp in ("E1", "E2"):
        log(f"  {hyp}: D1={mis[hyp]['D1']}/{tot['D1']} ({mis[hyp]['D1']/tot['D1']:.4f}) "
            f"H4={mis[hyp]['H4']}/{tot['H4']} ({mis[hyp]['H4']/tot['H4']:.4f}) "
            f"H1={mis[hyp]['H1']}/{tot['H1']} ({mis[hyp]['H1']/tot['H1']:.4f})")
    return 0


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