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_mtf_alignment.py

395 lines
16 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.7 SPEC TESTS — MTF ALIGNMENT (synthetic, from PROJECT SEMANTIC SPECIFICATION v1).
Truth : docs/SMC_MTF_ALIGNMENT_SPEC_v1.md -> spec_oracle() in this file
(expected results in spec_test_cases_mtf_alignment.json are derived
from SPEC S-N..S-ST, NOT from the audited code).
Code under : AF_Engine2_Agents.mqh (4 agents) + AF_Engine2_Aggregator.mqh
audit (AFAggregator::Compute) + AF_Engine1_MTFData.mqh (as-of cache)
-> code_port() — reported as an observation.
Discipline P3-S.7:
- 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.
- hierarchy diagnostic : REPORT (flat vs hierarchical) — MTF-T15.
Cases: MTF-T01..T22 (brief T01-T18 + spec-required T19-T22).
Usage: python spec_tests_mtf_alignment.py
Output: output/spec_tests_mtf_alignment_report.json
"""
import datetime as dt
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 (AF_AGG_W_*)
DIR_TOL = 0.05 # SPEC S-D (AF_E2_DIR_TOL)
BUY_TH = 0.20 # SPEC S-D (AF_AGG_BUY_TH)
MIN_SUP = 0.50 # SPEC S-D (AF_AGG_MIN_SUP)
BOOST = 1.5 # SPEC S-D (agreement boost)
MIN_BARS = 80 # SPEC S-H (AF_E2_MIN_BARS)
EPS = 1e-12
ORDER = ["N", "C", "E", "P"]
# =====================================================================
# SPEC ORACLE (truth) — derived from SMC_MTF_ALIGNMENT_SPEC_v1.md S-D / S-T
# =====================================================================
def spec_aggregate(votes):
"""SPEC S-D: 2-pass weighted vote (flat, same-level).
votes: dict N/C/E/P -> {bias, conf, dir}.
Returns dict(dir, buy, sell, bias, score, majDir).
"""
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,
"score": abs(bias), "majDir": majDir}
def _complete_votes(votes):
"""Fill buy/sell from bias when missing (buy=(1+bias)/2, sell=(1-bias)/2)."""
out = {}
for k in ORDER:
v = dict(votes.get(k) or {"bias": 0.0, "conf": 0.0, "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))
out[k] = v
return out
def _epoch(s):
return int(dt.datetime.strptime(s, "%Y-%m-%dT%H:%M:%SZ").replace(
tzinfo=dt.timezone.utc).timestamp())
def spec_asof(htf_close_epoch, decision_epoch, htf_period):
"""SPEC S-T: the newest closed HTF bar with close_time <= t is visible iff
its close_time <= decision time. Bar closes exactly at htf_close_epoch.
A decision at t sees the bar iff htf_close_epoch <= t."""
return htf_close_epoch <= decision_epoch
# =====================================================================
# CODE PORT (implementation under audit) — AFAggregator::Compute faithful
# =====================================================================
def code_aggregate(votes):
"""Faithful port of AFAggregator::Compute (AF_Engine2_Aggregator.mqh:77-170).
Identical arithmetic: base weights, pass-1 majority, pass-2 1.5x boost,
thresholds 0.20 / 0.50, dir = +1/-1/0.
"""
ag = [votes[k] for k in ORDER]
baseW = [W[k] for k in ORDER]
w1 = [baseW[i] * max(ag[i]["conf"], 0.0) for i in range(4)]
W1 = sum(w1)
bias1 = 0.0
if W1 > 0.0:
bias1 = sum(w1[i] * ag[i]["bias"] for i in range(4)) / W1
majDir = 1 if bias1 > DIR_TOL else (-1 if bias1 < -DIR_TOL else 0)
w2 = []
for i in range(4):
boost = 1.5 if (majDir != 0 and ag[i]["dir"] == majDir) else 1.0
w2.append(baseW[i] * max(ag[i]["conf"], 0.0) * boost)
W2 = sum(w2)
buy = sell = 0.0
if W2 > 0.0:
buy = sum(w2[i] * ag[i]["buy"] for i in range(4)) / W2
sell = sum(w2[i] * ag[i]["sell"] for i in range(4)) / 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, "majDir": majDir}
# =====================================================================
# STRUCTURAL ASSERTIONS (source-scan based, spec S-R / S-ST / S-A / S-I)
# =====================================================================
def scan_source():
"""Read-only source scan of the audited MQL5 files. Returns facts used by
the structural cases (independence, statelessness, flat aggregation,
tf-identity by convention, closed-bar lock)."""
base = os.path.join(HERE, "..", "..", "..")
agents = os.path.join(base, "MQL5", "Include", "AlgoForge",
"AF_Engine2_Agents.mqh")
agg = os.path.join(base, "MQL5", "Include", "AlgoForge",
"AF_Engine2_Aggregator.mqh")
e1 = os.path.join(base, "MQL5", "Include", "AlgoForge",
"AF_Engine1_MTFData.mqh")
defines = os.path.join(base, "MQL5", "Include", "AlgoForge",
"AF_Defines.mqh")
def read(p):
with open(p, encoding="utf-8", errors="replace") as f:
return f.read()
ta, tgg, te1, tdf = read(agents), read(agg), read(e1), read(defines)
# E agent (AFAgentEntry::Compute) reads only slot E: no second slot read.
e_independent = ("AFAgentEntry::Compute" in ta) and ("AFEngine1MTF &e1,int slot" in ta)
# stateless: agent classes carry no persistent members (no m_ fields)
stateless = ("class AFAgentNarrative" in ta and "class AFAgentContext" in ta
and "class AFAgentEntry" in ta and "class AFAgentPriceAction" in ta
and "int m_state" not in ta)
# flat aggregation: aggregator uses the four base weights + boost, no gate
flat = ("AF_AGG_W_N" in tgg) and ("boost" in tgg or "1.5" in tgg) and (
"BLOCK" not in tgg.upper())
# closed-bar lock in Engine 1
closed_bar = ("IsBarClosed" in te1) and ("skip" in te1)
# hardcoded TF macros
macros = all(("AF_E2_TF_S%d" % i) in tdf for i in (1, 2, 3, 4))
# runtime TF identity by convention: Register is idempotent per (symbol,tf)
by_convention = ("m_find(ENUM_TIMEFRAMES tf" in te1) and ("TfOf" in te1)
return {
"e_agent_single_slot": e_independent,
"agents_stateless": stateless,
"aggregator_flat_vote": flat,
"engine1_closed_bar_lock": closed_bar,
"hardcoded_tf_macros": macros,
"tf_identity_by_convention": by_convention,
}
# =====================================================================
# CASE RUNNERS
# =====================================================================
def run_vote(case, facts):
votes = _complete_votes(case["votes"])
spec = spec_aggregate(votes)
code = code_aggregate(votes)
exp = case["expected"]
exp_dir = int(exp.get("dir", 0))
checks = {
"spec_dir_matches_expected": bool(spec["dir"] == exp_dir),
"code_matches_spec": bool(code["dir"] == spec["dir"]),
}
if "bias_min" in exp:
checks["spec_bias_min"] = bool(spec["bias"] >= float(exp["bias_min"]))
if "conflict_blocked" in exp:
checks["no_conflict_block"] = bool(exp["conflict_blocked"] is False)
if "p_contributes" in exp:
checks["p_contributes_vote"] = bool(votes["P"]["conf"] > 0.0)
if "rejection_policy" in exp:
checks["no_rejection_policy"] = bool(exp["rejection_policy"] is False)
return {
"id": case["id"], "kind": case["kind"], "title": case["title"],
"spec_ref": case.get("spec_ref"),
"spec_oracle": {"dir": spec["dir"], "bias": round(float(spec["bias"]), 6),
"buy": round(float(spec["buy"]), 6),
"sell": round(float(spec["sell"]), 6)},
"code_port": {"dir": code["dir"], "bias": round(float(code["bias"]), 6)},
"code_matches_spec": bool(code["dir"] == spec["dir"]),
"checks": checks,
"pass": all(checks.values()),
"note": case.get("title"),
}
def run_asof(case, _facts):
tfs = case["tfs"]
htf_close = _epoch(case["htf_close"])
htf_period = tfs.get("H4", 14400)
exp = case["expected"]
checks = {}
if "future_visible" in exp:
before = _epoch(case["decision_before"])
checks["no_future_visibility"] = bool(
exp["future_visible"] is False and
not spec_asof(htf_close, before, htf_period))
else:
if "decision_before" in case:
before = _epoch(case["decision_before"])
checks["visible_before"] = bool(
spec_asof(htf_close, before, htf_period) == exp["visible_before"])
if "decision_at" in case:
at = _epoch(case["decision_at"])
checks["visible_at"] = bool(
spec_asof(htf_close, at, htf_period) == exp["visible_at"])
if "decision_after" in case:
after = _epoch(case["decision_after"])
checks["visible_after"] = bool(
spec_asof(htf_close, after, htf_period) == exp["visible_after"])
return {
"id": case["id"], "kind": case["kind"], "title": case["title"],
"spec_ref": case.get("spec_ref"),
"spec_oracle": {"asof_rule": "htf_close_time <= decision_time (S-T)"},
"code_port": {"engine1_closed_bar_lock": _facts["engine1_closed_bar_lock"]},
"code_matches_spec": _facts["engine1_closed_bar_lock"],
"checks": checks,
"pass": all(checks.values()) and _facts["engine1_closed_bar_lock"],
"note": case.get("title"),
}
def run_structural(case, facts):
kind = case["kind"]
exp = case["expected"]
checks = {}
if kind == "stale":
checks["no_stale_consumption"] = bool(exp["stale_consumption"] is False)
checks["agents_stateless"] = facts["agents_stateless"]
elif kind == "independence":
checks["e_agent_single_slot"] = facts["e_agent_single_slot"]
elif kind == "repeat":
checks["stateless_no_repeated_events"] = facts["agents_stateless"]
elif kind == "tf_identity":
checks["detected_at_runtime"] = bool(exp["detected_at_runtime"] is False)
checks["documented"] = bool(exp["documented"] is True)
checks["tf_identity_by_convention"] = facts["tf_identity_by_convention"]
elif kind == "hierarchy_diagnostic":
checks["flat"] = facts["aggregator_flat_vote"]
checks["hierarchical_gate"] = bool(exp["hierarchical_gate"] is False)
elif kind == "history":
checks["insufficient_history_neutral"] = bool(
int(case["bars_available"]) < MIN_BARS)
checks["agents_neutral_wait"] = bool(int(exp["dir"]) == 0)
else:
checks["unknown_kind"] = False
ok = all(checks.values())
return {
"id": case["id"], "kind": case["kind"], "title": case["title"],
"spec_ref": case.get("spec_ref"),
"spec_oracle": {"structural_rule": case.get("assert")},
"code_port": {k: bool(v) for k, v in facts.items()},
"code_matches_spec": ok,
"checks": checks,
"pass": ok,
"note": case.get("title"),
}
def run_symmetry(case, _facts):
votes = _complete_votes(case["votes"])
spec = spec_aggregate(votes)
mirror = {k: {"bias": -float(v["bias"]), "conf": float(v["conf"]),
"dir": -int(v["dir"]) if v["dir"] != 0 else 0}
for k, v in votes.items()}
mirror = _complete_votes(mirror)
spec_m = spec_aggregate(mirror)
code_m = code_aggregate(mirror)
exp_dir = int(case["expected"]["mirror_dir"])
checks = {
"mirror_flips_dir": bool(spec_m["dir"] == exp_dir),
"mirror_symmetry": bool(spec_m["dir"] == -spec["dir"]),
"code_matches_spec": bool(code_m["dir"] == spec_m["dir"]),
}
return {
"id": case["id"], "kind": case["kind"], "title": case["title"],
"spec_ref": case.get("spec_ref"),
"spec_oracle": {"dir": spec["dir"], "mirror_dir": spec_m["dir"]},
"code_port": {"mirror_dir": code_m["dir"]},
"code_matches_spec": bool(code_m["dir"] == spec_m["dir"]),
"checks": checks,
"pass": all(checks.values()),
"note": case.get("title"),
}
def run_case(case, facts):
kind = case["kind"]
if kind == "vote":
return run_vote(case, facts)
if kind == "asof":
return run_asof(case, facts)
if kind == "symmetry":
return run_symmetry(case, facts)
return run_structural(case, facts)
# =====================================================================
# MAIN
# =====================================================================
def main():
cases_path = os.path.join(HERE, "spec_test_cases_mtf_alignment.json")
cases = json.load(open(cases_path, encoding="utf-8"))["cases"]
facts = scan_source()
results = [run_case(c, facts) for c in cases]
n_pass = sum(1 for r in results if r["pass"])
n_total = len(results)
print(f"=== P3-S.7 SPEC TESTS — MTF ALIGNMENT ({n_pass}/{n_total} PASS) ===")
print(f"{'ID':<9} {'PASS':<6} {'kind':<20} {'dir s/c':<10} notes")
for r in results:
s = r.get("spec_oracle", {})
c = r.get("code_port", {})
sd = s.get("dir", "-")
cd = c.get("dir", "-")
print(f"{r['id']:<9} {str(r['pass']):<6} {r['kind']:<20} "
f"{str(sd)+'/'+str(cd):<10} {r['title'][:44]}")
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 (MTF Alignment).")
print("code_port == spec on all vote/symmetry/asof cases (the implementation "
"conforms to the project's flat-vote model).")
print("Structural diagnostics (facts):")
for k, v in facts.items():
print(f" {k:<32}: {v}")
report = {
"spec_doc": "docs/SMC_MTF_ALIGNMENT_SPEC_v1.md",
"phase": "P3-S.7",
"generated_utc": dt.datetime.now(dt.timezone.utc).isoformat(),
"constants": {"W": W, "DIR_TOL": DIR_TOL, "BUY_TH": BUY_TH,
"MIN_SUP": MIN_SUP, "BOOST": BOOST, "MIN_BARS": MIN_BARS},
"summary": {"total": n_total, "passed": n_pass,
"failed": n_total - n_pass},
"source_facts": {k: bool(v) for k, v in facts.items()},
"cases": results,
}
outdir = os.path.join(HERE, "output")
os.makedirs(outdir, exist_ok=True)
out_path = os.path.join(outdir, "spec_tests_mtf_alignment_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()
Copy permalink