# -*- coding: utf-8 -*- """P2.2 EQH/EQL adjudication: rekonstruksi semantik EA/v4.4 DetectEQ di Python. EA/v4.4 DetectEQ: - pasangan pivot BERURUTAN dalam list (g_sp[i-1], g_sp[i]), hanya bila keduanya bertipe sama (high-high / low-low) - tol = EQ_TOL_ATR * ATR SAAT INI (bar row t) - jarak bar >= EQ_BARS - sweep: bar b in (p2, total] dengan high[b] > p2.price (EQH) atau low[b] < p2.price (EQL) Python training (build_features) saat ini: - pasangan same-type BERURUTAN (melewati pivot tipe berlawanan) - tol = EQ_TOL_ATR * ATR DI BAR PIVOT Uji: rekonstruksi EA vs CSV mode 2 (f10/f11), atas pivot list full history. """ 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") 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 parse_ea_time(s): return int(dt.datetime.strptime(s, "%Y.%m.%d %H:%M") .replace(tzinfo=dt.timezone.utc).timestamp()) def eq_swept_ea(sp, h, l, A, n, eq_thr=TM.EQ_TOL_ATR, eq_bars=TM.EQ_BARS): """Rekonstruksi DetectEQ EA/v4.4. sp = list pivot kronologis [(p, price, is_high)].""" eqh = np.zeros(n, dtype=int) eql = np.zeros(n, dtype=int) for i in range(1, len(sp)): p1, pr1, h1 = sp[i - 1] p2, pr2, h2 = sp[i] if abs(p2 - p1) < eq_bars: continue # ATR basis: ATR SAAT INI = A[bar row]. Di sini A adalah series; utk flag # kumulatif dipakai A[bar sweep] - nilai tol dihitung pd bar p2 (waktu pair # terbentuk) spt EA (g_atr = ATR saat bar diproses). EA menghitung ulang # tiap row dgn g_atr terkini; rekonstruksi kumulatif: gunakan A[p2]. if h1 and h2 and abs(pr2 - pr1) <= eq_thr * A[p2]: q = np.where(h[p2 + 1:] > pr2)[0] if len(q): eqh[p2 + 1 + q[0]:] = 1 if (not h1) and (not h2) and abs(pr2 - pr1) <= eq_thr * A[p2]: q = np.where(l[p2 + 1:] < pr2)[0] if len(q): eql[p2 + 1 + q[0]:] = 1 return eqh, eql def eq_swept_ea_current(sp, h, l, A, n, eq_thr=TM.EQ_TOL_ATR, eq_bars=TM.EQ_BARS): """Varian tol = ATR pada SETIAP bar row (g_atr terkini per row).""" eqh = np.zeros(n, dtype=int) eql = np.zeros(n, dtype=int) for i in range(1, len(sp)): p1, pr1, h1 = sp[i - 1] p2, pr2, h2 = sp[i] if abs(p2 - p1) < eq_bars: continue if h1 and h2: # cari crossing pertama; flag dari crossing s.d. akhir q = np.where(h[p2 + 1:] > pr2)[0] if len(q): c = p2 + 1 + q[0] # pasangan memenuhi tol bila |pr2-pr1| <= eq_thr * A[b] utk SEMUA b >= c? # EA: tol dihitung dgn g_atr TERKINI (bar row). Flag di bar row r berlaku # bila |pr2-pr1| <= eq_thr * A[r] DAN crossing <= r. # -> pasangan dgn selisih besar baru "jadi EQH" saat ATR naik. # Rekonstruksi: eqh[r] = 1 utk r >= c bila |pr2-pr1| <= eq_thr*A[r]. eqh[c:] = 1 # koreksi: baris yg ATR-nya terlalu kecil utk memenuhi tol -> 0 eqh[c:] = (np.abs(pr2 - pr1) <= eq_thr * A[c:]).astype(int) if (not h1) and (not h2): q = np.where(l[p2 + 1:] < pr2)[0] if len(q): c = p2 + 1 + q[0] eql[c:] = (np.abs(pr2 - pr1) <= eq_thr * A[c:]).astype(int) return eqh, eql def main(): log("Load data...") t, o, h, l, c, v = load_npz("XAUUSD_M15") keep = t >= int(dt.datetime(2017, 1, 1, tzinfo=dt.timezone.utc).timestamp()) t, o, h, l, c, v = t[keep], o[keep], h[keep], l[keep], c[keep], v[keep] n = len(c) A = np.maximum(TM.atr_series(h, l, c), 1e-9) log("Build structure (swing) full history...") swing_at = np.zeros(n, dtype=int) sw = TM.build_structure(o, h, l, c, TM.SWING_LEN, False, swing_at, begin=100) sp = sw["pivots"] log(f" pivots={len(sp)}") log("Reconstruct EQH/EQL (varian A: tol ATR@pivot; varian B: tol ATR@row)...") eqhA, eqlA = eq_swept_ea(sp, h, l, A, n) eqhB, eqlB = eq_swept_ea_current(sp, h, l, A, n) # RECON-C: pair list dibatasi window EA (startBar=max(100, r-600); pivot valid # bila startBar+len <= p <= r-len), consecutive-list, tol ATR@row. # Optimasi: pair consecutive dihitung sekali dari full list (subset window tetap # consecutive), first-crossing dihitung sekali; per row hanya cek window+tol. log("RECON-C: windowed pair list (600 bar, spt EA)...") eqhC = np.zeros(n, dtype=int) eqlC = np.zeros(n, dtype=int) pairs = [] # (p1, pr1, p2, pr2, is_high, first_cross) for k in range(1, len(sp)): p1, pr1, h1 = sp[k - 1] p2, pr2, h2 = sp[k] if abs(p2 - p1) < TM.EQ_BARS: continue if h1 and h2: q = np.where(h[p2 + 1:] > pr2)[0] fc = p2 + 1 + q[0] if len(q) else n + 1 pairs.append((p1, p2, 1, fc, abs(pr2 - pr1))) if (not h1) and (not h2): q = np.where(l[p2 + 1:] < pr2)[0] fc = p2 + 1 + q[0] if len(q) else n + 1 pairs.append((p1, p2, 0, fc, abs(pr2 - pr1))) pairs = np.array(pairs, dtype=np.float64) log(f" pairs={len(pairs)}") for r in range(2 * TM.SWING_LEN, n): # cache M15 EA = 700 bar (maxBars), arrays full cache oldest-first. # begin = max(startBar, 2*len); startBar = max(100, total-600) = 100 utk total=700. # pivot valid: p_rel in [startBar-len, total-len-1] -> absolut # lo = r - total + 1 + (startBar - len), hi = r - len total = 700 start_bar = max(2 * TM.SWING_LEN, total - 600) lo = r - total + 1 + (start_bar - TM.SWING_LEN) hi = r - TM.SWING_LEN if hi <= lo: continue m = (pairs[:, 0] >= lo) & (pairs[:, 1] <= hi) & (pairs[:, 3] <= r) if m.any(): tol = TM.EQ_TOL_ATR * A[r] for row in pairs[m]: if row[4] <= tol: if row[2] == 1: eqhC[r] = 1 else: eqlC[r] = 1 log(" RECON-C done") # Python training as-is (same-type pairs, ATR@pivot) eqh_py, eql_py = np.zeros(n, dtype=int), np.zeros(n, dtype=int) lastH = lastL = None for (p, pr, is_high) in sp: tol = TM.EQ_TOL_ATR * A[p] if is_high: if lastH is not None and p - lastH[0] >= TM.EQ_BARS and abs(pr - lastH[1]) <= tol: if p + 1 < n: q = np.where(h[p + 1:] > pr)[0] if len(q): eqh_py[p + 1 + q[0]:] = 1 lastH = (p, pr) else: if lastL is not None and p - lastL[0] >= TM.EQ_BARS and abs(pr - lastL[1]) <= tol: if p + 1 < n: q = np.where(l[p + 1:] < pr)[0] if len(q): eql_py[p + 1 + q[0]:] = 1 lastL = (p, pr) log("Load EA CSV + join...") py_idx = {int(tt): i for i, tt in enumerate(t)} 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: tt = parse_ea_time(r[0].strip()) fea = [float(x) for x in r[3:22]] atr_ea = float(r[2]) except ValueError: continue if tt in py_idx: rows.append((py_idx[tt], fea, atr_ea)) log(f" joined={len(rows)}") for name, eqh, eql in (("PY-training", eqh_py, eql_py), ("RECON-A (ATR@pivot)", eqhA, eqlA), ("RECON-B (ATR@row)", eqhB, eqlB), ("RECON-C (window+ATR@row)", eqhC, eqlC)): m10 = sum(1 for bi, fea, _ in rows if eqh[bi] != int(fea[10])) m11 = sum(1 for bi, fea, _ in rows if eql[bi] != int(fea[11])) log(f" {name:22s} f10_eqh mismatch={m10}/{len(rows)} ({m10/len(rows):.4f}) " f"f11_eql mismatch={m11}/{len(rows)} ({m11/len(rows):.4f})") # sanity ATR vs EA d_atr = [abs(A[bi] - atr) for bi, _, atr in rows] log(f" sanity ATR py vs EA: max|d|={max(d_atr):.5f} mean|d|={sum(d_atr)/len(d_atr):.5f}") return 0 if __name__ == "__main__": sys.exit(main())