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