123 lines
4.5 KiB
Python
123 lines
4.5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""P2.3 Runtime context window — verifikasi formal f6/f14/f15/f17 + f3/f4/f5.
|
|
|
|
Hipotesis (dari kode EA):
|
|
- cache M15 = 700 bar, ProcessStructure begin = max(100, total-600) = 100
|
|
- pivot swing valid absolut p in [r-649, r-50]
|
|
- g_swHigh = harga pivot swing HIGH terakhir di window itu; g_swLow analog
|
|
- f6 = 2*(c-sw_low)/(sw_high-sw_low)-1 (0 bila rng=0)
|
|
- f14 = clamp((sw_high-c)/A, -10, 10) (0 bila sw_high=0)
|
|
- f15 = clamp((c-sw_low)/A, -10, 10) (0 bila sw_low=0)
|
|
- f17 = rng/A (0 bila rng=0)
|
|
- f3/f4/f5 = struktur state (trend/break) -> diuji terpisah (py_full sudah
|
|
match 0.17-0.35%; di sini dicek ulang dgn window).
|
|
"""
|
|
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")
|
|
CACHE = 700
|
|
SWING = 50
|
|
|
|
|
|
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 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 swing pivots (full history)...")
|
|
swing_at = np.zeros(n, dtype=int)
|
|
sw = TM.build_structure(o, h, l, c, SWING, False, swing_at, begin=100)
|
|
piv = np.array([(p, pr, 1 if ih else 0) for (p, pr, ih) in sw["pivots"]], dtype=np.float64)
|
|
log(f" pivots={len(piv)}")
|
|
|
|
log("Windowed sw_high/sw_low per bar (window [r-649, r-50])...")
|
|
sw_high = np.zeros(n)
|
|
sw_low = np.zeros(n)
|
|
ph = piv[piv[:, 2] == 1]
|
|
pl = piv[piv[:, 2] == 0]
|
|
# utk tiap bar r: pivot terakhir dgn p in [r-649, r-50]
|
|
# gunakan searchsorted maju: pivot dgn p <= r-50 dan >= r-649
|
|
lo_r = np.arange(n) - (CACHE - 1) + (max(100, CACHE - 600) - SWING)
|
|
hi_r = np.arange(n) - SWING
|
|
# lo_r = r - 699 + 50 = r - 649 ; hi_r = r - 50
|
|
idx_h = np.searchsorted(ph[:, 0], hi_r, side="right") - 1
|
|
okh = idx_h >= 0
|
|
sw_high[okh] = np.where(ph[idx_h[okh], 0] >= lo_r[okh], ph[idx_h[okh], 1], 0.0)
|
|
idx_l = np.searchsorted(pl[:, 0], hi_r, side="right") - 1
|
|
okl = idx_l >= 0
|
|
sw_low[okl] = np.where(pl[idx_l[okl], 0] >= lo_r[okl], pl[idx_l[okl], 1], 0.0)
|
|
|
|
# fitur
|
|
rng = sw_high - sw_low
|
|
eq_pos = np.zeros(n)
|
|
ok = (sw_high > 0) & (sw_low > 0) & (rng > 0)
|
|
eq_pos[ok] = 2.0 * (c[ok] - sw_low[ok]) / rng[ok] - 1.0
|
|
dist_high = np.zeros(n)
|
|
dist_low = np.zeros(n)
|
|
mh = sw_high > 0
|
|
dist_high[mh] = np.clip((sw_high[mh] - c[mh]) / A[mh], -10, 10)
|
|
ml = sw_low > 0
|
|
dist_low[ml] = np.clip((c[ml] - sw_low[ml]) / A[ml], -10, 10)
|
|
range_atr = np.zeros(n)
|
|
range_atr[ok] = rng[ok] / A[ok]
|
|
|
|
log("Join EA CSV...")
|
|
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]]
|
|
except ValueError:
|
|
continue
|
|
if tt in py_idx:
|
|
rows.append((py_idx[tt], fea))
|
|
log(f" joined={len(rows)}")
|
|
|
|
eps = 1e-6
|
|
for fi, name, arr in ((6, "f6_eqpos", eq_pos), (14, "f14_dhigh", dist_high),
|
|
(15, "f15_dlow", dist_low), (17, "f17_range", range_atr)):
|
|
d = [abs(arr[bi] - fea[fi]) for bi, fea in rows]
|
|
m = sum(1 for x in d if x > eps)
|
|
log(f" {name}: mismatch={m}/{len(rows)} ({m/len(rows):.4f}) max|d|={max(d):.6f}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|