SniperGold_ML/ml/parity/verify_bttf_rootcause.py

# -*- coding: utf-8 -*-
"""P2.1 VERIFIKASI ROOT CAUSE: EA BTTFBias = window 200 bar TERTUA dari cache 250 bar.

Hipotesis (terbukti dari FULLDUMP + isolate):
  EA: GetBar(slot, count-1-i) utk i=0..need-1  (count=250, need=200)
      -> window = bars[249..50] = 200 bar TERTUA cache = tf_bias_asof(E_ea - 50)
  di mana E_ea = bar HTF tertutup terakhir pd waktu komputasi tc = t+900.

Uji: EA_hat = tf_bias_asof(E_ea - 50) vs f0/f1/f2 CSV (mode 2) utk SEMUA row.
"""
import os
import sys
import csv
import datetime as dt

import numpy as np

HERE = os.path.dirname(os.path.abspath(__file__))
DATA = os.path.normpath(os.path.join(HERE, "..", "..", "..", "..",
                                     "Files", "AlgoForge", "Data"))
EA_CSV = os.path.join(HERE, "AlgoForge_bt_features_XAUUSD_M15.csv")
PERIOD = {"D1": 86400, "H4": 14400, "H1": 3600}
CACHE = 250
NEED = 200
LAG = CACHE - NEED          # 50


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 bttf_window(h, l, c, need):
    if need < 120:
        return 0
    s = 3
    m = need - 1
    up, dn = float("inf"), -float("inf")
    upb = dnb = -1
    trend = 0
    for i in range(s + 1, m):
        p = i - s
        if p >= s:
            isH = isL = True
            for kk in range(1, s + 1):
                if h[p] <= h[p + kk] or h[p] <= h[p - kk]:
                    isH = False
                if l[p] >= l[p + kk] or l[p] >= l[p - kk]:
                    isL = False
            if isH:
                up, upb = h[p], p
            if isL:
                dn, dnb = l[p], p
        if upb >= 0 and c[i] > up:
            trend = 1
            up, upb = float("inf"), -1
        if dnb >= 0 and c[i] < dn:
            trend = -1
            dn, dnb = -float("inf"), -1
    return trend


def bias_series(hh, hl, hc, lag):
    """bias[k] = bttf window 200 bar ending at k-lag (>= series start)."""
    n = len(hc)
    out = np.zeros(n, dtype=int)
    for k in range(n):
        end = k - lag
        if end < 0:
            continue
        start = max(0, end - NEED + 1)
        out[k] = bttf_window(hh[start:end + 1], hl[start:end + 1],
                             hc[start:end + 1], end - start + 1)
    return out


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


def main():
    log("Load data...")
    htf = {}
    for key in ("D1", "H4", "H1"):
        ht, hh, hl, hc = load_npz("XAUUSD_" + key)
        m = ht >= int(dt.datetime(2024, 1, 1, tzinfo=dt.timezone.utc).timestamp())
        htf[key] = (ht[m], hh[m], hl[m], hc[m])
        log(f"  {key}: n={int(m.sum())}")

    log("Load EA CSV (mode 2)...")
    rows = []
    with open(EA_CSV, 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:
                t = parse_ea_time(r[0].strip())
                feats = [float(x) for x in r[3:22]]
            except ValueError:
                continue
            rows.append((t, feats))
    log(f"  rows={len(rows)}")

    log("Precompute bias (lag=50) per HTF...")
    B = {}
    for key in ("D1", "H4", "H1"):
        ht, hh, hl, hc = htf[key]
        B[key] = bias_series(hh, hl, hc, LAG)
        log(f"  {key} done")

    log("Evaluate EA_hat = tf_bias_asof(E_ea - 50) vs EA...")
    mis = {k: 0 for k in ("D1", "H4", "H1")}
    tot = {k: 0 for k in ("D1", "H4", "H1")}
    first_mis = {k: 0 for k in ("D1", "H4", "H1")}
    for ri, (t, feats) in enumerate(rows):
        tc = t + 900
        for fi, key in enumerate(("D1", "H4", "H1")):
            ht, hh, hl, hc = htf[key]
            ea_val = int(feats[fi])
            per = PERIOD[key]
            e_ea = int(np.searchsorted(ht, tc - per, side="right")) - 1
            if 0 <= e_ea < len(hc):
                v = int(B[key][e_ea])
            else:
                v = 0
            tot[key] += 1
            if v != ea_val:
                mis[key] += 1
                if ri == 0:
                    first_mis[key] += 1
                print(f"  MISMATCH row={ri} t={r[0] if False else dt.datetime.fromtimestamp(t, dt.timezone.utc)} "
                      f"key={key} EA={ea_val} hat={v} e_ea={dt.datetime.fromtimestamp(int(ht[e_ea]), dt.timezone.utc)}")
    log("")
    log("=== MISMATCH EA_hat(lag=50) vs EA CSV ===")
    for key in ("D1", "H4", "H1"):
        log(f"  {key}: mismatch={mis[key]}/{tot[key]} rate={mis[key]/tot[key]:.4f} (first-row-mis={first_mis[key]})")
    return 0


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