766 lines
36 KiB
Python
766 lines
36 KiB
Python
# -*- 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()
|