395 lines
16 KiB
Python
395 lines
16 KiB
Python
|
# -*- 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()
|