379 lines
15 KiB
Python
379 lines
15 KiB
Python
|
# -*- 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())
|