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_zone_contract.py

766 lines
36 KiB
Python
RawPermalinkNormal ViewHistory

test: define F2 zone contract regression coverage
2026-08-22 22:32:56 +07:00
# -*- coding: utf-8 -*-
"""P3-S.12 SPEC TESTS — F2 ZONE CONTRACT (FVG + Order Block).
Truth : docs/SNIPERGOLD_CANONICAL_SETUP_CONTRACT_v1.md (frozen §K/§M/§O)
+ docs/SMC_FVG_SPEC_v1.md (S-1..S-13)
+ docs/SMC_ORDER_BLOCK_SPEC_v1.md (S-1..S-15)
-> spec_oracle_*() in this file (canonical ZONE contract).
Code under : Engine 2 AF_FindFVG / AF_FindOrderBlock (AF_Engine2_Agents.mqh)
audit and display AF_CollectFVG / AF_CollectOBs
(AF_Engine2_Display.mqh) -> repaired_port_*() (the F2 target).
Legacy port : legacy_port_*() = the PRE-F2 Engine-2 behavior (no mitigation,
FVG 1-bar lag) — kept as historical before/after evidence.
Discipline P3-S.12 (F2):
- spec oracle vs expected : ASSERT (canonical zone contract = truth)
- repaired port vs spec : ASSERT (the MQL5 fix must satisfy the contract)
- legacy port vs repaired : REPORT (before/after differential evidence)
- invariance (transform/future-mutation/symmetry) : ASSERT
- no AUC/PF/backtest/optimization/ML/human annotation in this file.
Cases: F2-T01..T20 (all 20 required by the brief §10).
Zone contract (frozen §K refined by the F2 brief §5/§11):
ZONE { ts(formation bar), dir, upper, lower, mit_state, invalidated }
mit_state : UNMITIGATED(0) | PARTIALLY_FILLED(1) | FULLY_MITIGATED(2)
PARTIALLY_FILLED is ACTIVE (S-9: partial fill != mitigation).
invalidated: terminal flag; zone-level = set when FULLY_MITIGATED
(frozen §M "invalidation = mitigation is the terminal
usable-state"); setup-level (F3) may set it independently.
consumer : available iff mit_state in {UNMITIGATED, PARTIALLY_FILLED}
and NOT invalidated. A zone is a persistent entity: one
formation = one zone, never re-emitted per bar (S-8).
FVG mit : wick full-fill — bull Low(j)<=bot / bear High(j)>=top (S-9).
OB mit : close-through full fill — bull Close(j)<bot / bear Close(j)>top
(S-9); scan includes the move candle (j>B), display-consistent.
no age expiry (W_zone_age = OPEN NUMERIC PARAMETER, default none).
Usage: python spec_tests_zone_contract.py
Output: output/p3_s12_zone_contract_report.json
"""
import json
import os
import sys
import numpy as np
HERE = os.path.dirname(os.path.abspath(__file__))
# ---- zone contract constants (SPEC local) ----
FVG_LOOKBACK = 40 # Engine-2 FVG consumer lookback (C agent; AF_E2_FVG_LOOKBACK)
AVG_N = 20 # OB avg-body window (AF_E2_LOOKBACK_AVG)
MIN_BARS_OB = 22 # OB detector requires avgN+2 closed bars
MOVE_MULT = 1.5 # OB strong-move body multiplier (AF_E3_MOVE_BODY)
EPS = 1e-9
# mitigation states (canonical)
UNMITIGATED = 0
PARTIALLY_FILLED = 1
FULLY_MITIGATED = 2
STATE_NAMES = {0: "UNMITIGATED", 1: "PARTIALLY_FILLED", 2: "FULLY_MITIGATED"}
# =====================================================================
# SPEC ORACLES (truth) — canonical ZONE contract
# =====================================================================
def _overlap(h_j, l_j, bot, top):
"""A bar's range [l_j,h_j] overlaps the zone [bot,top]."""
return h_j >= bot - EPS and l_j <= top + EPS
def spec_oracle_fvg(o, h, l, c, db):
"""Canonical FVG zone evaluator (S-1..S-13 + frozen §K).
Returns list of zone dicts: {b, d, top, bot, mit, partial, invalidated}
for every qualifying C3 bar b in [2, db]; state computed from bars (b, db].
"""
zones = []
for b in range(2, db + 1):
if l[b] > h[b - 2]:
zones.append(dict(b=b, d=1, top=float(l[b]), bot=float(h[b - 2])))
elif h[b] < l[b - 2]:
zones.append(dict(b=b, d=-1, top=float(l[b - 2]), bot=float(h[b])))
for z in zones:
full = False
partial = False
for j in range(z["b"] + 1, db + 1):
if z["d"] > 0 and l[j] <= z["bot"]: # wick full-fill (S-9)
full = True
if z["d"] < 0 and h[j] >= z["top"]: # wick full-fill (S-9)
full = True
if _overlap(h[j], l[j], z["bot"], z["top"]):
partial = True
z["mit"] = FULLY_MITIGATED if full else (PARTIALLY_FILLED if partial
else UNMITIGATED)
z["partial"] = partial
z["invalidated"] = full # zone-level terminal (frozen §M)
return zones
def avg_body(o, c, db, avg_n=AVG_N):
m = min(avg_n, db + 1)
if m <= 0:
return 0.0
s = 0.0
for i in range(db - m + 1, db + 1):
s += abs(c[i] - o[i])
return s / m
def spec_oracle_ob(o, h, l, c, db, avg_n=AVG_N, min_bars=MIN_BARS_OB,
move_mult=MOVE_MULT):
"""Canonical OB zone evaluator (S-1..S-15 + frozen §O/§K).
Zone for every qualifying pair (B=b, M=b+1) with b+1 <= db; state computed
from bars (b, db]. Mitigation scan includes M (display-consistent);
partial-fill scan is strictly AFTER M (j >= b+2) — the move candle is part
of formation, not a re-entry.
"""
n = db + 1
if n < min_bars:
return []
avg = avg_body(o, c, db, avg_n)
if avg <= 0.0:
return []
zones = []
for b in range(0, n - 1):
mb = abs(c[b + 1] - o[b + 1])
if mb < move_mult * avg:
continue
up = c[b + 1] > o[b + 1]
d = 0
if up and c[b] < o[b]:
d = 1
elif (not up) and c[b] > o[b]:
d = -1
if d == 0:
continue
zones.append(dict(b=b, d=d, top=float(h[b]), bot=float(l[b])))
for z in zones:
full = False
partial = False
for j in range(z["b"] + 1, db + 1): # includes M (display-consistent)
if z["d"] > 0 and c[j] < z["bot"]: # close-through (S-9)
full = True
if z["d"] < 0 and c[j] > z["top"]:
full = True
for j in range(z["b"] + 2, db + 1): # strictly after M (re-entry)
if _overlap(h[j], l[j], z["bot"], z["top"]):
partial = True
z["mit"] = FULLY_MITIGATED if full else (PARTIALLY_FILLED if partial
else UNMITIGATED)
z["partial"] = partial
z["invalidated"] = full
return zones
def newest_active(zones):
"""Canonical consumer query: newest zone that is ACTIVE (unmitigated and
not invalidated). Returns [b, d, top, bot] or None."""
act = [z for z in zones
if z["mit"] in (UNMITIGATED, PARTIALLY_FILLED) and not z["invalidated"]]
if not act:
return None
z = max(act, key=lambda z: z["b"])
return [z["b"], z["d"], float(z["top"]), float(z["bot"])]
# =====================================================================
# LEGACY PORTS (PRE-F2 Engine-2 behavior — historical before-evidence)
# =====================================================================
def legacy_fvg_port(o, h, l, c, db, lookback=FVG_LOOKBACK):
"""PRE-F2 AF_FindFVG: loop starts at i=1 (1-bar lag), NO mitigation filter,
returns the FIRST (newest) geometric FVG."""
cnt = db + 1
if cnt < 3:
return None
max_i = min(cnt - 3, lookback if lookback > 0 else cnt)
for i in range(1, max_i + 1):
c3 = db - i
if l[c3] > h[c3 - 2]:
return [1, float(l[c3]), float(h[c3 - 2])]
if h[c3] < l[c3 - 2]:
return [-1, float(l[c3 - 2]), float(h[c3])]
return None
def legacy_ob_port(o, h, l, c, db, avg_n=AVG_N, min_bars=MIN_BARS_OB,
move_mult=MOVE_MULT):
"""PRE-F2 AF_FindOrderBlock: newest pair first, NO mitigation filter."""
cnt = db + 1
if cnt < min_bars:
return None
avg = avg_body(o, c, db, avg_n)
if avg <= 0.0:
return None
for i in range(1, cnt - 1):
b = db - i
move = b + 1
if abs(c[move] - o[move]) < move_mult * avg:
continue
if c[move] > o[move] and c[b] < o[b]:
return [1, float(h[b]), float(l[b])]
if c[move] < o[move] and c[b] > o[b]:
return [-1, float(h[b]), float(l[b])]
return None
# =====================================================================
# REPAIRED PORTS (POST-F2 Engine-2 target — the MQL5 fix must match)
# =====================================================================
def repaired_fvg_port(o, h, l, c, db, lookback=FVG_LOOKBACK):
"""POST-F2 AF_FindFVG: i starts at 0 (newest closed bar eligible as C3,
S-6 — no 1-bar lag, no future candle) and skips zones already wick
full-filled (S-9); returns the FIRST (newest) UNMITIGATED zone."""
cnt = db + 1
if cnt < 3:
return None
max_i = min(cnt - 3, lookback if lookback > 0 else cnt)
for i in range(0, max_i + 1):
c3 = db - i
if l[c3] > h[c3 - 2]:
bot, top = float(h[c3 - 2]), float(l[c3])
if not any(l[j] <= bot for j in range(c3 + 1, cnt)):
return [1, top, bot]
if h[c3] < l[c3 - 2]:
bot, top = float(h[c3]), float(l[c3 - 2])
if not any(h[j] >= top for j in range(c3 + 1, cnt)):
return [-1, top, bot]
return None
def repaired_ob_port(o, h, l, c, db, avg_n=AVG_N, min_bars=MIN_BARS_OB,
move_mult=MOVE_MULT):
"""POST-F2 AF_FindOrderBlock: newest pair first, skips zones already
close-through full-filled (S-9)."""
cnt = db + 1
if cnt < min_bars:
return None
avg = avg_body(o, c, db, avg_n)
if avg <= 0.0:
return None
for i in range(1, cnt - 1):
b = db - i
move = b + 1
if abs(c[move] - o[move]) < move_mult * avg:
continue
d = 0
if c[move] > o[move] and c[b] < o[b]:
d = 1
elif c[move] < o[move] and c[b] > o[b]:
d = -1
if d == 0:
continue
top, bot = float(h[b]), float(l[b])
if d > 0 and any(c[j] < bot for j in range(b + 1, cnt)):
continue
if d < 0 and any(c[j] > top for j in range(b + 1, cnt)):
continue
return [d, top, bot]
return None
# =====================================================================
# HELPERS
# =====================================================================
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 mirror(o, h, l, c, pivot):
return (2.0 * pivot - o, 2.0 * pivot - l, 2.0 * pivot - h,
2.0 * pivot - c)
def apply_transform(o, h, l, c, transform):
if transform is None:
return o, h, l, c
kind = transform["kind"]
if kind == "scale":
k = transform["factor"]
return o * k, h * k, l * k, c * k
if kind == "translate":
v = transform["value"]
return o + v, h + v, l + v, c + v
raise ValueError("unknown transform " + kind)
def apply_mutations(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
def zones_sig(zones):
"""[(b, d, top, bot, mit, invalidated)] — comparison key."""
return [(z["b"], z["d"], z["top"], z["bot"], z["mit"], z["invalidated"])
for z in zones]
def vec_equal(a, b):
if a is None or b is None:
return a is None and b is None
return (a[0] == b[0] and a[1] == b[1] and abs(a[2] - b[2]) <= EPS
and abs(a[3] - b[3]) <= EPS)
def port_equal(a, b):
if a is None or b is None:
return a is None and b is None
return a[0] == b[0] and abs(a[1] - b[1]) <= EPS and abs(a[2] - b[2]) <= EPS
# =====================================================================
# F2 CASE TABLE (F2-T01..T20) — expected values derive from the canonical
# zone contract, NOT from the buggy implementation.
# mit encoding: 0=UNMITIGATED, 1=PARTIALLY_FILLED, 2=FULLY_MITIGATED
# =====================================================================
NEUT_HI = [100.6, 100.9, 100.51, 100.8] # above-zone neutral (no overlap)
NEUT_LO = [100.0, 100.3, 99.9, 100.1] # near-zone neutral
NEUT_FVG = [100.0, 100.5, 99.8, 100.3] # FVG base neutral
CASES = [
# ---- FVG ----
dict(id="F2-T01", kind="fvg", title="FVG formation (bullish, S-1/S-4)",
bar_spec=dict(neutral=NEUT_FVG, count=6,
patches=[[2, [100.0, 100.2, 99.9, 100.1]],
[3, [100.3, 100.9, 100.25, 100.8]],
[4, [100.5, 101.0, 100.4, 100.9]],
[5, [100.6, 100.8, 100.45, 100.7]]]),
decision_bar=5,
expected=dict(zones=[[4, 1, 100.4, 100.2, 0, False, False]],
zone_count=1, newest_active=[4, 1, 100.4, 100.2],
legacy=[1, 100.4, 100.2], repaired=[1, 100.4, 100.2])),
dict(id="F2-T02", kind="fvg", title="FVG remains ONE zone after many bars (S-8)",
bar_spec=dict(neutral=[100.6, 100.9, 100.5, 100.8], count=25,
patches=[[0, [100.6, 100.9, 100.2, 100.8]],
[1, [100.6, 100.9, 100.2, 100.8]],
[2, [100.0, 100.2, 99.9, 100.1]],
[3, [100.3, 100.9, 100.25, 100.8]],
[4, [100.5, 101.0, 100.4, 100.9]]]),
decision_bar=24,
extra_dbs=[dict(db=14, zone_ids=[4], newest_active=[4, 1, 100.4, 100.2])],
expected=dict(zones=[[4, 1, 100.4, 100.2, 0, False, False]],
zone_count=1, newest_active=[4, 1, 100.4, 100.2],
legacy=[1, 100.4, 100.2], repaired=[1, 100.4, 100.2])),
dict(id="F2-T03", kind="fvg", title="FVG partial fill -> still ACTIVE (S-9)",
bar_spec=dict(neutral=NEUT_FVG, count=6,
patches=[[2, [100.0, 100.2, 99.9, 100.1]],
[3, [100.3, 100.9, 100.25, 100.8]],
[4, [100.5, 101.0, 100.4, 100.9]],
[5, [100.25, 100.45, 100.22, 100.3]]]),
decision_bar=5,
expected=dict(zones=[[4, 1, 100.4, 100.2, 1, True, False]],
zone_count=1, newest_active=[4, 1, 100.4, 100.2],
legacy=[1, 100.4, 100.2], repaired=[1, 100.4, 100.2])),
dict(id="F2-T04", kind="fvg", title="FVG full mitigation (wick full-fill, S-9)",
bar_spec=dict(neutral=NEUT_FVG, count=7,
patches=[[2, [100.0, 100.2, 99.9, 100.1]],
[3, [100.3, 100.9, 100.25, 100.8]],
[4, [100.5, 101.0, 100.4, 100.9]],
[5, [100.6, 100.8, 100.45, 100.7]],
[6, [100.1, 100.45, 100.1, 100.2]]]),
decision_bar=6,
expected=dict(zones=[[4, 1, 100.4, 100.2, 2, True, True]],
zone_count=1, newest_active=None,
legacy=[1, 100.4, 100.2], repaired=None,
note="FVG-T10 equivalent: legacy returns mitigated zone (BUG-P3S4-001); repaired None.")),
dict(id="F2-T05", kind="fvg",
title="FVG mitigated zone NOT consumed; older active zone returned",
bar_spec=dict(neutral=NEUT_FVG, count=12,
patches=[[2, [100.0, 100.2, 99.9, 100.1]],
[3, [100.3, 100.9, 100.25, 100.8]],
[4, [100.5, 101.0, 100.4, 100.9]],
[5, [100.6, 100.8, 100.45, 100.7]],
[6, [100.7, 100.9, 100.5, 100.8]],
[7, [100.7, 100.9, 100.6, 100.8]],
[8, [100.8, 101.2, 100.7, 101.1]],
[9, [101.0, 101.3, 100.95, 101.2]],
[10, [101.0, 101.3, 100.96, 101.2]],
[11, [100.5, 100.95, 100.45, 100.6]]]),
decision_bar=11,
expected=dict(zones=[[4, 1, 100.4, 100.2, 0, False, False],
[9, 1, 100.95, 100.9, 2, True, True]],
zone_count=2, newest_active=[4, 1, 100.4, 100.2],
legacy=[1, 100.95, 100.9], repaired=[1, 100.4, 100.2],
note="newest zone (b=9) fully mitigated -> legacy consumes it (BUG); repaired returns older active b=4.")),
dict(id="F2-T06", kind="fvg", title="FVG invalidation is terminal (S-10/§K)",
bar_spec=dict(neutral=NEUT_FVG, count=10,
patches=[[2, [100.0, 100.2, 99.9, 100.1]],
[3, [100.3, 100.9, 100.25, 100.8]],
[4, [100.5, 101.0, 100.4, 100.9]],
[5, [100.6, 100.8, 100.45, 100.7]],
[6, [100.1, 100.45, 100.1, 100.2]],
[7, [100.1, 100.6, 100.05, 100.2]],
[8, [100.1, 100.6, 100.05, 100.2]],
[9, [100.1, 100.6, 100.05, 100.2]]]),
decision_bar=9,
extra_dbs=[dict(db=6, zone_ids=[4], newest_active=None)],
expected=dict(zones=[[4, 1, 100.4, 100.2, 2, True, True]],
zone_count=1, newest_active=None,
legacy=[1, 100.4, 100.2], repaired=None,
note="price re-enters the zone (bars 7-9) but the zone stays invalidated (irreversible); no reactivation.")),
dict(id="F2-T07", kind="fvg",
title="FVG formation timing boundary: C3 = newest closed bar (S-6)",
bar_spec=dict(neutral=NEUT_FVG, count=5,
patches=[[2, [100.0, 100.2, 99.9, 100.1]],
[3, [100.3, 100.9, 100.25, 100.8]],
[4, [100.5, 101.0, 100.4, 100.9]]]),
decision_bar=4,
expected=dict(zones=[[4, 1, 100.4, 100.2, 0, False, False]],
zone_count=1, newest_active=[4, 1, 100.4, 100.2],
legacy=None, repaired=[1, 100.4, 100.2],
note="BUG-P3S4-002: legacy skips C3=index0 (1-bar lag); repaired detects at C3 close.")),
dict(id="F2-T08", kind="fvg",
title="FVG one-bar timing regression: no future candle, no needless delay",
bar_spec=dict(neutral=NEUT_FVG, count=5,
patches=[[2, [100.0, 100.2, 99.9, 100.1]],
[3, [100.3, 100.9, 100.25, 100.8]],
[4, [100.5, 101.0, 100.4, 100.9]]]),
decision_bar=4,
extra_dbs=[dict(db=3, zone_ids=[], newest_active=None)],
expected=dict(zones=[[4, 1, 100.4, 100.2, 0, False, False]],
zone_count=1, newest_active=[4, 1, 100.4, 100.2],
legacy=None, repaired=[1, 100.4, 100.2],
note="db=3 (before C3 closes): NO zone (no future candle); db=4 (C3 closed): zone present (no needless 1-bar delay).")),
# ---- OB ----
dict(id="F2-T09", kind="ob", title="OB formation with move = newest bar (S-1/S-6)",
bar_spec=dict(neutral=NEUT_HI, count=25,
patches=[[23, [100.2, 100.5, 99.8, 100.1]],
[24, [100.3, 101.0, 100.2, 100.9]]]),
decision_bar=24,
expected=dict(zones=[[23, 1, 100.5, 99.8, 0, False, False]],
zone_count=1, newest_active=[23, 1, 100.5, 99.8],
legacy=[1, 100.5, 99.8], repaired=[1, 100.5, 99.8])),
dict(id="F2-T10", kind="ob", title="OB remains ONE zone after many bars (S-8)",
bar_spec=dict(neutral=NEUT_HI, count=25,
patches=[[10, [100.2, 100.5, 99.8, 100.1]],
[11, [100.3, 101.0, 100.2, 100.9]]]),
decision_bar=24,
extra_dbs=[dict(db=22, zone_ids=[10], newest_active=[10, 1, 100.5, 99.8])],
expected=dict(zones=[[10, 1, 100.5, 99.8, 0, False, False]],
zone_count=1, newest_active=[10, 1, 100.5, 99.8],
legacy=[1, 100.5, 99.8], repaired=[1, 100.5, 99.8])),
dict(id="F2-T11", kind="ob", title="OB partial fill -> still ACTIVE (S-9)",
bar_spec=dict(neutral=NEUT_LO, count=25,
patches=[[20, [100.2, 100.5, 99.8, 100.1]],
[21, [100.3, 101.0, 100.2, 100.9]],
[22, [100.3, 100.4, 99.9, 100.2]]]),
decision_bar=24,
expected=dict(zones=[[20, 1, 100.5, 99.8, 1, True, False]],
zone_count=1, newest_active=[20, 1, 100.5, 99.8],
legacy=[1, 100.5, 99.8], repaired=[1, 100.5, 99.8])),
dict(id="F2-T12", kind="ob", title="OB full mitigation (close-through, S-9)",
bar_spec=dict(neutral=NEUT_LO, count=25,
patches=[[20, [100.2, 100.5, 99.8, 100.1]],
[21, [100.3, 101.0, 100.2, 100.9]],
[22, [99.89, 100.0, 99.6, 99.7]]]),
decision_bar=24,
expected=dict(zones=[[20, 1, 100.5, 99.8, 2, True, True]],
zone_count=1, newest_active=None,
legacy=[1, 100.5, 99.8], repaired=None,
note="OB-T11 equivalent: legacy returns mitigated zone (BUG-P3S5-001); repaired None.")),
dict(id="F2-T13", kind="ob",
title="OB mitigated zone NOT consumed; older active zone returned",
bar_spec=dict(neutral=NEUT_HI, count=25,
patches=[[9, [100.7, 100.9, 100.55, 100.6]],
[10, [100.2, 100.5, 99.6, 99.9]],
[11, [100.0, 101.0, 99.9, 100.9]],
[20, [100.1, 100.6, 99.7, 100.0]],
[21, [100.2, 101.0, 100.1, 100.9]],
[22, [99.5, 100.0, 99.3, 99.6]]]),
decision_bar=24,
expected=dict(zones=[[10, 1, 100.5, 99.6, 1, True, False],
[20, 1, 100.6, 99.7, 2, True, True]],
zone_count=2, newest_active=[10, 1, 100.5, 99.6],
legacy=[1, 100.6, 99.7], repaired=[1, 100.5, 99.6],
note="newest zone (b=20) close-through mitigated -> legacy consumes it (BUG); repaired returns older active b=10.")),
dict(id="F2-T14", kind="ob", title="OB invalidation is terminal (S-10/S-12/§K)",
bar_spec=dict(neutral=NEUT_LO, count=25,
patches=[[20, [100.2, 100.5, 99.8, 100.1]],
[21, [100.3, 101.0, 100.2, 100.9]],
[22, [99.89, 100.0, 99.6, 99.7]],
[24, [100.0, 100.8, 99.9, 100.7]]]),
decision_bar=24,
extra_dbs=[dict(db=22, zone_ids=[20], newest_active=None)],
expected=dict(zones=[[20, 1, 100.5, 99.8, 2, True, True]],
zone_count=1, newest_active=None,
legacy=[1, 100.5, 99.8], repaired=None,
note="OB-T12/T16 equivalents: bar-24 opposite move does NOT reactivate the zone (no breaker, no invalidation-reversal).")),
dict(id="F2-T15", kind="fvg", title="Two independent FVG zones (S-11)",
bar_spec=dict(neutral=[100.0, 100.3, 99.9, 100.2], count=7,
patches=[[1, [100.1, 100.5, 100.0, 100.4]],
[2, [100.3, 100.6, 100.2, 100.5]],
[3, [100.6, 100.9, 100.55, 100.8]],
[4, [100.7, 100.9, 100.6, 100.8]],
[5, [100.7, 101.0, 100.65, 100.9]],
[6, [101.0, 101.3, 100.95, 101.2]]]),
decision_bar=6,
expected=dict(zones=[[3, 1, 100.55, 100.5, 0, False, False],
[6, 1, 100.95, 100.9, 0, False, False]],
zone_count=2, newest_active=[6, 1, 100.95, 100.9],
legacy=[1, 100.55, 100.5], repaired=[1, 100.95, 100.9],
note="FVG-T13 equivalent: legacy returns older b=3 (1-bar lag); repaired returns newest b=6.")),
dict(id="F2-T16", kind="ob", title="Two independent OB zones (S-11)",
bar_spec=dict(neutral=NEUT_LO, count=25,
patches=[[10, [100.2, 100.5, 99.8, 100.1]],
[11, [100.3, 101.0, 100.2, 100.9]],
[20, [100.4, 100.7, 100.1, 100.3]],
[21, [100.6, 101.3, 100.5, 101.2]]]),
decision_bar=24,
expected=dict(zones=[[10, 1, 100.5, 99.8, 1, True, False],
[20, 1, 100.7, 100.1, 1, True, False]],
zone_count=2, newest_active=[20, 1, 100.7, 100.1],
legacy=[1, 100.7, 100.1], repaired=[1, 100.7, 100.1])),
dict(id="F2-T17", kind="fvg", title="Overlapping FVG zones, no merge (S-11)",
bar_spec=dict(neutral=[100.0, 100.3, 99.9, 100.2], count=6,
patches=[[1, [100.1, 100.5, 100.0, 100.4]],
[2, [100.3, 100.6, 100.2, 100.5]],
[3, [100.6, 100.9, 100.55, 100.8]],
[4, [100.7, 101.0, 100.65, 100.9]],
[5, [100.9, 101.2, 100.95, 101.1]]]),
decision_bar=5,
expected=dict(zones=[[3, 1, 100.55, 100.5, 0, False, False],
[4, 1, 100.65, 100.6, 0, False, False],
[5, 1, 100.95, 100.9, 0, False, False]],
zone_count=3, newest_active=[5, 1, 100.95, 100.9],
legacy=[1, 100.65, 100.6], repaired=[1, 100.95, 100.9],
note="FVG-T12 equivalent: legacy returns b=4 (1-bar lag); repaired returns newest b=5.")),
dict(id="F2-T18", kind="ob", title="Overlapping OB zones, no merge (S-11)",
bar_spec=dict(neutral=NEUT_LO, count=25,
patches=[[18, [100.2, 100.5, 99.8, 100.1]],
[19, [100.3, 101.0, 100.2, 100.9]],
[20, [100.4, 100.7, 100.1, 100.3]],
[21, [100.6, 101.3, 100.5, 101.2]]]),
decision_bar=24,
expected=dict(zones=[[18, 1, 100.5, 99.8, 1, True, False],
[20, 1, 100.7, 100.1, 1, True, False]],
zone_count=2, newest_active=[20, 1, 100.7, 100.1],
legacy=[1, 100.7, 100.1], repaired=[1, 100.7, 100.1])),
dict(id="F2-T19", kind="fvg",
title="Future-bar mutation cannot alter past zone state (S-7)",
bar_spec=dict(neutral=NEUT_FVG, count=6,
patches=[[2, [100.0, 100.2, 99.9, 100.1]],
[3, [100.3, 100.9, 100.25, 100.8]],
[4, [100.5, 101.0, 100.4, 100.9]],
[5, [100.6, 100.8, 100.45, 100.7]]]),
decision_bar=5,
mutations=[[20, [200.0, 210.0, 199.0, 205.0]]],
expected=dict(zones=[[4, 1, 100.4, 100.2, 0, False, False]],
zone_count=1, newest_active=[4, 1, 100.4, 100.2],
legacy=[1, 100.4, 100.2], repaired=[1, 100.4, 100.2])),
dict(id="F2-T20", kind="fvg",
title="Bullish/bearish symmetry (S-4; runner mirrors every case)",
bar_spec=dict(neutral=[100.0, 100.2, 99.5, 99.7], count=6,
patches=[[2, [100.0, 100.1, 99.8, 99.9]],
[3, [99.7, 99.8, 99.2, 99.4]],
[4, [99.4, 99.6, 99.0, 99.3]],
[5, [99.3, 99.5, 99.0, 99.4]]]),
decision_bar=5,
expected=dict(zones=[[4, -1, 99.8, 99.6, 0, False, False]],
zone_count=1, newest_active=[4, -1, 99.8, 99.6],
legacy=[-1, 99.8, 99.6], repaired=[-1, 99.8, 99.6])),
]
# =====================================================================
# RUNNER
# =====================================================================
def evaluate(kind, o, h, l, c, db):
if kind == "fvg":
zones = spec_oracle_fvg(o, h, l, c, db)
legacy = legacy_fvg_port(o, h, l, c, db)
repaired = repaired_fvg_port(o, h, l, c, db)
else:
zones = spec_oracle_ob(o, h, l, c, db)
legacy = legacy_ob_port(o, h, l, c, db)
repaired = repaired_ob_port(o, h, l, c, db)
return zones, newest_active(zones), legacy, repaired
def run_case(case):
o, h, l, c = materialize(case)
db = case["decision_bar"]
kind = case["kind"]
transform = case.get("transform")
mutations = case.get("mutations")
neutral = case["bar_spec"]["neutral"]
zones, na, legacy, repaired = evaluate(kind, o, h, l, c, db)
exp = case["expected"]
exp_zones = [list(z) for z in exp["zones"]]
exp_na = (None if exp["newest_active"] is None else list(exp["newest_active"]))
checks = {}
# spec oracle == expected (canonical zone contract = truth)
got_sig = [[z["b"], z["d"], z["top"], z["bot"], z["mit"], z["partial"],
z["invalidated"]] for z in zones]
checks["zones_spec"] = bool(len(got_sig) == len(exp_zones))
if checks["zones_spec"]:
for g, e in zip(got_sig, exp_zones):
if (g[0] != e[0] or g[1] != e[1] or abs(g[2] - e[2]) > EPS
or abs(g[3] - e[3]) > EPS or g[4] != e[4] or g[5] != e[5]
or g[6] != e[6]):
checks["zones_spec"] = False
break
checks["zone_count_spec"] = bool(len(zones) == int(exp["zone_count"]))
checks["newest_active_spec"] = vec_equal(na, exp_na)
# extra decision bars (timing / persistence / terminal checks)
extra_ok = True
extra_info = []
for ex in case.get("extra_dbs", []):
edb = ex["db"]
z2, na2, _lg2, _rp2 = evaluate(kind, o, h, l, c, edb)
ids2 = [z["b"] for z in z2]
exp_ids = ex["zone_ids"]
exp_na2 = (None if ex["newest_active"] is None
else list(ex["newest_active"]))
same = (ids2 == exp_ids) and vec_equal(na2, exp_na2)
extra_ok = extra_ok and same
extra_info.append({"db": edb, "zone_ids": ids2,
"newest_active": na2, "expected_zone_ids": exp_ids,
"expected_newest_active": exp_na2, "ok": same})
checks["extra_dbs"] = bool(extra_ok)
# invariance: transform (scale/translate) keeps structure & state
o2, h2, l2, c2 = apply_transform(o, h, l, c, transform)
z2, _, _, _ = evaluate(kind, o2, h2, l2, c2, db)
sig_t = [[z["b"], z["d"], z["top"], z["bot"], z["mit"], z["partial"],
z["invalidated"]] for z in z2]
checks["invariant_transform"] = bool(sig_t == got_sig)
# invariance: future mutation (S-7 / T19)
if mutations:
o3, h3, l3, c3 = apply_mutations(o, h, l, c, mutations, neutral)
z3, _, _, _ = evaluate(kind, o3, h3, l3, c3, db)
sig_m = [[z["b"], z["d"], z["top"], z["bot"], z["mit"], z["partial"],
z["invalidated"]] for z in z3]
checks["invariant_future_mutation"] = bool(sig_m == got_sig)
# invariance: bull/bear symmetry (S-4) — mirror every case
pivot = (float(np.min(l)) + float(np.max(h))) / 2.0
om, hm, lm, cm = mirror(o, h, l, c, pivot)
zm, _, _, _ = evaluate(kind, om, hm, lm, cm, db)
sym_ok = (len(zm) == len(zones))
if sym_ok:
for z, zz in zip(zones, zm):
# mirrored: same b, dir flips, bounds mirror, state identical
if (z["b"] != zz["b"] or z["d"] != -zz["d"] or z["mit"] != zz["mit"]
or z["invalidated"] != zz["invalidated"]
or abs(zz["top"] - (2 * pivot - z["bot"])) > EPS
or abs(zz["bot"] - (2 * pivot - z["top"])) > EPS):
sym_ok = False
break
checks["symmetry"] = bool(sym_ok)
# repaired port == spec newest_active (the F2 contract the MQL5 fix satisfies)
spec_for_port = None if na is None else [na[1], na[2], na[3]]
checks["repaired_matches_spec"] = port_equal(repaired, spec_for_port)
checks["repaired_expected"] = port_equal(
repaired, (None if exp["repaired"] is None else list(exp["repaired"])))
# legacy port == expected legacy (before/after evidence)
checks["legacy_expected"] = port_equal(
legacy, (None if exp["legacy"] is None else list(exp["legacy"])))
return {
"id": case["id"],
"kind": kind,
"title": case["title"],
"zones": got_sig,
"newest_active": na,
"legacy_port": legacy,
"repaired_port": repaired,
"checks": checks,
"extra_dbs": extra_info,
"pass": all(checks.values()),
"note": exp.get("note") or case.get("note"),
}
def main():
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.12 SPEC TESTS — F2 ZONE CONTRACT ({n_pass}/{n_total} PASS) ===")
print(f"{'ID':<9} {'PASS':<6} {'zones':<5} {'newest':<26} {'legacy':<18} {'repaired':<18}")
for r in results:
na = r["newest_active"]
na_s = "-" if na is None else f"b{na[0]} d{na[1]} [{na[2]},{na[3]}]"
lg = r["legacy_port"]
lg_s = "-" if lg is None else f"d{lg[0]} [{lg[1]},{lg[2]}]"
rp = r["repaired_port"]
rp_s = "-" if rp is None else f"d{rp[0]} [{rp[1]},{rp[2]}]"
print(f"{r['id']:<9} {str(r['pass']):<6} {len(r['zones']):<5} {na_s:<26} "
f"{lg_s:<18} {rp_s:<18} {r['title'][:30]}")
if n_pass != n_total:
print("\nFAILED:")
for r in results:
if not r["pass"]:
print(" ", r["id"], r["checks"])
sys.exit(1)
print("\nAll spec assertions PASS — expected results are consistent with the "
"frozen canonical ZONE contract (P3-S.10 §K/§M/§O + P3-S.4 S-1..S-13 + "
"P3-S.5 S-1..S-15).")
print("Legacy ports document the PRE-F2 Engine-2 behavior (BUG-P3S4-001/-002, "
"BUG-P3S5-001); repaired ports are the F2 contract the MQL5 fix must match.")
# before/after evidence mapping (brief §27)
evidence = {
"FVG-T10 (full mitigation)": "F2-T04",
"FVG-T11 (invalidation not modeled)": "F2-T06",
"FVG-T12 (overlapping)": "F2-T17",
"FVG-T13 (multiple consecutive)": "F2-T15",
"OB-T11 (full mitigation)": "F2-T12",
"OB-T12 (invalidation not modeled)": "F2-T14",
"OB-T16 (breaker not implemented)": "F2-T14",
}
before_after = []
for r in results:
before_after.append({
"id": r["id"], "kind": r["kind"],
"zones": r["zones"], "newest_active": r["newest_active"],
"legacy_port (before)": r["legacy_port"],
"repaired_port (after)": r["repaired_port"],
})
report = {
"spec_docs": ["docs/SNIPERGOLD_CANONICAL_SETUP_CONTRACT_v1.md",
"docs/SMC_FVG_SPEC_v1.md", "docs/SMC_ORDER_BLOCK_SPEC_v1.md"],
"phase": "P3-S.12",
"generated_utc": __import__("datetime").datetime.now(
__import__("datetime").timezone.utc).isoformat(),
"constants": {"FVG_LOOKBACK": FVG_LOOKBACK, "AVG_N": AVG_N,
"MIN_BARS_OB": MIN_BARS_OB, "MOVE_MULT": MOVE_MULT,
"W_zone_age": "OPEN NUMERIC PARAMETER (default none)"},
"summary": {"total": n_total, "passed": n_pass, "failed": n_total - n_pass},
"evidence_mapping": evidence,
"before_after": before_after,
"cases": results,
}
outdir = os.path.join(HERE, "output")
os.makedirs(outdir, exist_ok=True)
out_path = os.path.join(outdir, "p3_s12_zone_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