# -*- coding: utf-8 -*- """ P2.1 HTF bias forensic — hypothesis test vs EA runtime dump. Tujuan: mengidentifikasi SEMANTIK WINDOW BTTFBias yang benar-benar dipakai EA runtime (AlgoForge_Backtest_Baseline.mq5 mode 2) pada tiap row M15, dan membandingkannya dengan semantik training (train_model.build_features). Fakta runtime yang sudah diverifikasi dari kode EA: - Row ditulis pada t = open time bar M15 tertutup terbaru (e1.Time(sM15,0)). - Fitur dihitung pada tick pertama bar M15 baru -> waktu komputasi tc = t + 900s. - Cache Engine 1 = HANYA bar tertutup (closed-bar lock), kapasitas 250 (D1/H4/H1). - BTTFBias: need = min(200, e1.Count(slot)); need < 120 -> "Neutral" (0). - Tester model=0 (every tick), FromDate=2026.01.01 00:00 UTC -> cache TF tinggi KOSONG di awal test (tidak ada history sebelum FromDate di tester). Hipotesis semantik window (per HTF, per row t): E_py = containing_htf(t) - 1 (semantik training: lag-1, as-of open t) E_ea = last closed HTF bar at tc (semantik runtime: as-of close tc) H_PY : tf_bias_asof(E_py) # training as-is H_CLOSED : tf_bias_asof(E_ea) # full history, window ends E_ea H_ANCHOR : window [anchor .. E_ea], need=min(200,count), need<120 -> 0 H_ANCHOR250: sama + cap cache 250 (need=min(200,min(250,count))) Output: - mismatch rate tiap hipotesis vs f0/f1/f2 EA (atas semua row parity) - detail deterministik utk beberapa timestamp terpilih """ import os import sys import csv import datetime as dt import numpy as np HERE = os.path.dirname(os.path.abspath(__file__)) SRC_TM = os.path.normpath(os.path.join(HERE, "..", "..", "..", "SniperGold_ML")) sys.path.insert(0, SRC_TM) import train_model as TM # noqa: E402 DATA = os.path.normpath(os.path.join(HERE, "..", "..", "..", "..", "Files", "AlgoForge", "Data")) EA_CSV = os.path.join(HERE, "AlgoForge_bt_features_XAUUSD_M15.csv") ANCHOR = int(dt.datetime(2026, 1, 1, tzinfo=dt.timezone.utc).timestamp()) TRIM = int(dt.datetime(2025, 1, 1, tzinfo=dt.timezone.utc).timestamp()) # cukup utk 200-bar window D1 PERIOD = {"D1": 86400, "H4": 14400, "H1": 3600} NEED_MAX = 200 NEED_MIN = 120 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 bttf_window(h, l, c, need): """BTTFBias inti (s=3 fractal + break), window h/l/c (0=tertua), need=len.""" if need < NEED_MIN: 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_full(hh, hl, hc, lag): """bias[k] = bttf window ending at k-lag (full history, need=200).""" 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_MAX + 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 parse_pick_time(s): return int(dt.datetime.strptime(s, "%Y-%m-%d %H:%M") .replace(tzinfo=dt.timezone.utc).timestamp()) def main(): log("Load data + trim (>= 2025-01-01)...") htf = {} for key in ("D1", "H4", "H1"): ht, ho, hh, hl, hc, hv = load_npz("XAUUSD_" + key) m = ht >= TRIM htf[key] = (ht[m], ho[m], hh[m], hl[m], hc[m]) log(f" {key}: n={int(m.sum())} first={dt.datetime.fromtimestamp(int(ht[m][0]), dt.timezone.utc)}") log("Load M15 (full, 2017+) + EA CSV...") mt, mo, mh, ml, mc, mv = load_npz("XAUUSD_M15") keep = mt >= int(dt.datetime(2017, 1, 1, tzinfo=dt.timezone.utc).timestamp()) mt, mo, mh, ml, mc, mv = mt[keep], mo[keep], mh[keep], ml[keep], mc[keep], mv[keep] 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" EA rows={len(rows)} M15(2017+) n={len(mc)}") # ---- precompute bias arrays per HTF ---- log("Precompute bias arrays...") B = {} # B[key][hyp] = array bias per bar index HYP = ["PY", "CLOSED", "ANCHOR", "ANCHOR250"] for key in ("D1", "H4", "H1"): ht, ho, hh, hl, hc = htf[key] B[key] = {} B[key]["PY"] = bias_full(hh, hl, hc, lag=1) log(f" {key} PY done") B[key]["CLOSED"] = bias_full(hh, hl, hc, lag=0) log(f" {key} CLOSED done") a_idx = int(np.searchsorted(ht, ANCHOR, side="left")) n = len(hc) for hyp, cap in (("ANCHOR", None), ("ANCHOR250", 250)): out = np.zeros(n, dtype=int) for k in range(a_idx, n): cnt = k - a_idx + 1 if cap is not None: cnt = min(cnt, cap) need = min(NEED_MAX, cnt) if need < NEED_MIN: continue start = max(a_idx, k - NEED_MAX + 1) out[k] = bttf_window(hh[start:k + 1], hl[start:k + 1], hc[start:k + 1], need) B[key][hyp] = out log(f" {key} {hyp} done") # ---- verifikasi pipeline: hitung py_full persis spt harness (TM.build_features) ---- log("Verifikasi vs TM.build_features (harness py_full)...") htf_full = {} for key in ("D1", "H4", "H1"): z = np.load(os.path.join(DATA, "XAUUSD_" + key + ".npz")) ht_, ho_, hh_, hl_, hc_ = (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)) htf_full[key] = (hh_, hl_, hc_, ht_) F, label, A, _ = TM.build_features((mt, mo, mh, ml, mc, mv), htf_full) py_idx = {int(tt): i for i, tt in enumerate(mt)} diff_py = {k: 0 for k in ("D1", "H4", "H1")} nchk = 0 for (t, feats) in rows: bi = py_idx.get(t) if bi is None: continue nchk += 1 for fi, key in enumerate(("D1", "H4", "H1")): if int(F[bi][fi]) != int(B[key]["PY"][int(np.searchsorted(htf[key][0], t, side="right")) - 1]): diff_py[key] += 1 log(f" [check] PY-mapping vs harness py_full: n={nchk} diff D1={diff_py['D1']} H4={diff_py['H4']} H1={diff_py['H1']}") # ---- uji agregasi: HTF dari M15 (harus == npz HTF bila feed tester konsisten) ---- log("Uji agregasi HTF dari M15 vs npz HTF...") for key, per in (("H1", 3600), ("H4", 14400), ("D1", 86400)): bucket = (mt // per) * per idx_sort = np.argsort(bucket, kind="stable") bs, a_o = bucket[idx_sort], mo[idx_sort] a_h = mh[idx_sort] a_l = ml[idx_sort] a_c = mc[idx_sort] uniq, starts = np.unique(bs, return_index=True) ends = np.append(starts[1:], len(bs)) g_o = np.array([a_o[s] for s in starts]) g_h = np.array([a_h[s:e].max() for s, e in zip(starts, ends)]) g_l = np.array([a_l[s:e].min() for s, e in zip(starts, ends)]) g_c = np.array([a_c[e - 1] for e in ends]) ht_, ho_, hh_, hl_, hc_ = htf[key][0], htf[key][1], htf[key][2], htf[key][3], htf[key][4] common = np.intersect1d(uniq, ht_) jj = np.searchsorted(uniq, common) jn = np.searchsorted(ht_, common) d_o = float(np.max(np.abs(g_o[jj] - ho_[jn]))) d_h = float(np.max(np.abs(g_h[jj] - hh_[jn]))) d_l = float(np.max(np.abs(g_l[jj] - hl_[jn]))) d_c = float(np.max(np.abs(g_c[jj] - hc_[jn]))) log(f" {key}: M15-agg vs npz max|d| open={d_o:.5f} high={d_h:.5f} low={d_l:.5f} " f"close={d_c:.5f} (n={len(jn)})") # ---- per row: nilai EA vs tiap hipotesis ---- log("Evaluate hypotheses per row...") mis = {h: {k: 0 for k in ("D1", "H4", "H1")} for h in HYP} tot = {k: 0 for k in ("D1", "H4", "H1")} best = {k: {h: 0 for h in HYP + ["NONE"]} for k in ("D1", "H4", "H1")} for (t, feats) in rows: tc = t + 900 for fi, key in enumerate(("D1", "H4", "H1")): ht, ho, hh, hl, hc = htf[key] ea_val = int(feats[fi]) per = PERIOD[key] cidx = int(np.searchsorted(ht, t, side="right")) - 1 e_ea = int(np.searchsorted(ht, tc - per, side="right")) - 1 tot[key] += 1 vals = {} if 0 <= cidx < len(hc): vals["PY"] = int(B[key]["PY"][cidx]) else: vals["PY"] = 0 if 0 <= e_ea < len(hc): vals["CLOSED"] = int(B[key]["CLOSED"][e_ea]) vals["ANCHOR"] = int(B[key]["ANCHOR"][e_ea]) vals["ANCHOR250"] = int(B[key]["ANCHOR250"][e_ea]) else: vals["CLOSED"] = vals["ANCHOR"] = vals["ANCHOR250"] = 0 for h in HYP: if vals[h] != ea_val: mis[h][key] += 1 hit = [h for h in HYP if vals[h] == ea_val] best[key][hit[0] if hit else "NONE"] += 1 log("") log("=== MISMATCH RATE vs EA (f0/f1/f2) ===") log(f"{'hypothesis':10s} {'D1(f0)':>10s} {'H4(f1)':>10s} {'H1(f2)':>10s}") for h in HYP: log(f"{h:10s} {mis[h]['D1']/tot['D1']:10.4f} {mis[h]['H4']/tot['H4']:10.4f} {mis[h]['H1']/tot['H1']:10.4f}") log("") log("=== BEST MATCH per row (fraksi row yg cocok dgn tiap hipotesis) ===") log(f"{'hypothesis':10s} {'D1(f0)':>10s} {'H4(f1)':>10s} {'H1(f2)':>10s}") for h in HYP + ["NONE"]: log(f"{h:10s} {best['D1'][h]/tot['D1']:10.4f} {best['H4'][h]/tot['H4']:10.4f} {best['H1'][h]/tot['H1']:10.4f}") # ---- detail deterministik ---- picks = [ "2026-01-05 12:00", "2026-03-18 01:00", "2026-04-01 00:00", "2026-06-16 13:30", "2026-07-20 15:45", ] pick_ts = [parse_pick_time(p) for p in picks] log("") log("=== DETAIL TIMESTAMP TERPILIH (PY vs CLOSED vs ANCHOR vs EA) ===") for t in pick_ts: tc = t + 900 log(f"row t={dt.datetime.fromtimestamp(t, dt.timezone.utc)} tc={dt.datetime.fromtimestamp(tc, dt.timezone.utc)}") for fi, key in enumerate(("D1", "H4", "H1")): ht, ho, hh, hl, hc = htf[key] per = PERIOD[key] cidx = int(np.searchsorted(ht, t, side="right")) - 1 e_ea = int(np.searchsorted(ht, tc - per, side="right")) - 1 log(f" {key:3s} contain={cidx}({dt.datetime.fromtimestamp(int(ht[cidx]), dt.timezone.utc).strftime('%m-%d')}) " f"E_ea={e_ea}({dt.datetime.fromtimestamp(int(ht[e_ea]), dt.timezone.utc).strftime('%m-%d')}) " f"PY={int(B[key]['PY'][cidx])} CLOSED={int(B[key]['CLOSED'][e_ea])} " f"ANCHOR={int(B[key]['ANCHOR'][e_ea])} ANCHOR250={int(B[key]['ANCHOR250'][e_ea])} " f"EA={int(feats[fi])}") return 0 if __name__ == "__main__": sys.exit(main())