265 lines
9.8 KiB
Python
265 lines
9.8 KiB
Python
# -*- 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<Open),
|
|
0 otherwise (S-1/S-4/S-10).
|
|
"""
|
|
n = decision_bar + 1
|
|
if n <= 0:
|
|
return 0, 0.0, 0.0, 0.0
|
|
avg = avg_body(o, c, decision_bar, avg_n)
|
|
if avg <= 0.0:
|
|
return 0, 0.0, avg, k * avg
|
|
body = abs(c[decision_bar] - o[decision_bar])
|
|
thr = k * avg
|
|
if body < thr:
|
|
return 0, body, avg, thr
|
|
return (1 if c[decision_bar] > 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()
|