SniperGold_ML/ml/p3/smc_semantic/test_f7_lifecycle.py

# -*- coding: utf-8 -*-
"""REGRESSION TESTS — f7 event lifecycle (P3-S.0).

R1 — Single onset     : satu sweep -> 1 event (bukan multiple)
R2 — Persistent cond  : kondisi sweep bertahan N bar -> 1 event (bukan N)
R3 — New independent  : sweep baru setelah expired -> event baru = 1
R4 — Invalidated      : event yg sudah expired TIDAK aktif sbg event
R5 — No-sweep         : tanpa sweep -> 0 event
R6 — Existing suite   : f10/f11 TIDAK berubah vs F cache P2.6; EA compile 0 error;
                        satu-satunya perubahan = f7 lifecycle

Plus HISTORICAL VALIDATION (section 10):
  BEFORE : f7 legacy 196.928 bar aktif / 2.835 onset
  AFTER  : f7 corrected (lifecycle) bar aktif & event runs
  Case   : perbandingan 60 golden cases ada di f7_v1_vs_v2_comparison.json

Usage: python test_f7_lifecycle.py
"""
import os
import sys
import json

import numpy as np

HERE = os.path.dirname(os.path.abspath(__file__))
sys.path.insert(0, HERE)
import smc_semantic_common as SC

PASS = []
FAIL = []


def check(name, cond, detail=""):
    (PASS if cond else FAIL).append(name)
    print(f"  [{'PASS' if cond else 'FAIL'}] {name}  {detail}")


# =====================================================================
# R1-R5: unit test f7_lifecycle (pure function)
# =====================================================================
def test_r1_single_onset():
    # satu grab di bar 10; state legacy tetap 1 sampai akhir
    n = 200
    sd = np.zeros(n, dtype=int)
    sb = np.full(n, -1, dtype=int)
    sd[10:] = 1
    sb[10:] = 10
    f7 = SC.f7_lifecycle(sd, sb)
    active = f7 != 0
    events = int(np.sum(np.diff(np.concatenate(([0], active.astype(int)))) == 1))
    active_bars = int(active.sum())
    check("R1 single onset -> 1 event", events == 1,
          f"events={events} active_bars={active_bars} (expect 1 event)")
    check("R1 active bounded", active_bars <= SC.SEQ_WINDOW + 1,
          f"active_bars={active_bars} <= SEQ_WINDOW+1={SC.SEQ_WINDOW + 1}")


def test_r2_persistent_condition():
    # kondisi "benar" selama N bar (state tetap 1), harus tetap 1 event
    n = 500
    sd = np.zeros(n, dtype=int)
    sb = np.full(n, -1, dtype=int)
    sd[50:] = -1
    sb[50:] = 50
    f7 = SC.f7_lifecycle(sd, sb)
    active = f7 != 0
    events = int(np.sum(np.diff(np.concatenate(([0], active.astype(int)))) == 1))
    check("R2 persistent condition -> 1 event (bukan N)", events == 1,
          f"events={events} (condition true for 450 bars)")


def test_r3_new_independent_sweep():
    # grab1 di bar 10, expired (41 bar), grab2 di bar 100 -> 2 event terpisah
    n = 300
    sd = np.zeros(n, dtype=int)
    sb = np.full(n, -1, dtype=int)
    sd[10:60] = 1
    sb[10:60] = 10
    sd[100:] = -1
    sb[100:] = 100
    f7 = SC.f7_lifecycle(sd, sb)
    active = f7 != 0
    events = int(np.sum(np.diff(np.concatenate(([0], active.astype(int)))) == 1))
    check("R3 new independent sweep -> 2 events", events == 2,
          f"events={events} (expect 2)")


def test_r4_invalidated_expired():
    # grab di bar 0, r=100 -> age 100 > SEQ_WINDOW -> f7 = 0
    n = 200
    sd = np.full(n, 1, dtype=int)
    sb = np.full(n, 0, dtype=int)
    f7 = SC.f7_lifecycle(sd, sb)
    check("R4 expired event not active", f7[100] == 0 and f7[10] == 1,
          f"f7[10]={f7[10]} f7[100]={f7[100]} (age 100 > 40)")


def test_r5_no_sweep():
    sd = np.zeros(300, dtype=int)
    sb = np.full(300, -1, dtype=int)
    f7 = SC.f7_lifecycle(sd, sb)
    check("R5 no sweep -> 0 events", int((f7 != 0).sum()) == 0)


