# -*- 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()