# -*- 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()