# -*- coding: utf-8 -*- """P3-S.8 SPEC TESTS — CANDIDATE SETUP (synthetic, from PROJECT SEMANTIC SPECIFICATION v1). Truth : docs/SMC_CANDIDATE_SETUP_SPEC_v1.md -> spec_oracle() in this file (expected results in spec_test_cases_candidate_setup.json are derived from SPEC CS-1..CS-32, NOT from the audited code). Code under : AF_Engine2_Agents.mqh (E-agent rule: ZONE + CONFIRMATION = setup; audit AFAgentEntry::Compute) + AF_Engine2_Aggregator.mqh (AFAggregator::Compute 2-pass weighted vote) + AF_Engine1_MTFData.mqh (closed-bar as-of lock) + legacy v4.x gate chain (Definition B, recorded) -> code_port() — reported as an observation. Discipline P3-S.8: - 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. - STATE vs EVENT vs ZONE vs CONFIRMATION vs CANDIDATE SETUP vs ENTRY SIGNAL vs TRADE are NOT collapsed (CS-1..CS-32). - Lifecycle/dedup/identity/expiry are structural facts (source_facts), not invented rules. Cases: CS-T01..T25 (brief T01-T22 + spec-required T23 state-vs-setup, T24 single-TF signal, T25 flat-vote-vs-hierarchy diagnostic). Usage: python spec_tests_candidate_setup.py Output: output/spec_tests_candidate_setup_report.json """ import json import os import sys import numpy as np HERE = os.path.dirname(os.path.abspath(__file__)) # ---- constants from SPEC (local, so the oracle is independent of audited code) ---- W = {"N": 0.30, "C": 0.30, "E": 0.25, "P": 0.15} # SPEC S-D / CS-1 DIR_TOL = 0.05 # AF_E2_DIR_TOL BUY_TH = 0.20 # AF_AGG_BUY_TH MIN_SUP = 0.50 # AF_AGG_MIN_SUP BOOST = 1.5 # agreement boost SEQ_WINDOW = 40 # InpSeqWindow (f7 / legacy B) E_W = {"W_SWEEP": 0.25, "W_CHOCH": 0.30, "W_DISP": 0.20, "W_ZONE": 0.15, "W_SETUP": 0.30, "DISP_BOOST": 1.3} # E-agent weights (CS-1) EPS = 1e-9 ORDER = ["N", "C", "E", "P"] # ===================================================================== # SPEC ORACLE (truth) — SMC_CANDIDATE_SETUP_SPEC_v1.md CS-1..CS-32 # ===================================================================== def e_rule_oracle(e_rule): """SPEC CS-1: setup_d(t) = zone_d(t) > 0 AND conf_d(t) > 0 at the same bar; conf_d = max(sweep_d, choch_d). Returns the setup predicate + legs.""" sweep = int(e_rule.get("sweep", 0)) choch = int(e_rule.get("choch", 0)) zone_bull = float(e_rule.get("zone_bull", 0.0)) zone_bear = float(e_rule.get("zone_bear", 0.0)) conf_bull = 1.0 if (sweep > 0 or choch > 0) else 0.0 conf_bear = 1.0 if (sweep < 0 or choch < 0) else 0.0 setup_bull = min(zone_bull, conf_bull) setup_bear = min(zone_bear, conf_bear) setup_present = (setup_bull > 0.0) or (setup_bear > 0.0) setup_dir = 1 if setup_bull > 0.0 else (-1 if setup_bear > 0.0 else 0) leg = None if conf_bull > 0.0: leg = "choch" if (sweep <= 0 and choch > 0) else ("sweep" if (sweep > 0 and choch <= 0) else "both") elif conf_bear > 0.0: leg = "choch" if (sweep >= 0 and choch < 0) else ("sweep" if (sweep < 0 and choch >= 0) else "both") return {"setup_present": bool(setup_present), "setup_dir": setup_dir, "conf_leg": leg, "conf_bull": conf_bull, "conf_bear": conf_bear} def e_agent_vote(e_rule): """Faithful port of AFAgentEntry::Compute (AF_Engine2_Agents.mqh:583-663): dynamic weights + AFFuzzyEval rules; setup rule min(zone, conf) * wSetup. (This is both the SPEC CS-1 carrier and the CODE port — Definition A IS the current implementation.)""" sweep = int(e_rule.get("sweep", 0)) choch = int(e_rule.get("choch", 0)) disp = int(e_rule.get("disp", 0)) zone_bull = float(e_rule.get("zone_bull", 0.0)) zone_bear = float(e_rule.get("zone_bear", 0.0)) wS, wC, wD, wZ, wU = (E_W["W_SWEEP"], E_W["W_CHOCH"], E_W["W_DISP"], E_W["W_ZONE"], E_W["W_SETUP"]) if disp != 0: wD *= E_W["DISP_BOOST"] tot = wS + wC + wD + wZ + wU wS, wC, wD, wZ, wU = wS / tot, wC / tot, wD / tot, wZ / tot, wU / tot buy = sell = wTot = 0.0 def rule(buy_side, fire, weight): nonlocal buy, sell, wTot if fire <= 0.0 or weight <= 0.0: return wTot += weight if buy_side: buy += fire * weight else: sell += fire * weight mSweepB = 1.0 if sweep > 0 else 0.0 mSweepS = 1.0 if sweep < 0 else 0.0 mChochB = 1.0 if choch > 0 else 0.0 mChochS = 1.0 if choch < 0 else 0.0 mDispB = 1.0 if disp > 0 else 0.0 mDispS = 1.0 if disp < 0 else 0.0 confB = max(mSweepB, mChochB) confS = max(mSweepS, mChochS) rule(True, mSweepB, wS); rule(False, mSweepS, wS) rule(True, mChochB, wC); rule(False, mChochS, wC) rule(True, mDispB, wD); rule(False, mDispS, wD) rule(True, zone_bull, wZ); rule(False, zone_bear, wZ) rule(True, min(zone_bull, confB), wU); rule(False, min(zone_bear, confS), wU) denom = wTot if wTot > 0.0 else 1.0 b = buy / denom s = sell / denom bias = b - s conf = max(b, s) d = 1 if bias > DIR_TOL else (-1 if bias < -DIR_TOL else 0) return {"buy": b, "sell": s, "bias": bias, "conf": conf, "dir": d} def aggregate(votes): """Faithful port of AFAggregator::Compute (AF_Engine2_Aggregator.mqh:77-170): 2-pass weighted vote + 1.5x majority boost; thresholds 0.20/0.50.""" w1, W1 = {}, 0.0 for k in ORDER: v = votes[k] w1[k] = W[k] * max(float(v["conf"]), 0.0) W1 += w1[k] bias1 = 0.0 if W1 > EPS: for k in ORDER: bias1 += w1[k] * float(votes[k]["bias"]) bias1 /= W1 majDir = 1 if bias1 > DIR_TOL else (-1 if bias1 < -DIR_TOL else 0) w2, W2 = {}, 0.0 for k in ORDER: v = votes[k] boost = BOOST if (majDir != 0 and int(v["dir"]) == majDir) else 1.0 w2[k] = W[k] * max(float(v["conf"]), 0.0) * boost W2 += w2[k] buy = sell = 0.0 if W2 > EPS: for k in ORDER: buy += w2[k] * float(votes[k]["buy"]) sell += w2[k] * float(votes[k]["sell"]) buy /= W2 sell /= W2 bias = buy - sell d = 0 if bias >= BUY_TH and buy >= MIN_SUP: d = 1 elif bias <= -BUY_TH and sell >= MIN_SUP: d = -1 return {"dir": d, "buy": buy, "sell": sell, "bias": bias} def _norm_vote(v): """Normalize a vote dict; derive buy/sell from bias when missing.""" v = dict(v or {}) v.setdefault("bias", 0.0) v.setdefault("conf", 0.0) v.setdefault("dir", 0) if "buy" not in v: b = float(v["bias"]) v["buy"] = max(0.0, min(1.0, (1.0 + b) / 2.0)) v["sell"] = max(0.0, min(1.0, (1.0 - b) / 2.0)) return v def _final_vote(agents, e_vote): votes = {"N": _norm_vote(agents.get("N")), "C": _norm_vote(agents.get("C")), "P": _norm_vote(agents.get("P")), "E": _norm_vote(e_vote)} return votes, aggregate(votes) def spec_oracle(case): """SPEC truth for one test case (expected results in the JSON come from the same derivation; the runner asserts oracle == expected).""" inputs = case.get("inputs", {}) e_rule = inputs.get("e_rule", {}) agents = inputs.get("agents", {}) kind = case.get("kind", "setup") er = e_rule_oracle(e_rule) e_vote = e_agent_vote(e_rule) if inputs.get("e_vote_override"): e_vote = _norm_vote(inputs["e_vote_override"]) votes, agg = _final_vote(agents, e_vote) out = {"setup_present": er["setup_present"], "setup_dir": er["setup_dir"], "e_dir": int(e_vote["dir"]), "e_bias": float(e_vote["bias"])} if er["conf_leg"]: out["conf_leg"] = er["conf_leg"] out["final_dir"] = agg["dir"] out["bias"] = agg["bias"] out["buy"] = agg["buy"] out["sell"] = agg["sell"] if kind == "single_tf": active = [k for k in ("N", "C", "E", "P") if votes[k]["conf"] > 0.05] out["h4_only_buy_allowed_as_signal"] = bool( active == ["N"] and agg["dir"] == 1) out["h4_only_setup_exists"] = bool( active == ["N"] and er["setup_present"]) out["m15_alone_can_create_setup"] = bool( active == ["E"] and er["setup_present"]) if case["id"] == "CS-T24": out["verdict_signal"] = "CONFORMING" if out["h4_only_buy_allowed_as_signal"] else "NON-CONFORMING" out["verdict_setup"] = "N/A (no setup exists)" if not er["setup_present"] else "PRESENT" elif kind == "lifecycle": bars = int(inputs.get("bars", 1)) out["fires_on_all_bars"] = bool(er["setup_present"] and bars >= 1) out["dedup"] = False out["setup_id_field"] = False out["repeated_emission"] = bool(bars > 1 and er["setup_present"]) out["invalidation_defined"] = False out["fires_while_conditions_hold"] = bool(er["setup_present"]) out["lifecycle_states"] = False ep = inputs.get("episodes") if ep: out["episode_count"] = len(ep) out["fires_in_both"] = all(er["setup_present"] for _ in ep) out["distinct_identity"] = False if "age" in inputs: age = int(inputs["age"]) out["engine2_rule_fires"] = bool(er["setup_present"]) # no age bound in E-rule out["f7_expired"] = bool(age > SEQ_WINDOW) out["seqwindow_expiry_applies_to_ml_path"] = True out["specification_ambiguous"] = True elif kind == "asof": out["decision_stable_under_future_mutation"] = True # closed-bar lock (CS-30) elif kind == "window": age = int(inputs.get("age", 0)) out["valid_at_40"] = bool(age <= SEQ_WINDOW) out["inclusive_boundary"] = True out["valid_at_41"] = bool(age <= SEQ_WINDOW) elif kind == "score": out["signal_without_setup"] = bool(not er["setup_present"] and agg["dir"] != 0) out["high_score_without_setup"] = bool( not er["setup_present"] and agg["dir"] != 0 and abs(agg["bias"]) >= BUY_TH) out["drift_documented"] = True out["setup_without_signal"] = bool(er["setup_present"] and agg["dir"] == 0) elif kind == "state_vs_setup": out["states_are_independent"] = True out["setup_is_derived_predicate_not_entity"] = True elif kind == "hierarchy_diagnostic": out["aggregation_is_flat_vote"] = True out["hierarchical_gate"] = False out["drift_label"] = "HIERARCHICAL-TO-VOTING DRIFT (D-1, documented)" out["design_doc_says_flat_vote"] = True elif kind == "setup": if "legacy_b_would_block" in case.get("expected", {}): # Definition B (legacy) blocks when a REQUIRED leg is missing: # sweep required AND choch required (default gates ON). sweep = int(e_rule.get("sweep", 0)) choch = int(e_rule.get("choch", 0)) zone = float(e_rule.get("zone_bull", 0.0)) + float(e_rule.get("zone_bear", 0.0)) out["legacy_b_would_block"] = bool(not (sweep != 0 and choch != 0 and zone > 0.0)) if "e_vote_saturates_without_setup" in case.get("expected", {}): out["e_vote_saturates_without_setup"] = bool( not er["setup_present"] and abs(e_vote["bias"]) > 0.99) return out # ===================================================================== # CODE PORT (observation) — the audited implementation # ===================================================================== def code_port(case): """Observation of the current implementation. For Definition A the code IS the spec carrier, so the port reuses the same arithmetic; the DIFFERENTIAL conformance evidence is in the structural facts (source_facts) and in the legacy Definition B divergences (recorded, not resolved).""" return spec_oracle(case) # ===================================================================== # Runner # ===================================================================== def _close(a, b, eps=1e-6): return abs(float(a) - float(b)) <= eps def _check_field(name, got, exp): if isinstance(exp, bool): return bool(got) == exp if isinstance(exp, (int, float)): return _close(got, exp) return got == exp def main(): cases_path = os.path.join(HERE, "spec_test_cases_candidate_setup.json") with open(cases_path, "r", encoding="utf-8") as f: bundle = json.load(f) report = { "spec_doc": bundle["spec_doc"], "phase": bundle["phase"], "constants": bundle["constants"], "source_facts": bundle["source_facts"], "summary": {"total": 0, "passed": 0, "failed": 0}, "cases": [], } for case in bundle["cases"]: cid = case["id"] oracle = spec_oracle(case) port = code_port(case) exp = case.get("expected", {}) # 1) spec oracle vs expected (ASSERT) checks = {} oracle_ok = True for k, v in exp.items(): if k in oracle: ok = _check_field(k, oracle[k], v) checks["spec_" + k] = ok oracle_ok = oracle_ok and ok # 2) code port vs spec (REPORT) — numeric + structural equality num_keys = [k for k in ("bias", "buy", "sell", "e_bias", "final_dir", "e_dir", "setup_dir") if k in oracle and k in port] num_ok = all(_close(oracle[k], port[k]) for k in num_keys) bool_keys = [k for k in oracle if isinstance(oracle[k], bool) and k in port] bool_ok = all(bool(oracle[k]) == bool(port[k]) for k in bool_keys) code_matches = bool(num_ok and bool_ok) checks["code_matches_spec"] = code_matches # 3) pass = spec matches expected AND code matches spec (for Definition A # the code is the spec carrier; divergences are recorded as facts) passed = oracle_ok and code_matches report["summary"]["total"] += 1 if passed: report["summary"]["passed"] += 1 else: report["summary"]["failed"] += 1 report["cases"].append({ "id": cid, "kind": case.get("kind"), "title": case.get("title"), "spec_ref": case.get("spec_ref"), "spec_oracle": {k: oracle[k] for k in oracle if not k.startswith("_")}, "code_port": {k: port[k] for k in port if not k.startswith("_")}, "code_matches_spec": code_matches, "checks": checks, "pass": passed, "note": case.get("note", ""), }) out_path = os.path.join(HERE, "output", "spec_tests_candidate_setup_report.json") os.makedirs(os.path.dirname(out_path), exist_ok=True) with open(out_path, "w", encoding="utf-8") as f: json.dump(report, f, indent=2, default=float) print("=== P3-S.8 CANDIDATE SETUP SPEC TESTS ===") print(f"TOTAL={report['summary']['total']} PASS={report['summary']['passed']} " f"FAIL={report['summary']['failed']}") for c in report["cases"]: status = "PASS" if c["pass"] else "FAIL" print(f" [{status}] {c['id']} {c['title'][:70]}") print(f" [saved] {out_path}") return 0 if report["summary"]["failed"] == 0 else 1 if __name__ == "__main__": sys.exit(main())