forked from chiki2bum2/SniperGold_ML
278 lines
11 KiB
Python
278 lines
11 KiB
Python
 | # -*- 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))
|