Aliwkarimi98/SniperGold_ML
Watch
1
0
Fork
You've already forked SniperGold_ML
 1
forked from chiki2bum2/SniperGold_ML
Code Pull requests Activity
SniperGold_ML/ml/parity/htf_forensic.py

295 lines
12 KiB
Python
RawPermalinkNormal ViewHistory

fix: establish runtime-training feature contract and parity (P2 source-of-truth)
2026-08-21 21:00:06 +07:00
# -*- 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())
Copy permalink