forked from chiki2bum2/SniperGold_ML
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()
|