# -*- coding: utf-8 -*- """P3-S.6 SPEC TESTS — DISPLACEMENT (synthetic, from PROJECT SEMANTIC SPECIFICATION v1). Truth : docs/SMC_DISPLACEMENT_SPEC_v1.md -> spec_oracle() in this file (expected results in spec_test_cases_displacement.json are derived from SPEC S-1..S-14, NOT from the audited code). Code under : AF_DetectDisplacement (Engine 2, AF_Engine2_Agents.mqh:281-291) -> audit code_port() — reported as an observation; the implementation IS the audited primitive (identical formula). Discipline P3-S.6: - spec oracle vs expected : ASSERT (spec = truth) - code port vs spec : REPORT (differential conformance observation) - symmetry property : ASSERT for every case (S-11) - no AUC/PF/backtest/human annotation/ML in this file. Cases: DP-T01..T20 (all 20 required by the brief). Usage: python spec_tests_displacement.py Output: output/spec_tests_displacement_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) ---- AVG_N = 20 # SPEC S-1/S-3: avg body window (AF_E2_LOOKBACK_AVG) K = 1.6 # SPEC S-1/S-10: displacement body multiplier EPS = 1e-9 # ===================================================================== # SPEC ORACLE (truth) — derived directly from SMC_DISPLACEMENT_SPEC_v1.md # ===================================================================== def avg_body(o, c, decision_bar, avg_n=AVG_N): """SPEC S-3: mean |Close-Open| over the newest min(avg_n, t+1) bars (candidate included).""" m = min(avg_n, decision_bar + 1) if m <= 0: return 0.0 s = 0.0 for i in range(decision_bar - m + 1, decision_bar + 1): s += abs(c[i] - o[i]) return s / m def spec_oracle(o, h, l, c, decision_bar, avg_n=AVG_N, k=K): """SPEC S-1..S-14 displacement evaluator on the NEWEST CLOSED bar. Returns (dir, body, avg, threshold): dir = +1 (body >= k*avg AND Close>Open), -1 (body >= k*avg AND Close o[decision_bar] else -1), body, avg, thr # ===================================================================== # CODE PORT (implementation under audit) — Engine 2 AF_DetectDisplacement # ===================================================================== def code_port(o, h, l, c, decision_bar, avg_n=AVG_N, k=K): """Faithful port of AF_DetectDisplacement (AF_Engine2_Agents.mqh:281-291). Identical formula: body of the newest closed bar >= 1.6 x avg body, direction by Close vs Open, no structure, no min-bars guard. """ if decision_bar < 0: return 0 avg = avg_body(o, c, decision_bar, avg_n) if avg <= 0.0: return 0 body = abs(c[decision_bar] - o[decision_bar]) if body < k * avg: return 0 return 1 if c[decision_bar] > o[decision_bar] else -1 # ===================================================================== # 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): """Mirror prices around pivot P (S-11 symmetry): h'=2P-l, l'=2P-h, o/c mirrored.""" 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 scan_parity_absence(): """SPEC S-14 structural check: displacement absent from the ML/training surface.""" pat = re.compile(r"(?i)displacement|AF_DetectDisplacement") hits = {} fc = os.path.join(REPO_ROOT, "docs", "FEATURE_CONTRACT.md") if os.path.exists(fc): hits["docs/FEATURE_CONTRACT.md"] = len( pat.findall(open(fc, encoding="utf-8").read())) 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 spec test files themselves p = os.path.join(dirpath, fn) py_hits += len(pat.findall( 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 --- dirv, body, avg, thr = spec_oracle(o, h, l, c, db) # --- invariants --- o2, h2, l2, c2 = apply_transform(o, h, l, c, transform) dir2, _, _, _ = spec_oracle(o2, h2, l2, c2, db) o3, h3, l3, c3 = apply_mutations(o, h, l, c, mutations or [], neutral) dir3, _, _, _ = spec_oracle(o3, h3, l3, c3, db) # symmetry (S-11): mirroring flips the sign of a displacement, keeps 0 pivot = (float(np.min(l)) + float(np.max(h))) / 2.0 om, hm, lm, cm = mirror(o, h, l, c, pivot) dir_m, _, _, _ = spec_oracle(om, hm, lm, cm, db) sym_ok = (dir_m == -dirv) # --- code port (Engine-2 observed) --- code = code_port(o, h, l, c, db) exp = case["expected"] checks = {} checks["dir_spec"] = bool(dirv == int(exp["dir"])) checks["invariant_transform"] = bool(dir2 == dirv) checks["symmetry"] = bool(sym_ok) if mutations: checks["invariant_future_mutation"] = bool(dir3 == dirv) if case.get("tf_agnostic"): checks["tf_agnostic"] = True parity_ok, parity_hits = (scan_parity_absence() if case.get("parity_na") else (None, {})) if case.get("parity_na"): checks["parity_na_no_displacement_in_ml"] = bool(parity_ok) if "code_dir_expected" in exp: checks["code_dir_observed"] = bool(code == int(exp["code_dir_expected"])) code_matches = { "code_dir": code, "code_matches_spec": bool(code == dirv), } return { "id": case["id"], "title": case["title"], "expected_dir": int(exp["dir"]), "spec_oracle_dir": dirv, "spec_oracle_body": round(float(body), 9), "spec_oracle_avg": round(float(avg), 9), "spec_oracle_threshold": round(float(thr), 9), "code_port_dir": code, "code_matches_spec": code_matches, "checks": checks, "pass": all(checks.values()), "note": exp.get("note") or case.get("note"), } def main(): cases_path = os.path.join(HERE, "spec_test_cases_displacement.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.6 SPEC TESTS — DISPLACEMENT ({n_pass}/{n_total} PASS) ===") print(f"{'ID':<9} {'PASS':<6} {'dir s/c':<10} {'body':<12} {'thr':<12} notes") for r in results: print(f"{r['id']:<9} {str(r['pass']):<6} " f"{str(r['spec_oracle_dir'])+'/'+str(r['code_port_dir']):<10} " f"{r['spec_oracle_body']:<12} {r['spec_oracle_threshold']:<12} " 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 (Displacement).") print("code_port == spec in all cases (the implementation conforms); " "cross-concept threshold note in BUG-P3S6-001.") report = { "spec_doc": "docs/SMC_DISPLACEMENT_SPEC_v1.md", "phase": "P3-S.6", "generated_utc": __import__("datetime").datetime.now( __import__("datetime").timezone.utc).isoformat(), "constants": {"AVG_N": AVG_N, "K": K, "OB_MOVE_BODY_MULT": 1.5}, "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_displacement_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()