238 lines
8.9 KiB
Python
238 lines
8.9 KiB
Python
# -*- 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")
|