# =====================================================================
# R6: f10/f11 unchanged + EA source contains fix + only f7 changed
# =====================================================================
def test_r6_existing_suite():
    t, o, h, l, c, v, htf = SC.load_data()
    F = SC.load_F()
    A = SC.atr_series(h, l, c)
    n = len(c)
    sw_full, inn_full = SC.build_structures(o, h, l, c)
    pairs_h, pairs_l = SC.eq_pairs(h, l, sw_full["pivots"])

    # f10/f11 annotator state vs F cache (sample bars utk kecepatan)
    rng = np.random.RandomState(1).choice(np.arange(1500, n - 100), size=4000, replace=False)
    mism10 = mism11 = 0
    for r in rng:
        st = SC.eq_state_at(int(r), pairs_h, pairs_l, A)
        if st["eqh"] != int(F[r, 10]):
            mism10 += 1
        if st["eql"] != int(F[r, 11]):
            mism11 += 1
    check("R6 f10 unchanged vs F cache", mism10 == 0, f"mismatch={mism10}/4000")
    check("R6 f11 unchanged vs F cache", mism11 == 0, f"mismatch={mism11}/4000")

    # EA source memuat fix lifecycle
    ea = open(r"D:\TradingTerminal\HFM Metatrader 5\MQL5\Experts\AlgoForge_Backtest_Baseline.mq5",
              encoding="utf-8", errors="replace").read()
    check("R6 EA source contains f7 lifecycle fix",
          "AF_BT_SEQ_WINDOW" in ea and "(tc-1-g_swpBar)<=AF_BT_SEQ_WINDOW" in ea)
    # v4.4/v4.5
    for p in (r"D:\TradingTerminal\HFM Metatrader 5\MQL5\Indicators\Downloads\SniperGold_SMC_ProPlus_v4_4.mq5",
              r"D:\TradingTerminal\HFM Metatrader 5\MQL5\Indicators\Downloads\SniperGold_SMC_ProPlus_v4_5.mq5"):
        src = open(p, encoding="utf-8", errors="replace").read()
        check(f"R6 {os.path.basename(p)} contains fix",
              "g_swpBar>=0 && (tc-1-g_swpBar)" in src)
    print("  [note] R6 EA/v4.4/v4.5 compile 0 error: diverifikasi manual via MetaEditor "
          "(0 errors, 0 warnings)")


# =====================================================================
# HISTORICAL VALIDATION (section 10)
# =====================================================================
def historical_validation():
    t, o, h, l, c, v, htf = SC.load_data()
    F = SC.load_F()
    n = len(c)
    sw_full, inn_full = SC.build_structures(o, h, l, c)
    sd, sb = SC.sweep_state(h, l, c, inn_full["pivots"])
    f7_new = SC.f7_lifecycle(sd, sb)
    f7_old = F[:, 7].astype(int)

    onset = len(np.unique(sb[sb >= 0]))
    old_active = int((f7_old != 0).sum())
    new_active = int((f7_new != 0).sum())
    # event runs
    def runs(x):
        r = []
        cur = x[0]
        ln = 1
        for i in range(1, n):
            if x[i] == cur:
                ln += 1
            else:
                if cur != 0:
                    r.append(ln)
                cur, ln = x[i], 1
        if cur != 0:
            r.append(ln)
        return np.array(r) if r else np.array([0])

    ro, rn = runs(f7_old), runs(f7_new)

    # per-onset lifetime dlm output corrected: utk tiap onset b, jumlah bar r dgn
    # sweep_bar[r]==b dan f7_new[r]!=0. Ini WAJIB <= SEQ_WINDOW+1 (41) — bukti
    # bahwa SATU onset tidak menghasilkan TRUE selama ratusan bar.
    lifetimes = []
    for b in np.unique(sb[sb >= 0]):
        b = int(b)
        m = (sb == b) & (f7_new != 0)
        lifetimes.append(int(m.sum()))
    lifetimes = np.array(lifetimes) if lifetimes else np.array([0])

    res = {
        "onsets": int(onset),
        "before_active_bars": old_active,
        "before_pct": round(100 * old_active / n, 2),
        "after_active_bars": new_active,
        "after_pct": round(100 * new_active / n, 2),
        "before_runs": int(len(ro)),
        "after_runs": int(len(rn)),
        "before_run_median": int(np.median(ro)),
        "after_run_median": int(np.median(rn)) if len(rn) else None,
        "before_run_max": int(ro.max()),
        "after_run_max": int(rn.max()) if len(rn) else None,
        "per_onset_lifetime_max": int(lifetimes.max()) if len(lifetimes) else 0,
        "per_onset_lifetime_median": int(np.median(lifetimes)) if len(lifetimes) else 0,
        "events_removed_bars": old_active - new_active,
        "note": "run max > 41 = rantai event (grab baru dlm window menyegarkan); "
                "per-onset lifetime tetap <= 41",
    }
    print("\nHISTORICAL VALIDATION (XAUUSD M15 2017-2026, n=%d):" % n)
    for k, vv in res.items():
        print(f"  {k}: {vv}")
    check("HIST before: persistent state (99.95%)", old_active > 0.95 * n)
    check("HIST after: active bars turun drastis", new_active < 0.6 * old_active,
          f"{new_active} < 0.6*{old_active}")
    check("HIST after: SATU onset lifetime <= SEQ_WINDOW+1",
          res["per_onset_lifetime_max"] <= SC.SEQ_WINDOW + 1,
          f"max per-onset lifetime={res['per_onset_lifetime_max']} (<=41)")
    check("HIST after: run max turun", res["after_run_max"] < res["before_run_max"],
          f"{res['after_run_max']} < {res['before_run_max']}")
    return res


if __name__ == "__main__":
    print("=" * 60)
    print("R1-R5 (unit, f7_lifecycle)")
    print("=" * 60)
    test_r1_single_onset()
    test_r2_persistent_condition()
    test_r3_new_independent_sweep()
    test_r4_invalidated_expired()
    test_r5_no_sweep()
    print("\n" + "=" * 60)
    print("R6 (existing suite / no side-effect)")
    print("=" * 60)
    test_r6_existing_suite()
    print("\n" + "=" * 60)
    hist = historical_validation()
    print("\n" + "=" * 60)
    print(f"RESULT: {len(PASS)} PASS, {len(FAIL)} FAIL")
    if FAIL:
        print("FAILED:", FAIL)
        sys.exit(1)
    print("ALL REGRESSION TESTS PASS")