295 lines
12 KiB
Python
295 lines
12 KiB
Python
|
# -*- 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())
|