SniperGold_ML/ml/parity/verify_window.py

# -*- 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())