199 lines
7.7 KiB
Python
199 lines
7.7 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""SMC SEMANTIC COMMON — infra bersama utk P3 SMC Semantic Golden Dataset (Fase: Liquidity Sweep).
|
|
|
|
Reuse p3_common (load data 2017+, F cache P2.6, ATR, provenance) + train_model
|
|
(build_structure) — TIDAK membuat framework duplikat.
|
|
|
|
Mengimplementasikan SEMANTIK KODE SAAT INI secara eksak (FEATURE_CONTRACT v1.0,
|
|
parity-verified thd EA AlgoForge_Backtest_Baseline.mq5 / v4.4/v4.5):
|
|
|
|
f7 sweep_dir : DetectLiquidityGrabs — pivot INTERNAL (len=5), GRAB_WINDOW=8,
|
|
wick menembus level + close kembali (REJECTION ADA), state persist.
|
|
f10 eqh_swept : DetectEQ — pasangan swing high berurutan (len=50),
|
|
|dp| <= 0.10*ATR(row), swept bila ada bar setelah p2 dgn high>pr2
|
|
(REJECTION TIDAK ADA — dipilih via AUC, lihat DESAIN_MTF_v45.md).
|
|
f11 eql_swept : DetectEQ — sama utk lows (REJECTION TIDAK ADA).
|
|
|
|
Disiplin P3: OBSERVE -> HYPOTHESIS -> TEST -> RECORD -> CLASSIFY.
|
|
Tidak ada definisi Sweep baru; hanya mengukur apa yang dilakukan kode sekarang.
|
|
"""
|
|
import os
|
|
import sys
|
|
import json
|
|
import hashlib
|
|
|
|
import numpy as np
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
sys.path.insert(0, os.path.normpath(os.path.join(HERE, ".."))) # ml/p3
|
|
sys.path.insert(0, os.path.normpath(os.path.join(HERE, "..", "..", "..", "..",
|
|
"SniperGold_ML"))) # Shared Projects\SniperGold_ML
|
|
|
|
import p3_common as P3 # data + F cache + ATR + provenance
|
|
import train_model as TM # build_structure + constants
|
|
|
|
# ---- konstanta FEATURE_CONTRACT v1.0 (sumber: train_model.py / EA #define) ----
|
|
SWING_LEN = TM.SWING_LEN # 50 (InpSwingLen, v4.4)
|
|
INTERNAL_LEN = TM.INTERNAL_LEN # 5 (InpInternalLen, v4.4)
|
|
EQ_TOL_ATR = TM.EQ_TOL_ATR # 0.10 (InpEQThreshold, v4.4)
|
|
EQ_BARS = TM.EQ_BARS # 3 (InpEQBarsConfirm, v4.4)
|
|
GRAB_WINDOW = TM.GRAB_WINDOW # 8 (InpGrabWindow, v4.4)
|
|
SEQ_WINDOW = 40 # InpSeqWindow (v4.4/v4.5): sweep -> CHoCH -> entry chain
|
|
# = event validity/expiry utk f7 (P3-S.0 f7 lifecycle fix)
|
|
ATR_PERIOD = TM.ATR_PERIOD # 14
|
|
# window pivot valid absolut utk f10/f11 & sw_high/sw_low (FEATURE_CONTRACT §0)
|
|
W_LO = 649
|
|
W_HI = 50
|
|
|
|
DECISION_TF = "M15"
|
|
SYMBOL = "XAUUSD"
|
|
|
|
|
|
def load_data():
|
|
"""t,o,h,l,c,v (M15, 2017+) + htf dict (D1/H4/H1) — sama dgn P2.6/P3."""
|
|
return P3.load_data()
|
|
|
|
|
|
def load_F():
|
|
"""F (n,19) corrected P2.6 cache — utk cross-check & stratifikasi sampling."""
|
|
return P3.load_F()
|
|
|
|
|
|
def atr_series(h, l, c):
|
|
return P3.atr_series(h, l, c)
|
|
|
|
|
|
def build_structures(o, h, l, c):
|
|
"""Swing (len=50) + internal (len=5) pivot lists, full-feed, begin=100.
|
|
Semantik identik build_features_p2 (parity-verified 1e-9 thd EA)."""
|
|
n = len(c)
|
|
sw_at = np.zeros(n, dtype=int)
|
|
sw_full = TM.build_structure(o, h, l, c, SWING_LEN, False, sw_at, begin=100)
|
|
inn_full = TM.build_structure(o, h, l, c, INTERNAL_LEN, True,
|
|
sw_full["trend"].copy(), begin=100)
|
|
return sw_full, inn_full
|
|
|
|
|
|
def sweep_state(h, l, c, inn_pivots):
|
|
"""f7 sweep_dir + sweep_bar per bar (eksak build_features_p2 / EA DetectLiquidityGrabs).
|
|
Returns (sweep_dir int[n], sweep_bar int[n]).
|
|
NOTE: ini state LEGACY (persist tanpa batas umur) — bukan output f7 final."""
|
|
n = len(c)
|
|
sweep_dir = np.zeros(n, dtype=int)
|
|
sweep_bar = np.full(n, -1, dtype=int)
|
|
cur_sd, cur_sb = 0, -1
|
|
for (p, lvl, is_high) in inn_pivots:
|
|
last = min(n - 1, p + GRAB_WINDOW)
|
|
for b in range(p + 1, last + 1):
|
|
if is_high and h[b] > lvl and c[b] < lvl:
|
|
if b > cur_sb:
|
|
cur_sd, cur_sb = -1, b
|
|
break
|
|
if (not is_high) and l[b] < lvl and c[b] > lvl:
|
|
if b > cur_sb:
|
|
cur_sd, cur_sb = 1, b
|
|
break
|
|
if cur_sb >= 0:
|
|
sweep_dir[cur_sb:] = cur_sd
|
|
sweep_bar[cur_sb:] = cur_sb
|
|
return sweep_dir, sweep_bar
|
|
|
|
|
|
def f7_lifecycle(sweep_dir, sweep_bar):
|
|
"""P3-S.0 corrected f7: event lifecycle dgn expiration SEQ_WINDOW (40 bar).
|
|
NO_SWEEP -> SWEEP_ONSET -> EXPIRED -> NO_SWEEP.
|
|
f7[r] = sweep_dir[r] jika grab onset di sweep_bar[r] dan (r - sweep_bar[r]) <= SEQ_WINDOW,
|
|
selain itu 0. Trigger condition TIDAK diubah — hanya membatasi umur state sbg event.
|
|
Return int[n]."""
|
|
n = len(sweep_dir)
|
|
out = np.zeros(n, dtype=int)
|
|
for r in range(n):
|
|
sb = int(sweep_bar[r])
|
|
if sb >= 0 and (r - sb) <= SEQ_WINDOW:
|
|
out[r] = int(sweep_dir[r])
|
|
return out
|
|
|
|
|
|
def eq_pairs(h, l, sp):
|
|
"""EQH/EQL pasangan berurutan (p1,p2,first_cross,|dp|) — eksak build_features_p2.
|
|
sp = sw_full['pivots'] : list (p, price, is_high_int)."""
|
|
n = len(h)
|
|
pairs_h = [] # (p1, p2, first_cross, |dp|)
|
|
pairs_l = []
|
|
for k in range(1, len(sp)):
|
|
p1, pr1, ih1 = sp[k - 1]
|
|
p2, pr2, ih2 = sp[k]
|
|
if abs(p2 - p1) < EQ_BARS:
|
|
continue
|
|
if ih1 and ih2:
|
|
q = np.where(h[p2 + 1:] > pr2)[0]
|
|
fc = p2 + 1 + q[0] if len(q) else n + 1
|
|
pairs_h.append((p1, p2, fc, abs(pr2 - pr1)))
|
|
if (not ih1) and (not ih2):
|
|
q = np.where(l[p2 + 1:] < pr2)[0]
|
|
fc = p2 + 1 + q[0] if len(q) else n + 1
|
|
pairs_l.append((p1, p2, fc, abs(pr2 - pr1)))
|
|
pairs_h = np.array(pairs_h, dtype=np.float64) if pairs_h else np.zeros((0, 4))
|
|
pairs_l = np.array(pairs_l, dtype=np.float64) if pairs_l else np.zeros((0, 4))
|
|
return pairs_h, pairs_l
|
|
|
|
|
|
def eq_state_at(r, pairs_h, pairs_l, A):
|
|
"""f10/f11 state + detail pair paling baru yg memenuhi window [r-649, r-50],
|
|
first_cross<=r, |dp|<=tol (tol=0.10*ATR(r)). Returns dict."""
|
|
lo_r, hi_r = r - W_LO, r - W_HI
|
|
tol = EQ_TOL_ATR * A[r]
|
|
out = {"eqh": 0, "eql": 0, "pair_h": None, "pair_l": None,
|
|
"tol_atr": float(tol), "lo_r": int(lo_r), "hi_r": int(hi_r)}
|
|
if len(pairs_h):
|
|
mh = ((pairs_h[:, 0] >= lo_r) & (pairs_h[:, 1] <= hi_r) &
|
|
(pairs_h[:, 2] <= r) & (pairs_h[:, 3] <= tol))
|
|
if mh.any():
|
|
sel = pairs_h[mh]
|
|
k = int(np.argmax(sel[:, 1])) # pair dgn p2 paling baru
|
|
out["eqh"] = 1
|
|
out["pair_h"] = [int(sel[k, 0]), int(sel[k, 1]), int(sel[k, 2]),
|
|
float(sel[k, 3])]
|
|
if len(pairs_l):
|
|
ml = ((pairs_l[:, 0] >= lo_r) & (pairs_l[:, 1] <= hi_r) &
|
|
(pairs_l[:, 2] <= r) & (pairs_l[:, 3] <= tol))
|
|
if ml.any():
|
|
sel = pairs_l[ml]
|
|
k = int(np.argmax(sel[:, 1]))
|
|
out["eql"] = 1
|
|
out["pair_l"] = [int(sel[k, 0]), int(sel[k, 1]), int(sel[k, 2]),
|
|
float(sel[k, 3])]
|
|
return out
|
|
|
|
|
|
def active_grab_at(r, sweep_dir, sweep_bar, h, l, c):
|
|
"""Detail grab internal paling baru yg aktif di bar r (f7).
|
|
Returns dict atau None bila tidak ada grab aktif."""
|
|
sb = int(sweep_bar[r])
|
|
if sb < 0:
|
|
return None
|
|
d = int(sweep_dir[r])
|
|
return {"sweep_bar": sb, "dir": d,
|
|
"sweep_dir_state": d}
|
|
|
|
|
|
def provenance():
|
|
p = P3.provenance()
|
|
p["SMC_SEMANTIC_PHASE"] = "LIQUIDITY_SWEEP"
|
|
p["contract"] = {
|
|
"SWING_LEN": SWING_LEN, "INTERNAL_LEN": INTERNAL_LEN,
|
|
"EQ_TOL_ATR": EQ_TOL_ATR, "EQ_BARS": EQ_BARS,
|
|
"GRAB_WINDOW": GRAB_WINDOW, "ATR_PERIOD": ATR_PERIOD,
|
|
"window": f"[r-{W_LO}, r-{W_HI}]",
|
|
}
|
|
return p
|
|
|
|
|
|
def save_json(name, obj):
|
|
outdir = os.path.join(HERE, "output")
|
|
os.makedirs(outdir, exist_ok=True)
|
|
path = os.path.join(outdir, name)
|
|
with open(path, "w", encoding="utf-8") as f:
|
|
json.dump(obj, f, indent=2, default=float)
|
|
print(f" [saved] {path}")
|
|
return path
|