Aliwkarimi98/SniperGold_ML
Watch
1
0
Fork
You've already forked SniperGold_ML
 1
forked from chiki2bum2/SniperGold_ML
Code Pull requests Activity
SniperGold_ML/ml/p3/smc_semantic/spec_tests_event_contract.py

565 lines
23 KiB
Python
RawPermalinkNormal ViewHistory

test: define F1 event contract regression coverage
2026-08-22 22:01:26 +07:00
# -*- coding: utf-8 -*-
"""P3-S.11 F1 SPEC TESTS — EVENT CONTRACT (sweep + CHoCH validity / supersession).
Truth : docs/SNIPERGOLD_CANONICAL_SETUP_CONTRACT_v1.md (frozen, P3-S.10)
+ docs/P3_S10_OWNER_ADJUDICATION.md OD-3/OD-4
+ SMC_LIQUIDITY_SWEEP_SPEC_v1.md R-H (valid b..b+40, expired b+41)
+ SMC_CHOCH_MSS_SPEC_v1.md S-9/S-10/S-12 (chain validity).
Code under : EVENT CONSUMERS of the ML-path EA AlgoForge_Backtest_Baseline.mq5
audit (f9 choch_confirms, f18 confluence) — before (legacy stale state)
vs after (canonical event validity) — plus the DETECTORS
(DetectLiquidityGrabs / ProcessStructure internal) which must
remain UNCHANGED (F1-T10 regression).
Canonical EVENT (frozen):
EVENT { ts, dir, valid_until, superseded, source }
- W_sweep = 40, W_choch = 40 (bars, M15 chain frame).
- active iff onset <= r <= onset + W (r = decision bar; valid at age W,
EXPIRED at age W+1).
- superseded : a newer onset of the same type replaced it (permanent; the
older event never reactivates).
- chain : sweep onset <= CHoCH onset <= sweep onset + W_sweep;
both fresh at r; directions equal.
Disiplin P3-S.11:
- F1-T01..T09 : ASSERT the canonical contract (oracle = contract).
- legacy vs repaired : REPORT (differential before/after observation).
- F1-T10 : ASSERT existing P3-S.2/P3-S.3 detector suites still PASS
(detection semantics unchanged).
- no AUC/PF/backtest/human annotation/ML in this file.
Usage: python spec_tests_event_contract.py
Output: output/p3_s11_event_contract_report.json
"""
import json
import os
import subprocess
import sys
import numpy as np
HERE = os.path.dirname(os.path.abspath(__file__))
# ---- canonical constants (frozen contract P3-S.10 OD-3) ----
W_SWEEP = 40 # Liquidity EVENT validity (M15 bars)
W_CHOCH = 40 # CHoCH/MSS EVENT validity (M15 bars)
GRAB_WINDOW = 8 # sweep detection window (UNCHANGED, P3-S.2 R-C)
INTERNAL_LEN = 5 # fractal confirmation (UNCHANGED, P3-S.3 S-2)
# =====================================================================
# CANONICAL EVENT CONTRACT (reference implementation — the contract)
# =====================================================================
class AFEvent:
"""Canonical EVENT {ts=bar, dir, valid_until, superseded, source}."""
__slots__ = ("bar", "dir", "valid_until", "superseded", "source")
def __init__(self, bar, direction, valid_until, superseded, source):
self.bar = int(bar)
self.dir = int(direction)
self.valid_until = int(valid_until)
self.superseded = bool(superseded)
self.source = source
def active_at(self, r):
"""Contract: active iff onset <= r <= valid_until AND not superseded."""
return (not self.superseded) and (self.bar <= r <= self.valid_until)
def as_dict(self):
return {"ts": self.bar, "dir": self.dir,
"valid_until": self.valid_until,
"superseded": self.superseded, "source": self.source}
def _supersede(events):
"""A newer onset of the same type permanently supersedes an older event."""
for e in events:
if any(x.bar > e.bar for x in events):
e.superseded = True
return events
def detect_sweep_events(h, l, c, refs, decision_bar,
grab_window=GRAB_WINDOW, w=W_SWEEP):
"""Sweep DETECTION (UNCHANGED, mirror DetectLiquidityGrabs / P3-S.2 R-C)
+ canonical EVENT construction. refs: [(p, lvl, is_high)]. Onset order asc."""
events = []
for (p, lvl, ih) in refs:
last = min(decision_bar, p + grab_window)
for b in range(p + 1, last + 1):
if ih and h[b] > lvl and c[b] < lvl: # buy-side swept -> bearish -1
events.append(AFEvent(b, -1, b + w, False, "sweep"))
break
if (not ih) and l[b] < lvl and c[b] > lvl: # sell-side swept -> bullish +1
events.append(AFEvent(b, 1, b + w, False, "sweep"))
break
return _supersede(events)
def detect_choch_events(o, h, l, c, refs, swing_breaks, decision_bar, w=W_CHOCH):
"""CHoCH DETECTION (UNCHANGED, mirror P3-S.3 spec_oracle S-3..S-10)
+ canonical EVENT construction. refs: [(p, level, is_high)].
swing_breaks: [(bar, trend)]. Returns list[AFEvent] onset asc."""
n = decision_bar + 1
events = []
up_target, up_bar = float("inf"), -1
dn_target, dn_bar = -float("inf"), -1
trend = 0
confirm = {}
for (p, lvl, ih) in refs:
confirm.setdefault(p + INTERNAL_LEN, []).append((p, lvl, ih))
for i in range(n):
for (p, lvl, ih) in confirm.get(i, []):
if ih:
up_target, up_bar = lvl, p
else:
dn_target, dn_bar = lvl, p
if up_bar >= 0 and c[i] > up_target:
choch = (trend < 0)
allow = swing_at_of(swing_breaks, i) >= 0
if allow and choch:
events.append(AFEvent(i, 1, i + w, False, "choch"))
trend = 1
up_target, up_bar = float("inf"), -1
if dn_bar >= 0 and c[i] < dn_target:
choch = (trend > 0)
allow = swing_at_of(swing_breaks, i) <= 0
if allow and choch:
events.append(AFEvent(i, -1, i + w, False, "choch"))
trend = -1
dn_target, dn_bar = -float("inf"), -1
return _supersede(events)
def swing_at_of(swing_breaks, i):
v = 0
for (b, val) in swing_breaks:
if b <= i:
v = val
else:
break
return v
def latest_active(events, r):
"""Latest (newest onset) event <= r; active iff within its validity window.
Supersession is permanent: an older event never reactivates (matches the
single-latest-state of the EA)."""
cand = [e for e in events if e.bar <= r]
if not cand:
return None
e = max(cand, key=lambda x: x.bar)
return e if e.active_at(r) else None
def chain_valid(sweep_events, choch_events, r):
"""Canonical setup chain (frozen §D/§M + OD-4):
latest active sweep AND latest active CHoCH, sweep onset <= choch onset
<= sweep onset + W_sweep, directions equal."""
swp = latest_active(sweep_events, r)
chc = latest_active(choch_events, r)
if swp is None or chc is None:
return False
if chc.bar < swp.bar:
return False
if (chc.bar - swp.bar) > W_SWEEP:
return False
if chc.dir != swp.dir:
return False
return True
def f7_contract(sweep_events, r):
"""f7 sweep_dir under the contract: latest active sweep -> dir else 0."""
swp = latest_active(sweep_events, r)
return swp.dir if swp is not None else 0
def f9_contract(sweep_events, choch_events, r):
"""f9 choch_confirms under the contract: 1 iff the canonical chain holds."""
return 1 if chain_valid(sweep_events, choch_events, r) else 0
def f18_event_legs_contract(sweep_events, choch_events, r, bias):
"""f18 confluence — the three event-sensitive +15 legs under the contract.
(HTF/premium/EQ legs are untouched by F1 and not re-tested here.)"""
cb = 0
swp = latest_active(sweep_events, r)
swp_valid = swp is not None
chain = chain_valid(sweep_events, choch_events, r)
if swp_valid and swp.dir != 0:
cb += 15
if chain:
cb += 15
if swp_valid and ((swp.dir == 1 and bias > 0) or (swp.dir == -1 and bias < 0)):
cb += 15
return cb
# =====================================================================
# LEGACY PORT (BEFORE) — old EA consumers: raw persistent state, NO validity
# =====================================================================
def legacy_f9(sweep_state, choch_state):
"""OLD EA chochOK (BUG-P3S3-001): cd!=0 && cb>=sb && cd==sd — no expiry."""
sb, sd = sweep_state
cb, cd = choch_state
return 1 if (cd != 0 and cb >= sb and cd == sd) else 0
def legacy_f18_event_legs(sweep_state, choch_state, bias):
"""OLD EA f18 event legs (BUG-P3S2-001/BUG-P3S3-001): raw state, no expiry."""
sb, sd = sweep_state
cb, cd = choch_state
cbx = 0
if sd != 0:
cbx += 15
if cd != 0 and cd == sd:
cbx += 15
if (sd == 1 and bias > 0) or (sd == -1 and bias < 0):
cbx += 15
return cbx
# =====================================================================
# HELPERS (synthetic case materialization — same shape as P3-S.2/S.3)
# =====================================================================
def materialize(case):
bs = case["bar_spec"]
bars = [list(bs["neutral"]) for _ in range(bs["count"])]
for (idx, ohlc) in bs["patches"]:
if idx < len(bars):
bars[idx] = list(ohlc)
arr = np.asarray(bars, dtype=float)
return arr[:, 0], arr[:, 1], arr[:, 2], arr[:, 3]
def sweep_state_of(events):
"""Latest (newest onset) sweep event state (bar, dir), (-1,0) if none."""
if not events:
return (-1, 0)
e = max(events, key=lambda x: x.bar)
return (e.bar, e.dir)
def choch_state_of(events):
if not events:
return (-1, 0)
e = max(events, key=lambda x: x.bar)
return (e.bar, e.dir)
def mutate_future(o, h, l, c, mutations, neutral):
o = o.copy(); h = h.copy(); l = l.copy(); c = c.copy()
for (idx, ohlc) in mutations:
while idx >= len(o):
o = np.append(o, neutral[0])
h = np.append(h, neutral[1])
l = np.append(l, neutral[2])
c = np.append(c, neutral[3])
o[idx], h[idx], l[idx], c[idx] = ohlc
return o, h, l, c
# =====================================================================
# TEST CASES — F1-T01..T09 (canonical contract assertions)
# =====================================================================
NEUTRAL = [100.0, 101.5, 99.0, 100.5] # o,h,l,c (small body)
# CHoCH scaffolding: bearish internal break at bar 10 (LOW ref 101.0 confirmed
# at 9) sets prior internal trend = -1, then a bullish break above HIGH ref
# 102.0 produces a bullish CHoCH. Swing timeline flat (permissive gate).
CHOCH_REFS = [(4, 101.0, 0), (8, 102.0, 1)]
SWING_BREAKS = [(5, 0)]
CHOCH_BREAK_BARS = {30: [100.0, 104.0, 100.0, 103.0],
14: [100.0, 104.0, 100.0, 103.0],
55: [100.0, 104.0, 100.0, 103.0]}
def case_single_sweep():
"""F1-T01 single sweep event + F1-T02 expiration boundary."""
patches = {12: [100.0, 101.0, 97.0, 99.0]} # low<98, close>98 -> +1 at 12
return {
"id": "F1-T01",
"title": "Single sweep event + boundary (valid at +40, expired at +41)",
"bar_spec": {"neutral": NEUTRAL, "count": 120, "patches": sorted(patches.items())},
"given_references": [(10, 98.0, 0)],
"assert": [
("events_count", lambda evs, cevs, r: len(evs) == 1),
("onset_bar", lambda evs, cevs, r: evs[0].bar == 12),
("onset_dir", lambda evs, cevs, r: evs[0].dir == 1),
("valid_until", lambda evs, cevs, r: evs[0].valid_until == 12 + W_SWEEP),
("superseded_false", lambda evs, cevs, r: not evs[0].superseded),
("active_at_40", lambda evs, cevs, r: f7_contract(evs, 12 + W_SWEEP) == 1),
("expired_at_41", lambda evs, cevs, r: f7_contract(evs, 12 + W_SWEEP + 1) == 0),
("source_sweep", lambda evs, cevs, r: evs[0].source == "sweep"),
],
}
def case_consumer_expiration():
"""F1-T03 sweep consumer expiration: age 41+ -> NO VALID SWEEP EVENT."""
patches = {10: [100.0, 101.0, 100.0, 100.4], # bearish internal break (trend -> -1)
12: [100.0, 101.0, 97.0, 99.0], # sweep +1 at 12
14: CHOCH_BREAK_BARS[14]} # bullish CHoCH +1 at 14
return {
"id": "F1-T03",
"title": "Sweep consumer expiration (age 43 -> NO VALID SWEEP; legacy f9 still 1)",
"bar_spec": {"neutral": NEUTRAL, "count": 120, "patches": sorted(patches.items())},
"given_references": [(10, 98.0, 0)],
"given_choch_references": CHOCH_REFS,
"given_swing_breaks": SWING_BREAKS,
"assert": [
("no_active_sweep_at_55", lambda evs, cevs, r: latest_active(evs, 55) is None),
("f7_zero_at_55", lambda evs, cevs, r: f7_contract(evs, 55) == 0),
("f9_zero_at_55", lambda evs, cevs, r: f9_contract(evs, cevs, 55) == 0),
],
}
def case_supersession():
"""F1-T04 new sweep supersedes old (deterministic supersession)."""
patches = {12: [100.0, 101.0, 97.0, 99.0], # sweep A +1 at 12 (LOW ref 98)
20: [100.0, 102.0, 98.0, 100.0], # HIGH ref bar 102
22: [100.0, 103.0, 101.0, 101.5]} # sweep B -1 at 22 (high>102, close<102)
return {
"id": "F1-T04",
"title": "New sweep supersedes old (deterministic supersession)",
"bar_spec": {"neutral": NEUTRAL, "count": 120, "patches": sorted(patches.items())},
"given_references": [(10, 98.0, 0), (20, 102.0, 1)],
"assert": [
("two_events", lambda evs, cevs, r: len(evs) == 2),
("old_superseded", lambda evs, cevs, r: evs[0].superseded),
("new_active", lambda evs, cevs, r: not evs[1].superseded),
("latest_is_new", lambda evs, cevs, r: latest_active(evs, 30).bar == 22),
("f7_uses_new", lambda evs, cevs, r: f7_contract(evs, 30) == -1),
("no_fallback_after_expiry", lambda evs, cevs, r: f7_contract(evs, 70) == 0),
],
}
def case_choch_inside_chain():
"""F1-T05 CHoCH inside valid sweep window -> VALID (f9=1)."""
patches = {10: [100.0, 101.0, 100.0, 100.4], # bearish internal break at 10
12: [100.0, 101.0, 97.0, 99.0], # sweep +1 at 12
30: CHOCH_BREAK_BARS[30]} # bullish CHoCH +1 at 30
return {
"id": "F1-T05",
"title": "CHoCH inside valid sweep window -> chain VALID (f9=1)",
"bar_spec": {"neutral": NEUTRAL, "count": 120, "patches": sorted(patches.items())},
"given_references": [(10, 98.0, 0)],
"given_choch_references": CHOCH_REFS,
"given_swing_breaks": SWING_BREAKS,
"assert": [
("choch_onset_30", lambda evs, cevs, r: latest_active(cevs, 40).bar == 30),
("chain_valid_at_40", lambda evs, cevs, r: f9_contract(evs, cevs, 40) == 1),
("dir_match", lambda evs, cevs, r: latest_active(evs, 40).dir ==
latest_active(cevs, 40).dir),
],
}
def case_choch_outside_chain():
"""F1-T06 CHoCH outside the valid sweep chain -> INVALID (f9=0; legacy 1)."""
patches = {10: [100.0, 101.0, 100.0, 100.4],
12: [100.0, 101.0, 97.0, 99.0], # sweep +1 at 12
55: CHOCH_BREAK_BARS[55]} # bullish CHoCH +1 at 55 (43 bars after)
return {
"id": "F1-T06",
"title": "CHoCH outside sweep validity chain -> INVALID for canonical chain",
"bar_spec": {"neutral": NEUTRAL, "count": 120, "patches": sorted(patches.items())},
"given_references": [(10, 98.0, 0)],
"given_choch_references": CHOCH_REFS,
"given_swing_breaks": SWING_BREAKS,
"assert": [
("sweep_expired_at_55", lambda evs, cevs, r: f7_contract(evs, 55) == 0),
("chain_invalid", lambda evs, cevs, r: f9_contract(evs, cevs, 55) == 0),
("chain_bound_violated", lambda evs, cevs, r: (55 - 12) > W_SWEEP),
],
}
def case_choch_expiry():
"""F1-T07 CHoCH consumer expiration: stale CHoCH must not confirm."""
patches = {10: [100.0, 101.0, 100.0, 100.4],
12: [100.0, 101.0, 97.0, 99.0], # sweep +1 at 12
30: CHOCH_BREAK_BARS[30]} # bullish CHoCH +1 at 30
return {
"id": "F1-T07",
"title": "CHoCH consumer expiration (age > W_choch -> f9=0)",
"bar_spec": {"neutral": NEUTRAL, "count": 120, "patches": sorted(patches.items())},
"given_references": [(10, 98.0, 0)],
"given_choch_references": CHOCH_REFS,
"given_swing_breaks": SWING_BREAKS,
"assert": [
("choch_expired_at_75", lambda evs, cevs, r: latest_active(cevs, 75) is None),
("f9_zero_at_75", lambda evs, cevs, r: f9_contract(evs, cevs, 75) == 0),
],
}
def case_no_lookahead():
"""F1-T08 future-bar mutation must not change past event state."""
patches = {12: [100.0, 101.0, 97.0, 99.0]} # sweep +1 at 12
return {
"id": "F1-T08",
"title": "No look-ahead: future mutation does not alter past event state",
"bar_spec": {"neutral": NEUTRAL, "count": 120, "patches": sorted(patches.items())},
"given_references": [(10, 98.0, 0)],
"decision_bar": 50,
"mutations": [(60, [100.0, 110.0, 99.0, 109.0])], # future bearish sweep bar
"assert": [
("state_unchanged_at_50", lambda evs, cevs, r: f7_contract(evs, r) == 1),
("no_lookahead_invariant", lambda evs, cevs, r:
len(evs) == 1 and evs[0].bar == 12),
],
}
def case_persistent_condition():
"""F1-T09 persistent condition -> ONE event, not repeated emissions."""
patches = {10: [100.0, 100.0, 98.0, 100.0]}
for b in range(12, 52): # condition persists 40 bars
patches[b] = [100.0, 101.0, 97.0, 99.0]
return {
"id": "F1-T09",
"title": "Persistent condition -> ONE event (no duplicate emission)",
"bar_spec": {"neutral": NEUTRAL, "count": 120, "patches": sorted(patches.items())},
"given_references": [(10, 98.0, 0)],
"assert": [
("one_event", lambda evs, cevs, r: len(evs) == 1),
("active_window_bounded", lambda evs, cevs, r:
f7_contract(evs, 51) == 1 and f7_contract(evs, 53) == 0),
],
}
# =====================================================================
# RUNNER
# =====================================================================
def run_case(case):
o, h, l, c = materialize(case)
refs = [tuple(r) for r in case["given_references"]]
db = case.get("decision_bar", case["bar_spec"]["count"] - 1)
sweep_events = detect_sweep_events(h, l, c, refs, db)
choch_events = []
if "given_choch_references" in case:
crefs = [tuple(r) for r in case["given_choch_references"]]
sbreaks = [tuple(s) for s in case.get("given_swing_breaks", [])]
choch_events = detect_choch_events(o, h, l, c, crefs, sbreaks, db)
# F1-T08: future mutation invariance (mutations at bars > db must not change
# the event state known at db)
if "mutations" in case:
o3, h3, l3, c3 = mutate_future(o, h, l, c, case["mutations"], NEUTRAL)
evs3 = detect_sweep_events(h3, l3, c3, refs, db)
else:
evs3 = None
checks = {}
for (name, fn) in case["assert"]:
try:
checks[name] = bool(fn(sweep_events, choch_events, db))
except Exception as ex:
checks[name] = False
checks[name + "_error"] = str(ex)
if evs3 is not None:
checks["no_lookahead_future_mutation"] = bool(
len(evs3) == len(sweep_events) and
all(a.bar == b.bar and a.dir == b.dir for a, b in zip(evs3, sweep_events)))
# legacy vs contract differential (BEFORE/AFTER) for f9/f18
swp_state = sweep_state_of(sweep_events)
chc_state = choch_state_of(choch_events)
legacy_f9_v = legacy_f9(swp_state, chc_state)
contract_f9_v = f9_contract(sweep_events, choch_events, db)
legacy_f18_v = legacy_f18_event_legs(swp_state, chc_state, 1)
contract_f18_v = f18_event_legs_contract(sweep_events, choch_events, db, 1)
return {
"id": case["id"],
"title": case["title"],
"sweep_events": [e.as_dict() for e in sweep_events],
"choch_events": [e.as_dict() for e in choch_events],
"latest_active_sweep": latest_active(sweep_events, db).as_dict()
if latest_active(sweep_events, db) else None,
"f7_contract": f7_contract(sweep_events, db),
"f9_contract": contract_f9_v,
"f9_legacy": legacy_f9_v,
"f18_event_legs_contract": contract_f18_v,
"f18_event_legs_legacy": legacy_f18_v,
"checks": checks,
"pass": all(v for k, v in checks.items() if not k.endswith("_error")),
}
def regression_p3s2_s3():
"""F1-T10: existing detector synthetic suites remain unchanged and PASS."""
ok = True
detail = {}
for (name, script) in (("P3-S.2 liquidity sweep", "spec_tests_liquidity_sweep.py"),
("P3-S.3 choch/mss", "spec_tests_choch_mss.py")):
r = subprocess.run([sys.executable, os.path.join(HERE, script)],
capture_output=True, text=True)
ok = ok and (r.returncode == 0)
detail[name] = {"returncode": r.returncode, "pass": r.returncode == 0}
return ok, detail
def main():
cases = [case_single_sweep(), case_consumer_expiration(), case_supersession(),
case_choch_inside_chain(), case_choch_outside_chain(),
case_choch_expiry(), case_no_lookahead(), case_persistent_condition()]
results = [run_case(c) for c in cases]
n_pass = sum(1 for r in results if r["pass"])
n_total = len(results)
print(f"=== P3-S.11 F1 SPEC TESTS — EVENT CONTRACT ({n_pass}/{n_total} PASS) ===")
print(f"{'ID':<8} {'PASS':<6} {'f7':<4} {'f9 c/l':<8} notes")
for r in results:
print(f"{r['id']:<8} {str(r['pass']):<6} {r['f7_contract']:<4} "
f"{str(r['f9_contract'])+'/'+str(r['f9_legacy']):<8} {r['title'][:48]}")
for r in results:
if not r["pass"]:
print(" FAILED:", r["id"], r["checks"])
# F1-T10 regression
reg_ok, reg_detail = regression_p3s2_s3()
if n_pass != n_total or not reg_ok:
print("\nF1 TESTS FAILED")
if not reg_ok:
print(" F1-T10 regression failed:", reg_detail)
sys.exit(1)
print("\nAll F1 contract assertions PASS; P3-S.2/P3-S.3 detection suites PASS "
"(detection semantics unchanged).")
stale_legacy = sum(1 for r in results if r["f9_legacy"] == 1 and r["f9_contract"] == 0)
total_f9_on = sum(1 for r in results if r["f9_legacy"] == 1)
report = {
"doc": "docs/P3_S11_EVENT_CONTRACT_REPAIR.md",
"phase": "P3-S.11",
"generated_utc": __import__("datetime").datetime.now(
__import__("datetime").timezone.utc).isoformat(),
"constants": {"W_SWEEP": W_SWEEP, "W_CHOCH": W_CHOCH,
"GRAB_WINDOW": GRAB_WINDOW, "INTERNAL_LEN": INTERNAL_LEN},
"summary": {"total": n_total, "passed": n_pass,
"failed": n_total - n_pass,
"regression": reg_detail,
"before_after": {
"legacy_f9_active_cases": total_f9_on,
"stale_usage_eliminated_cases": stale_legacy,
}},
"cases": results,
}
outdir = os.path.join(HERE, "output")
os.makedirs(outdir, exist_ok=True)
out_path = os.path.join(outdir, "p3_s11_event_contract_report.json")
with open(out_path, "w", encoding="utf-8") as f:
json.dump(report, f, indent=2, default=str)
print(f"[saved] {out_path}")
if __name__ == "__main__":
main()
Copy permalink