# -*- coding: utf-8 -*- """P2.5: build_features_p2 — training feed dengan CORRECTED semantics (FEATURE_CONTRACT v1.0). Perbedaan vs train_model.build_features: 1. f0-f2 : bias = tf_bias_asof(E_ea), E_ea = bar HTF tertutup terakhir pada tc = t+900 (bukan as-of open dgn lag-1; bukan window tertua-200) 2. f3-f5, f8-f9 : struktur dari SLICE cache 700 berakhir di bar row, begin=100 (persis EA ProcessStructure) 3. f6/f14/f15/f17 : sw_high/sw_low = pivot swing terakhir di [t-649, t-50] (0 bila kosong) 4. f10/f11 : pasangan consecutive-list + tol = EQ_TOL_ATR * ATR(row) + window [t-649, t-50] 5. f7 (sweep), f12/f13 (delta), f16 (mom), ATR : sama dgn build_features (verified) """ import os import sys import numpy as np HERE = os.path.dirname(os.path.abspath(__file__)) SRC_TM = os.path.normpath(os.path.join(HERE, "..", "..", "..", "SniperGold_ML")) if SRC_TM not in sys.path: sys.path.insert(0, SRC_TM) import train_model as TM def _pivots_vec(h, l, length): """Pivot detection vectorized (semantik EA IsPivotHigh/Low: equal allowed).""" n = len(h) is_ph = np.ones(n, dtype=bool) is_pl = np.ones(n, dtype=bool) for k in range(1, length + 1): is_ph &= (h >= np.roll(h, k)) & (h >= np.roll(h, -k)) is_pl &= (l <= np.roll(l, k)) & (l <= np.roll(l, -k)) is_ph[:length] = False is_ph[n - length:] = False is_pl[:length] = False is_pl[n - length:] = False return is_ph, is_pl def build_structure_fast(o, h, l, c, length, internal, swing_at, begin=100): """build_structure dgn pivot vectorized (output: trend/last_break/choch).""" n = len(c) is_ph, is_pl = _pivots_vec(h, l, length) trend = np.zeros(n, dtype=int) last_break = np.full(n, -1, dtype=int) choch_dir = np.zeros(n, dtype=int) choch_bar = np.full(n, -1, dtype=int) up_target, dn_target = float("inf"), -float("inf") up_bar, dn_bar = -1, -1 cur_trend = 0 cur_cd, cur_cb = 0, -1 cur_lb = -1 for i in range(begin, n): p = i - length if p >= length and (is_ph[p] or is_pl[p]): if is_ph[p]: up_target, up_bar = h[p], p if is_pl[p]: dn_target, dn_bar = l[p], p broke = False if up_bar >= 0 and c[i] > up_target: choch = (cur_trend < 0) allow = (not internal) or swing_at[i] >= 0 if internal and allow and choch: cur_cd, cur_cb = 1, i cur_trend = 1 up_target, up_bar = float("inf"), -1 cur_lb = i broke = True if dn_bar >= 0 and c[i] < dn_target: choch = (cur_trend > 0) allow = (not internal) or swing_at[i] <= 0 if internal and allow and choch: cur_cd, cur_cb = -1, i cur_trend = -1 dn_target, dn_bar = -float("inf"), -1 cur_lb = i broke = True trend[i] = cur_trend last_break[i] = i if broke else cur_lb choch_dir[i] = cur_cd choch_bar[i] = cur_cb return dict(trend=trend, last_break=last_break, choch_dir=choch_dir, choch_bar=choch_bar) def bias_series(hh, hl, hc, lag=0): """bias[k] = BTTF window 200 bar terbaru berakhir k (k inert). tf_bias_asof(k).""" n = len(hc) out = np.zeros(n, dtype=int) for k in range(n): start = max(0, k - 199) out[k] = TM.tf_bias_asof(hh, hl, hc, k) if k >= 0 else 0 return out def build_features_p2(t, o, h, l, c, v, htf, idxs=None): """Corrected training feed utk bar indeks idxs (default semua). htf: dict {'D1': (hh, hl, hc, ht), 'H4': ..., 'H1': ...} Return F (n,19) untuk bar yg diminta (idxs), atau (n,19) utk semua. """ n = len(c) A = np.maximum(TM.atr_series(h, l, c), 1e-9) # ---- HTF bias: precompute series sekali ---- bs = {} for key, per in (("D1", 86400), ("H4", 14400), ("H1", 3600)): hh, hl, hc, ht_ = htf[key] bs[key] = bias_series(hh, hl, hc, lag=0) # ---- full-feed pivot lists (swing + internal) utk sw_high/low, EQH/EQL, sweep ---- sw_at = np.zeros(n, dtype=int) sw_full = TM.build_structure(o, h, l, c, TM.SWING_LEN, False, sw_at, begin=100) inn_full = TM.build_structure(o, h, l, c, TM.INTERNAL_LEN, True, sw_full["trend"].copy(), begin=100) sp = sw_full["pivots"] piv = np.array([(p, pr, 1 if ih else 0) for (p, pr, ih) in sp], dtype=np.float64) ph = piv[piv[:, 2] == 1] pl = piv[piv[:, 2] == 0] # sweep (f7) — full feed (verified 0.0000) sweep_dir = np.zeros(n, dtype=int) sweep_bar = np.full(n, -1) cur_sd, cur_sb = 0, -1 for (p, lvl, is_high) in inn_full["pivots"]: last = min(n - 1, p + TM.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 # EQH/EQL pairs (consecutive full-list; valid bila keduanya di window [r-649, r-50]) 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) < TM.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)) if idxs is None: idxs = np.arange(n) F = np.zeros((len(idxs), 19)) for ri, r in enumerate(idxs): # ---- f0-f2: HTF bias as-of close ---- tc = t[r] + 900 for fi, (key, per) in enumerate((("D1", 86400), ("H4", 14400), ("H1", 3600))): hh_, hl_, hc_, ht_ = htf[key] e = int(np.searchsorted(ht_, tc - per, side="right")) - 1 F[ri, fi] = int(bs[key][e]) if 0 <= e < len(hc_) else 0 # ---- struktur windowed (slice 700, begin=100) ---- s = max(0, r - 699) o_w, h_w, l_w, c_w = o[s:r + 1], h[s:r + 1], l[s:r + 1], c[s:r + 1] sw_at_w = np.zeros(len(c_w), dtype=int) sw_w = build_structure_fast(o_w, h_w, l_w, c_w, TM.SWING_LEN, False, sw_at_w, begin=100) inn_w = build_structure_fast(o_w, h_w, l_w, c_w, TM.INTERNAL_LEN, True, sw_w["trend"].copy(), begin=100) sw_t, sw_lb = int(sw_w["trend"][-1]), int(sw_w["last_break"][-1]) in_t, in_lb = int(inn_w["trend"][-1]), int(inn_w["last_break"][-1]) in_cd, in_cb = int(inn_w["choch_dir"][-1]), int(inn_w["choch_bar"][-1]) # f5 chart bias (EA ComputeAll) b5 = in_t if (in_lb >= sw_lb and in_t != 0) else sw_t if b5 == 0: b5 = sw_t if sw_t != 0 else in_t # ---- sw_high/sw_low windowed ---- lo_r = r - 649 hi_r = r - 50 sw_high = 0.0 sw_low = 0.0 jh = int(np.searchsorted(ph[:, 0], hi_r, side="right")) - 1 if jh >= 0 and ph[jh, 0] >= lo_r: sw_high = float(ph[jh, 1]) jl = int(np.searchsorted(pl[:, 0], hi_r, side="right")) - 1 if jl >= 0 and pl[jl, 0] >= lo_r: sw_low = float(pl[jl, 1]) # ---- f6/f14/f15/f17 ---- rng = sw_high - sw_low eq_pos = 2.0 * (c[r] - sw_low) / rng - 1.0 if (sw_high > 0 and sw_low > 0 and rng > 0) else 0.0 d_high = max(-10.0, min(10.0, (sw_high - c[r]) / A[r])) if sw_high > 0 else 0.0 d_low = max(-10.0, min(10.0, (c[r] - sw_low) / A[r])) if sw_low > 0 else 0.0 r_atr = rng / A[r] if rng > 0 else 0.0 # ---- f9 chochOK (bandingkan dlm indeks absolut) ---- in_cb_abs = in_cb + s if in_cb >= 0 else -1 choch_ok = 1.0 if (in_cd != 0 and in_cb_abs >= sweep_bar[r] and in_cd == sweep_dir[r]) else 0.0 # ---- f10/f11 EQH/EQL (window + ATR row) ---- eqh = 0 eql = 0 tol = TM.EQ_TOL_ATR * A[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(): eqh = 1 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(): eql = 1 # ---- f12/f13 delta ---- kk = min(TM.DELTA_BARS, r - 1) s_d = 0.0 tv = 0.0 if kk > 0: for j in range(r - kk, r): rj = h[j] - l[j] if rj <= 0: rj = 0.01 # _Point XAUUSD (EA: rng<=0 -> _Point) body = abs(c[j] - o[j]) uw = h[j] - max(o[j], c[j]) lw = min(o[j], c[j]) - l[j] bb, ss = (body + lw, uw) if c[j] >= o[j] else (lw, body + uw) tt = bb + ss if tt <= 0: tt = 0.01 s_d += (bb - ss) / tt * v[j] tv += v[j] d_dir = 1 if s_d > 0 else (-1 if s_d < 0 else 0) d_mag = max(-1.0, min(1.0, s_d / tv)) if tv > 0 else 0.0 # ---- f16 mom ---- mom = (c[r] - c[r - 20]) / A[r] if r >= 21 else 0.0 F[ri] = [F[ri, 0], F[ri, 1], F[ri, 2], sw_t, in_t, b5, eq_pos, sweep_dir[r], in_cd, choch_ok, eqh, eql, d_dir, d_mag, d_high, d_low, mom, r_atr, 0.0] F[:, 18] = TM.confluence_feature(F[:, :18]) return F # ---------------------------------------------------------------------- if __name__ == "__main__": import sys import datetime as dt HERE = os.path.dirname(os.path.abspath(__file__)) DATA = os.path.normpath(os.path.join(HERE, "..", "..", "..", "..", "Files", "AlgoForge", "Data")) 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)) 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] htf = {} for key in ("D1", "H4", "H1"): ht_, ho_, hh_, hl_, hc_, hv_ = load_npz("XAUUSD_" + key) htf[key] = (hh_, hl_, hc_, ht_) # test subset idxs = np.arange(100000, min(100200, len(c))) F = build_features_p2(t, o, h, l, c, v, htf, idxs=idxs) print("F shape:", F.shape) print("sample row 0:", F[0].round(4))