Aliwkarimi98/SniperGold_ML
Watch
1
0
Fork
You've already forked SniperGold_ML
 1
forked from chiki2bum2/SniperGold_ML
Code Pull requests Activity
SniperGold_ML/ml/p3/smc_semantic/spec_tests_candidate_setup.py

379 lines
15 KiB
Python
RawPermalinkNormal ViewHistory

research: checkpoint P3-S.7/P3-S.8/P3-S.9 evidence chain (MTF alignment, candidate setup, architecture review)
2026-08-22 21:53:38 +07:00
# -*- 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())
Copy permalink