565 lines
23 KiB
Python
565 lines
23 KiB
Python
# -*- 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()
|