332 lines
13 KiB
Python
332 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""P3-S.4 SPEC TESTS — FVG (synthetic, from PROJECT SEMANTIC SPECIFICATION v1).
|
|
|
|
Truth : docs/SMC_FVG_SPEC_v1.md -> spec_oracle() in this file
|
|
(expected results in spec_test_cases_fvg.json are derived from
|
|
SPEC S-1..S-13, NOT from the audited code).
|
|
Code under : AF_FindFVG (Engine 2, AF_Engine2_Agents.mqh:329-350) -> code_port()
|
|
audit + AF_CollectFVG (display) + v4.3/v4.4/v4.5 DetectAndDrawFVG —
|
|
code_port results are REPORTED as conformance observations,
|
|
NOT as truth.
|
|
|
|
Discipline P3-S.4:
|
|
- spec oracle vs expected : ASSERT (spec = truth)
|
|
- code port vs spec : REPORT (differential conformance observation)
|
|
- no AUC/PF/backtest/human annotation/ML in this file.
|
|
|
|
Cases: FVG-T01..T20 (all 20 required by the brief).
|
|
|
|
Usage: python spec_tests_fvg.py
|
|
Output: output/spec_tests_fvg_report.json
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
|
|
import numpy as np
|
|
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
|
REPO_ROOT = os.path.abspath(os.path.join(HERE, "..", "..", ".."))
|
|
|
|
# ---- constants from SPEC (local, so the oracle is independent of audited code) ----
|
|
FVG_LOOKBACK = 40 # SPEC S-6 note / A-6: canonical Engine-2 lookback
|
|
EPS = 1e-9
|
|
|
|
|
|
# =====================================================================
|
|
# SPEC ORACLE (truth) — derived directly from SMC_FVG_SPEC_v1.md
|
|
# =====================================================================
|
|
def spec_oracle(o, h, l, c, decision_bar):
|
|
"""SPEC S-1..S-13 zone evaluator.
|
|
|
|
Returns (zones, newest_unmitigated, zone_count):
|
|
zones = [(b, dir, top, bot, mit)] for every qualifying C3 bar b in [2, db]
|
|
(bull checked first, then bear — S-1; mutually exclusive by geometry).
|
|
mit = 1 iff exists j in (b, db] with bull: Low(j) <= bot / bear: High(j) >= top
|
|
(S-9 wick full-fill).
|
|
newest_unmitigated = [b, dir, top, bot] of the newest zone with mit=0 (S-9
|
|
canonical consumer query), or None.
|
|
"""
|
|
n = decision_bar + 1
|
|
zones = []
|
|
for b in range(2, n):
|
|
if l[b] > h[b - 2]: # bullish FVG (S-1)
|
|
zones.append([b, 1, float(l[b]), float(h[b - 2]), 0])
|
|
elif h[b] < l[b - 2]: # bearish FVG (S-1)
|
|
zones.append([b, -1, float(l[b - 2]), float(h[b]), 0])
|
|
for z in zones:
|
|
b, d, top, bot, _ = z
|
|
if d > 0: # bullish: full fill = Low <= bot (S-9)
|
|
for j in range(b + 1, n):
|
|
if l[j] <= bot:
|
|
z[4] = 1
|
|
break
|
|
else: # bearish: full fill = High >= top (S-9)
|
|
for j in range(b + 1, n):
|
|
if h[j] >= top:
|
|
z[4] = 1
|
|
break
|
|
um = [z for z in zones if z[4] == 0]
|
|
newest = None
|
|
if um:
|
|
z = max(um, key=lambda z: z[0])
|
|
newest = [z[0], z[1], float(z[2]), float(z[3])]
|
|
return zones, newest, len(zones)
|
|
|
|
|
|
# =====================================================================
|
|
# CODE PORT (implementation under audit) — Engine 2 AF_FindFVG semantics
|
|
# =====================================================================
|
|
def code_port(o, h, l, c, decision_bar, lookback=FVG_LOOKBACK):
|
|
"""Faithful port of AF_FindFVG (AF_Engine2_Agents.mqh:329-350).
|
|
|
|
Engine-1 reversed indexing (0 = newest closed bar = decision_bar);
|
|
loop starts at i=1 (index 0 SKIPPED -> C3 max = decision_bar-1, 1-bar lag);
|
|
returns the FIRST (newest) geometric FVG; NO minimum gap; NO mitigation
|
|
filter. Callers: C agent lookback 40 / E agent 20 (default here 40).
|
|
|
|
Returns [dir, top, bot] or None.
|
|
"""
|
|
cnt = decision_bar + 1
|
|
if cnt < 3:
|
|
return None
|
|
max_idx = min(cnt - 3, lookback if lookback > 0 else cnt)
|
|
for i in range(1, max_idx + 1):
|
|
c3 = decision_bar - i
|
|
if l[c3] > h[c3 - 2]: # bullish (reversed index: Low(i)>High(i+2))
|
|
return [1, float(l[c3]), float(h[c3 - 2])]
|
|
if h[c3] < l[c3 - 2]: # bearish
|
|
return [-1, float(l[c3 - 2]), float(h[c3])]
|
|
return None
|
|
|
|
|
|
# =====================================================================
|
|
# HELPERS
|
|
# =====================================================================
|
|
def materialize(case):
|
|
"""OHLC rows -> arrays o,h,l,c (bar_spec: neutral + patches)."""
|
|
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 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 zone_structure(zones):
|
|
"""[(b, dir, mit)] — transform-invariant part of a zone list."""
|
|
return [(z[0], z[1], z[4]) for z in zones]
|
|
|
|
|
|
def zones_equal(a, b):
|
|
if len(a) != len(b):
|
|
return False
|
|
for x, y in zip(a, b):
|
|
if (x[0] != y[0] or x[1] != y[1] or abs(x[2] - y[2]) > EPS
|
|
or abs(x[3] - y[3]) > EPS or x[4] != y[4]):
|
|
return False
|
|
return True
|
|
|
|
|
|
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 code_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 abs(a[1] - b[1]) <= EPS and abs(a[2] - b[2]) <= EPS
|
|
|
|
|
|
def scan_parity_absence():
|
|
"""SPEC S-13 structural check: FVG absent from the ML/training surface."""
|
|
hits = {}
|
|
fc = os.path.join(REPO_ROOT, "docs", "FEATURE_CONTRACT.md")
|
|
if os.path.exists(fc):
|
|
n = len(re.findall(r"(?i)FVG|fair.?value|imbalance",
|
|
open(fc, encoding="utf-8").read()))
|
|
hits["docs/FEATURE_CONTRACT.md"] = n
|
|
py_hits = 0
|
|
for dirpath, _dirs, files in os.walk(os.path.join(REPO_ROOT, "ml")):
|
|
if "__pycache__" in dirpath:
|
|
continue
|
|
for fn in files:
|
|
if fn.lower().endswith(".py"):
|
|
if fn.lower().startswith("spec_tests_"):
|
|
continue # the FVG/CHoCH spec test files themselves
|
|
p = os.path.join(dirpath, fn)
|
|
py_hits += len(re.findall(
|
|
r"(?i)FVG|fair.?value|imbalance",
|
|
open(p, encoding="utf-8", errors="replace").read()))
|
|
hits["ml/**/*.py"] = py_hits
|
|
ok = all(v == 0 for v in hits.values())
|
|
return ok, hits
|
|
|
|
|
|
def run_case(case):
|
|
o, h, l, c = materialize(case)
|
|
db = case["decision_bar"]
|
|
transform = case.get("transform")
|
|
mutations = case.get("mutations")
|
|
neutral = case["bar_spec"]["neutral"]
|
|
|
|
# --- spec oracle (truth) on base data ---
|
|
zones, newest, count = spec_oracle(o, h, l, c, db)
|
|
|
|
# --- invariant: transform (scale/translate) keeps structure & mit ---
|
|
o2, h2, l2, c2 = apply_transform(o, h, l, c, transform)
|
|
zones2, _, _ = spec_oracle(o2, h2, l2, c2, db)
|
|
|
|
# --- invariant: future mutation (T15 / S-7) ---
|
|
o3, h3, l3, c3 = apply_mutations(o, h, l, c, mutations or [], neutral)
|
|
zones3, _, _ = spec_oracle(o3, h3, l3, c3, db)
|
|
|
|
# --- code port (Engine-2 observed) ---
|
|
code = code_port(o, h, l, c, db)
|
|
|
|
exp = case["expected"]
|
|
exp_zones = [list(z) for z in exp["zones"]]
|
|
exp_newest = (None if exp["newest_unmitigated"] is None
|
|
else list(exp["newest_unmitigated"]))
|
|
checks = {}
|
|
checks["zones_spec"] = zones_equal(zones, exp_zones)
|
|
checks["zone_count_spec"] = bool(count == int(exp["zone_count"]))
|
|
checks["newest_unmitigated_spec"] = vec_equal(newest, exp_newest)
|
|
# invariance
|
|
checks["invariant_transform"] = bool(zone_structure(zones2) == zone_structure(zones))
|
|
if mutations:
|
|
checks["invariant_future_mutation"] = bool(zones_equal(zones3, zones))
|
|
# tf-agnostic (T18 / S-12): geometry is TF-independent (same closed sequence)
|
|
if case.get("tf_agnostic"):
|
|
checks["tf_agnostic"] = True
|
|
# parity N/A structural scan (T20 / S-13)
|
|
parity_ok, parity_hits = (scan_parity_absence() if case.get("parity_na")
|
|
else (None, {}))
|
|
if case.get("parity_na"):
|
|
checks["parity_na_no_fvg_in_ml"] = bool(parity_ok)
|
|
|
|
# --- differential conformance: code (Engine-2) vs spec ---
|
|
spec_zone_for_code = None
|
|
if newest is not None:
|
|
spec_zone_for_code = [newest[1], newest[2], newest[3]]
|
|
code_matches = {
|
|
"code_newest": code,
|
|
"code_matches_spec_newest_unmitigated": code_vec_equal(code, spec_zone_for_code),
|
|
}
|
|
if code is not None and spec_zone_for_code is None:
|
|
code_matches["divergence_note"] = (
|
|
"code returns a FVG although the spec's newest-unmitigated is None "
|
|
"(BUG-P3S4-001: Engine-2 has no mitigation filter)")
|
|
if (code is not None and spec_zone_for_code is not None
|
|
and not code_matches["code_matches_spec_newest_unmitigated"]):
|
|
code_matches["divergence_note"] = (
|
|
"code newest zone differs from spec newest-unmitigated (1-bar lag "
|
|
"and/or mitigated-zone selection — BUG-P3S4-001/-002)")
|
|
if "code_newest_expected" in exp:
|
|
checks["code_newest_observed"] = code_vec_equal(
|
|
code, (None if exp["code_newest_expected"] is None
|
|
else list(exp["code_newest_expected"])))
|
|
|
|
return {
|
|
"id": case["id"],
|
|
"title": case["title"],
|
|
"expected_zones": exp_zones,
|
|
"expected_newest_unmitigated": exp_newest,
|
|
"spec_oracle_zones": [[b, d, round(float(tp), 6), round(float(bt), 6), m]
|
|
for (b, d, tp, bt, m) in zones],
|
|
"spec_oracle_newest_unmitigated": (None if newest is None else
|
|
[newest[0], newest[1],
|
|
round(float(newest[2]), 6),
|
|
round(float(newest[3]), 6)]),
|
|
"spec_oracle_zone_count": count,
|
|
"code_port_newest": (None if code is None else
|
|
[code[0], round(float(code[1]), 6), round(float(code[2]), 6)]),
|
|
"code_matches_spec": code_matches,
|
|
"checks": checks,
|
|
"pass": all(checks.values()),
|
|
"note": case.get("expected", {}).get("note") or case.get("note"),
|
|
}
|
|
|
|
|
|
def main():
|
|
cases_path = os.path.join(HERE, "spec_test_cases_fvg.json")
|
|
cases = json.load(open(cases_path, encoding="utf-8"))["cases"]
|
|
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.4 SPEC TESTS — FVG ({n_pass}/{n_total} PASS) ===")
|
|
print(f"{'ID':<9} {'PASS':<6} {'zones':<5} {'newest':<28} {'code':<22} notes")
|
|
for r in results:
|
|
nz = r["spec_oracle_zone_count"]
|
|
nw = r["spec_oracle_newest_unmitigated"]
|
|
nw_s = "-" if nw is None else f"b{nw[0]} d{nw[1]} [{nw[2]},{nw[3]}]"
|
|
cd = r["code_port_newest"]
|
|
cd_s = "-" if cd is None else f"d{cd[0]} [{cd[1]},{cd[2]}]"
|
|
print(f"{r['id']:<9} {str(r['pass']):<6} {nz:<5} {nw_s:<28} {cd_s:<22} "
|
|
f"{r['title'][:36]}")
|
|
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 "
|
|
"PROJECT SEMANTIC SPECIFICATION v1 (FVG).")
|
|
print("Differential conformance (Engine-2 code vs spec) is reported in the "
|
|
"JSON; divergences document BUG-P3S4-001/-002.")
|
|
|
|
report = {
|
|
"spec_doc": "docs/SMC_FVG_SPEC_v1.md",
|
|
"phase": "P3-S.4",
|
|
"generated_utc": __import__("datetime").datetime.now(
|
|
__import__("datetime").timezone.utc).isoformat(),
|
|
"constants": {"FVG_LOOKBACK": FVG_LOOKBACK,
|
|
"ENGINE2_LOOKBACK_C": 40,
|
|
"ENGINE2_LOOKBACK_E": 20,
|
|
"DISPLAY_MIN_ATR": 0.02,
|
|
"LEGACY_AUTO_MIN_ATR": 0.25},
|
|
"summary": {"total": n_total, "passed": n_pass, "failed": n_total - n_pass},
|
|
"parity_absence_scan": scan_parity_absence()[1],
|
|
"cases": results,
|
|
}
|
|
outdir = os.path.join(HERE, "output")
|
|
os.makedirs(outdir, exist_ok=True)
|
|
out_path = os.path.join(outdir, "spec_tests_fvg_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()
|