324 lines
13 KiB
Python
324 lines
13 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
"""P3-S.3 SPEC TESTS — CHOCH / MSS (synthetic, dari PROJECT SEMANTIC SPECIFICATION v1).
|
||
|
|
|
||
|
|
Truth : docs/SMC_CHOCH_MSS_SPEC_v1.md -> spec_oracle() di file ini
|
||
|
|
(expected result di spec_test_cases_choch_mss.json diturunkan
|
||
|
|
dari SPEC S-3..S-10, BUKAN dari kode).
|
||
|
|
Code under : ProcessStructure (EA AlgoForge_Backtest_Baseline.mq5 + v4.4/v4.5)
|
||
|
|
audit + emission f8/f9 -> code_port() — hasilnya DILAPORKAN sebagai
|
||
|
|
observasi konformansi, BUKAN definisi truth.
|
||
|
|
|
||
|
|
Disiplin P3-S.3:
|
||
|
|
- spec oracle vs expected : ASSERT (spec = truth)
|
||
|
|
- code port vs spec : REPORT (differential conformance observation)
|
||
|
|
- tidak ada AUC/PF/backtest/human annotation/ML di file ini.
|
||
|
|
|
||
|
|
Kasus: CH-T01..T20 (wajib T01-T15 + T16/T17 + tambahan gate/supersession).
|
||
|
|
|
||
|
|
Usage: python spec_tests_choch_mss.py
|
||
|
|
Output: output/spec_tests_choch_mss_report.json
|
||
|
|
"""
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
|
||
|
|
import numpy as np
|
||
|
|
|
||
|
|
HERE = os.path.dirname(os.path.abspath(__file__))
|
||
|
|
|
||
|
|
# ---- konstanta dari SPEC (lokal agar oracle independen dari kode diaudit) ----
|
||
|
|
INTERNAL_LEN = 5 # SPEC S-1/S-2: fractal 5/5
|
||
|
|
SEQ_WINDOW = 40 # SPEC S-9/S-10: validitas rantai (InpSeqWindow v4.4)
|
||
|
|
EPS = 1e-9
|
||
|
|
|
||
|
|
|
||
|
|
# =====================================================================
|
||
|
|
# SPEC ORACLE (truth) — diturunkan langsung dari SMC_CHOCH_MSS_SPEC_v1.md
|
||
|
|
# =====================================================================
|
||
|
|
def swing_at_of(swing_breaks, i):
|
||
|
|
"""SPEC S-7: SwingTrendAt(i) = trend swing break terakhir <= i (default 0)."""
|
||
|
|
v = 0
|
||
|
|
for (b, val) in swing_breaks:
|
||
|
|
if b <= i:
|
||
|
|
v = val
|
||
|
|
else:
|
||
|
|
break
|
||
|
|
return v
|
||
|
|
|
||
|
|
|
||
|
|
def spec_oracle(o, h, l, c, refs, swing_breaks, decision_bar, sweep=None,
|
||
|
|
internal_len=INTERNAL_LEN, seq_window=SEQ_WINDOW):
|
||
|
|
"""Evaluator SPEC (S-3..S-10).
|
||
|
|
|
||
|
|
refs : [(p, level, is_high)] — internal pivot GIVEN (valid per S-2);
|
||
|
|
menjadi target break hanya setelah konfirmasi p+len <= b (S-8).
|
||
|
|
swing_breaks : [(bar, trend)] — timeline swing utk gate S-7.
|
||
|
|
Return : (events, f8_at_decision, f8_series, choch_bar_last, f9_chain)
|
||
|
|
events = [(b, dir, level, p)] kronologis.
|
||
|
|
"""
|
||
|
|
n = decision_bar + 1
|
||
|
|
events = []
|
||
|
|
up_target, up_bar = float("inf"), -1
|
||
|
|
dn_target, dn_bar = -float("inf"), -1
|
||
|
|
trend = 0
|
||
|
|
choch_dir, choch_bar = 0, -1
|
||
|
|
confirm = {}
|
||
|
|
for (p, lvl, ih) in refs: # konfirmasi pada bar p+len
|
||
|
|
confirm.setdefault(p + internal_len, []).append((p, lvl, ih))
|
||
|
|
f8 = np.zeros(n, dtype=int)
|
||
|
|
for i in range(n):
|
||
|
|
for (p, lvl, ih) in confirm.get(i, []): # target = pivot terakhir terkonfirmasi
|
||
|
|
if ih:
|
||
|
|
up_target, up_bar = lvl, p
|
||
|
|
else:
|
||
|
|
dn_target, dn_bar = lvl, p
|
||
|
|
# bullish break (S-3/S-6: close strict > level)
|
||
|
|
if up_bar >= 0 and c[i] > up_target:
|
||
|
|
choch = (trend < 0) # S-7: prior internal trend bearish
|
||
|
|
allow = swing_at_of(swing_breaks, i) >= 0 # S-7/A-4: gate swing tidak turun
|
||
|
|
if allow and choch: # S-3: event reversal
|
||
|
|
events.append((i, 1, up_target, up_bar))
|
||
|
|
choch_dir, choch_bar = 1, i
|
||
|
|
trend = 1 # trend internal SELALU berbalik
|
||
|
|
up_target, up_bar = float("inf"), -1
|
||
|
|
# bearish break
|
||
|
|
if dn_bar >= 0 and c[i] < dn_target:
|
||
|
|
choch = (trend > 0)
|
||
|
|
allow = swing_at_of(swing_breaks, i) <= 0
|
||
|
|
if allow and choch:
|
||
|
|
events.append((i, -1, dn_target, dn_bar))
|
||
|
|
choch_dir, choch_bar = -1, i
|
||
|
|
trend = -1
|
||
|
|
dn_target, dn_bar = -float("inf"), -1
|
||
|
|
f8[i] = choch_dir # S-9: state last CHoCH
|
||
|
|
# S-9/S-10/S-12: f9 rantai = CHoCH SETELAH sweep & searah & age <= SEQ_WINDOW
|
||
|
|
# (tanpa sweep -> 0; rantai tidak ada)
|
||
|
|
f9_chain = 0
|
||
|
|
if events and sweep:
|
||
|
|
sb, sd = sweep[-1]
|
||
|
|
b, d, lvl, p = events[-1]
|
||
|
|
if b >= sb and d == sd and (decision_bar - b) <= seq_window:
|
||
|
|
f9_chain = d
|
||
|
|
return events, int(f8[decision_bar]), f8, choch_bar, f9_chain
|
||
|
|
|
||
|
|
|
||
|
|
# =====================================================================
|
||
|
|
# CODE PORT (implementasi yang diaudit) — semantik EA ProcessStructure
|
||
|
|
# =====================================================================
|
||
|
|
def code_port(o, h, l, c, refs, swing_breaks, sweep, decision_bar,
|
||
|
|
internal_len=INTERNAL_LEN, seq_window=SEQ_WINDOW):
|
||
|
|
"""Port setia EA ProcessStructure(internal=true) + emission f8/f9.
|
||
|
|
|
||
|
|
refs : pivot internal g_ip (given, sama dgn oracle).
|
||
|
|
sweep : [(bar, dir)] — g_swpBar/g_swpDir legacy (state persist).
|
||
|
|
Return: (events, f8_code, f9_code, choch_bar_last)
|
||
|
|
Perbedaan vs spec_oracle:
|
||
|
|
- f9 TANPA expiry (chochOK = dir!=0 && bar>=swpBar && dir==swpDir) — EA:487
|
||
|
|
- f8 = last CHoCH (state, tanpa expiry) — EA:500
|
||
|
|
"""
|
||
|
|
n = decision_bar + 1
|
||
|
|
events = []
|
||
|
|
up_target, up_bar = float("inf"), -1
|
||
|
|
dn_target, dn_bar = -float("inf"), -1
|
||
|
|
trend = 0
|
||
|
|
choch_dir, choch_bar = 0, -1
|
||
|
|
confirm = {}
|
||
|
|
for (p, lvl, ih) in refs:
|
||
|
|
confirm.setdefault(p + internal_len, []).append((p, lvl, ih))
|
||
|
|
for i in range(n):
|
||
|
|
for (p, lvl, ih) in confirm.get(i, []):
|
||
|
|
if ih:
|
||
|
|
up_target, up_bar = lvl, p
|
||
|
|
else:
|
||
|
|
dn_target, dn_bar = lvl, p
|
||
|
|
if up_bar >= 0 and c[i] > up_target:
|
||
|
|
choch = (trend < 0)
|
||
|
|
allow = swing_at_of(swing_breaks, i) >= 0
|
||
|
|
if allow and choch:
|
||
|
|
events.append((i, 1, up_target, up_bar))
|
||
|
|
choch_dir, choch_bar = 1, i
|
||
|
|
trend = 1
|
||
|
|
up_target, up_bar = float("inf"), -1
|
||
|
|
if dn_bar >= 0 and c[i] < dn_target:
|
||
|
|
choch = (trend > 0)
|
||
|
|
allow = swing_at_of(swing_breaks, i) <= 0
|
||
|
|
if allow and choch:
|
||
|
|
events.append((i, -1, dn_target, dn_bar))
|
||
|
|
choch_dir, choch_bar = -1, i
|
||
|
|
trend = -1
|
||
|
|
dn_target, dn_bar = -float("inf"), -1
|
||
|
|
# emission (EA ComputeMLFeatures)
|
||
|
|
f8_code = choch_dir
|
||
|
|
f9_code = 0
|
||
|
|
if sweep:
|
||
|
|
sb, sd = sweep[-1]
|
||
|
|
f9_code = 1 if (choch_dir != 0 and choch_bar >= sb and choch_dir == sd) else 0
|
||
|
|
return events, f8_code, f9_code, choch_bar
|
||
|
|
|
||
|
|
|
||
|
|
# =====================================================================
|
||
|
|
# HELPERS
|
||
|
|
# =====================================================================
|
||
|
|
def materialize(case):
|
||
|
|
"""Baris OHLC -> arrays o,h,l,c ; dukungan bar_spec (neutral + patches list)."""
|
||
|
|
bs = case["bar_spec"]
|
||
|
|
bars = [list(bs["neutral"]) for _ in range(bs["count"])]
|
||
|
|
for (idx, ohlc) in bs["patches"]:
|
||
|
|
if idx < len(bars):
|
||
|
|
bars[idx] = list(ohlc)
|
||
|
|
arr = np.asarray(bars, dtype=float)
|
||
|
|
return arr[:, 0], arr[:, 1], arr[:, 2], arr[:, 3]
|
||
|
|
|
||
|
|
|
||
|
|
def apply_transform(o, h, l, c, refs, transform):
|
||
|
|
if transform is None:
|
||
|
|
return o, h, l, c, refs
|
||
|
|
kind = transform["kind"]
|
||
|
|
if kind == "scale":
|
||
|
|
k = transform["factor"]
|
||
|
|
return (o * k, h * k, l * k, c * k,
|
||
|
|
[(p, lvl * k, ih) for (p, lvl, ih) in refs])
|
||
|
|
if kind == "translate":
|
||
|
|
v = transform["value"]
|
||
|
|
return (o + v, h + v, l + v, c + v,
|
||
|
|
[(p, lvl + v, ih) for (p, lvl, ih) in refs])
|
||
|
|
raise ValueError("unknown transform " + kind)
|
||
|
|
|
||
|
|
|
||
|
|
def apply_mutations(o, h, l, c, mutations, neutral):
|
||
|
|
"""Mutasi bar (bisa melebihi panjang array -> perluas dgn bar netral)."""
|
||
|
|
o = o.copy(); h = h.copy(); l = l.copy(); c = c.copy()
|
||
|
|
for (idx, ohlc) in mutations:
|
||
|
|
while idx >= len(o):
|
||
|
|
o = np.append(o, neutral[0])
|
||
|
|
h = np.append(h, neutral[1])
|
||
|
|
l = np.append(l, neutral[2])
|
||
|
|
c = np.append(c, neutral[3])
|
||
|
|
o[idx], h[idx], l[idx], c[idx] = ohlc
|
||
|
|
return o, h, l, c
|
||
|
|
|
||
|
|
|
||
|
|
def events_equal(a, b):
|
||
|
|
if len(a) != len(b):
|
||
|
|
return False
|
||
|
|
for x, y in zip(a, b):
|
||
|
|
if x[0] != y[0] or x[1] != y[1] or abs(x[2] - y[2]) > EPS or x[3] != y[3]:
|
||
|
|
return False
|
||
|
|
return True
|
||
|
|
|
||
|
|
|
||
|
|
def run_case(case):
|
||
|
|
o, h, l, c = materialize(case)
|
||
|
|
refs = [tuple(r) for r in case["given_references"]]
|
||
|
|
swing_breaks = [tuple(s) for s in case.get("given_swing_breaks", [])]
|
||
|
|
sweep = [tuple(s) for s in case.get("given_sweep", [])]
|
||
|
|
db = case["decision_bar"]
|
||
|
|
transform = case.get("transform")
|
||
|
|
mutations = case.get("mutations")
|
||
|
|
neutral = case["bar_spec"]["neutral"]
|
||
|
|
|
||
|
|
# --- spec oracle (truth) pada data dasar ---
|
||
|
|
events, f8, f8_series, cb_last, f9_chain = spec_oracle(
|
||
|
|
o, h, l, c, refs, swing_breaks, db, sweep)
|
||
|
|
|
||
|
|
# --- invariant: transform (scale/translate) ---
|
||
|
|
o2, h2, l2, c2, refs2 = apply_transform(o, h, l, c, refs, transform)
|
||
|
|
events2, f8_2, _, _, _ = spec_oracle(o2, h2, l2, c2, refs2, swing_breaks, db)
|
||
|
|
|
||
|
|
# --- invariant: future mutation (T14 / S-8) ---
|
||
|
|
o3, h3, l3, c3 = apply_mutations(o, h, l, c, mutations or [], neutral)
|
||
|
|
events3, f8_3, _, _, _ = spec_oracle(o3, h3, l3, c3, refs, swing_breaks, db)
|
||
|
|
|
||
|
|
# --- code port (implementasi diaudit) ---
|
||
|
|
c_events, c_f8, c_f9, c_cb = code_port(o, h, l, c, refs, swing_breaks,
|
||
|
|
sweep, db)
|
||
|
|
|
||
|
|
exp = case["expected"]
|
||
|
|
checks = {}
|
||
|
|
checks["events"] = events_equal(events, [tuple(e) for e in exp["events"]])
|
||
|
|
checks["f8_at_decision"] = bool(f8 == int(exp["f8_at_decision"]))
|
||
|
|
if "f9_chain_at_decision" in exp:
|
||
|
|
checks["f9_chain_spec"] = bool(f9_chain == int(exp["f9_chain_at_decision"]))
|
||
|
|
if "code_f9_at_decision" in exp:
|
||
|
|
checks["code_f9_observed"] = bool(c_f9 == int(exp["code_f9_at_decision"]))
|
||
|
|
# invariance transform: struktural (b, dir, p) identik; level ikut transform
|
||
|
|
struct = lambda es: [(b, d, p) for (b, d, lv, p) in es]
|
||
|
|
checks["invariant_transform"] = bool(struct(events2) == struct(events) and
|
||
|
|
f8_2 == f8)
|
||
|
|
# invariance future mutation: event & f8 di decision_bar TIDAK berubah
|
||
|
|
checks["invariant_future_mutation"] = bool(events_equal(events3, events) and
|
||
|
|
f8_3 == f8)
|
||
|
|
|
||
|
|
# differential conformance: code vs spec
|
||
|
|
code_matches = {
|
||
|
|
"f8": bool(c_f8 == f8),
|
||
|
|
"events": events_equal(c_events, events),
|
||
|
|
"f9_chain": None,
|
||
|
|
}
|
||
|
|
if sweep:
|
||
|
|
code_matches["f9_chain"] = bool(c_f9 == f9_chain) # True = code sesuai spec
|
||
|
|
if not code_matches["f9_chain"]:
|
||
|
|
code_matches["f9_divergence_note"] = (
|
||
|
|
"code f9 TANPA expiry (BUG-P3S3-001): stale CHoCH tetap confirm")
|
||
|
|
|
||
|
|
return {
|
||
|
|
"id": case["id"],
|
||
|
|
"title": case["title"],
|
||
|
|
"expected_events": exp["events"],
|
||
|
|
"expected_f8": exp["f8_at_decision"],
|
||
|
|
"spec_oracle_events": [[b, d, round(float(lv), 6), p] for (b, d, lv, p) in events],
|
||
|
|
"spec_oracle_f8": f8,
|
||
|
|
"spec_oracle_f9_chain": int(f9_chain),
|
||
|
|
"code_port_events": [[b, d, round(float(lv), 6), p] for (b, d, lv, p) in c_events],
|
||
|
|
"code_port_f8": int(c_f8),
|
||
|
|
"code_port_f9": int(c_f9),
|
||
|
|
"code_matches_spec": code_matches,
|
||
|
|
"checks": checks,
|
||
|
|
"pass": all(checks.values()),
|
||
|
|
"note": case.get("expected", {}).get("note") or case.get("note"),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
cases_path = os.path.join(HERE, "spec_test_cases_choch_mss.json")
|
||
|
|
cases = json.load(open(cases_path, encoding="utf-8"))["cases"]
|
||
|
|
results = [run_case(c) for c in cases]
|
||
|
|
|
||
|
|
n_pass = sum(1 for r in results if r["pass"])
|
||
|
|
n_total = len(results)
|
||
|
|
print(f"=== P3-S.3 SPEC TESTS — CHOCH/MSS ({n_pass}/{n_total} PASS) ===")
|
||
|
|
print(f"{'ID':<8} {'PASS':<6} {'f8 spec':<8} {'f8 code':<8} {'f9 s/c':<8} notes")
|
||
|
|
for r in results:
|
||
|
|
cm = r["code_matches_spec"]
|
||
|
|
f9s = r["spec_oracle_f9_chain"]
|
||
|
|
f9c = r["code_port_f9"]
|
||
|
|
print(f"{r['id']:<8} {str(r['pass']):<6} {r['spec_oracle_f8']:<8} "
|
||
|
|
f"{r['code_port_f8']:<8} {str(f9s)+'/'+str(f9c):<8} {r['title'][:40]}")
|
||
|
|
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 (CHoCH/MSS).")
|
||
|
|
|
||
|
|
report = {
|
||
|
|
"spec_doc": "docs/SMC_CHOCH_MSS_SPEC_v1.md",
|
||
|
|
"phase": "P3-S.3",
|
||
|
|
"generated_utc": __import__("datetime").datetime.now(
|
||
|
|
__import__("datetime").timezone.utc).isoformat(),
|
||
|
|
"constants": {"INTERNAL_LEN": INTERNAL_LEN, "SEQ_WINDOW": SEQ_WINDOW},
|
||
|
|
"summary": {"total": n_total, "passed": n_pass, "failed": n_total - n_pass},
|
||
|
|
"cases": results,
|
||
|
|
}
|
||
|
|
outdir = os.path.join(HERE, "output")
|
||
|
|
os.makedirs(outdir, exist_ok=True)
|
||
|
|
out_path = os.path.join(outdir, "spec_tests_choch_mss_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()
|