# -*- coding: utf-8 -*- """P3-S.3 AUDIT — CHOCH STATE/EVENT FORENSIC (diagnostic, mirror f7 P3-S.0). Mengukur perilaku POPULASI state CHoCH (f8/f9/f18) pada XAUUSD M15 2017-2026 dengan semantik RUNTIME (equal-allowed pivot, build_structure_fast): - onsets : bar di mana CHoCH event BARU tercatat (choch_bar == r) - active bars : bar dengan choch_dir != 0 (state f8 "last CHoCH") - repetition : active/onsets - run stats : panjang run state (dir konstan) & age (r - choch_bar) - f9 staleness: chochOK (chochBar >= swpBar & searah) tanpa expiry -> berapa banyak bar dengan CHoCH age > SEQ_WINDOW masih "confirm" - f18 staleness: kontribusi +15 (chochDir==swpDir) pada CHoCH basi Output: output/audit_choch_state.json Disiplin: OBSERVE -> RECORD. Bukan oracle; bukan definisi. Spec = truth. """ import os import sys import json import numpy as np HERE = os.path.dirname(os.path.abspath(__file__)) REPO_ML = os.path.normpath(os.path.join(HERE, "..", "..", "..", "ml")) # publish/AlgoForge/ml REPO_P3 = os.path.normpath(os.path.join(REPO_ML, "p3")) REPO_PARITY = os.path.normpath(os.path.join(REPO_ML, "parity")) SRC_LEGACY = os.path.normpath(os.path.join( HERE, "..", "..", "..", "..", "..", "SniperGold_ML")) # Shared Projects\SniperGold_ML sys.path.insert(0, HERE) sys.path.insert(0, REPO_ML) sys.path.insert(0, REPO_P3) sys.path.insert(0, REPO_PARITY) sys.path.insert(0, SRC_LEGACY) import p3_common as P3 import smc_semantic_common as SC import build_features_p2 as BFP # Checkout ini nested (publish\AlgoForge) -> path relatif p3_common meleset; # data riil ada di MQL5\Files\AlgoForge\Data. Patch konstanta data root. P3.DATA = r"D:\TradingTerminal\HFM Metatrader 5\MQL5\Files\AlgoForge\Data" SEQ_WINDOW = SC.SEQ_WINDOW # 40 def main(): t, o, h, l, c, v, htf = SC.load_data() n = len(c) # ---- struktur runtime-semantics (equal allowed), full-feed, begin=100 ---- sw_at = np.zeros(n, dtype=int) sw = BFP.build_structure_fast(o, h, l, c, SC.SWING_LEN, False, sw_at, begin=100) inn = BFP.build_structure_fast(o, h, l, c, SC.INTERNAL_LEN, True, sw["trend"].copy(), begin=100) cd = inn["choch_dir"] # state f8 (last CHoCH direction) cb = inn["choch_bar"] # bar event CHoCH terakhir new_event = (cb == np.arange(n)) # bar di mana event BARU tercatat onsets = int(new_event.sum()) active = int((cd != 0).sum()) rep = active / onsets if onsets else 0.0 # run stat: panjang run arah konstan (cd != 0) runs = [] run_len = 0 prev = 0 for x in cd: if x != 0 and x == prev: run_len += 1 else: if prev != 0: runs.append(run_len) run_len = 1 if x != 0 else 0 prev = x if prev != 0: runs.append(run_len) runs = np.array(runs, dtype=int) if runs else np.zeros(0, dtype=int) # age state (bar setelah event) mask = cb >= 0 age = np.where(mask, np.arange(n) - cb, -1) age_active = age[(cd != 0) & mask] # ---- sweep state legacy (utk f9/f18); pivot list = feed internal build_structure_fast ---- is_ph, is_pl = BFP._pivots_vec(h, l, SC.INTERNAL_LEN) piv = [] for p in range(SC.INTERNAL_LEN, n - SC.INTERNAL_LEN): if p >= 95 and (is_ph[p] or is_pl[p]): piv.append((p, float(h[p]) if is_ph[p] else float(l[p]), bool(is_ph[p]))) sweep_dir, sweep_bar = SC.sweep_state(h, l, c, piv) # f9 chochOK (kode EA, tanpa expiry): chochDir!=0 & chochBar>=swpBar & dir sama chochOK = np.zeros(n, dtype=bool) for r in range(n): if cd[r] != 0 and cb[r] >= sweep_bar[r] and cd[r] == sweep_dir[r]: chochOK[r] = True # f9 pada CHoCH basi (age > SEQ_WINDOW) ok_basi = chochOK & (age > SEQ_WINDOW) ok_fresh = chochOK & (age >= 0) & (age <= SEQ_WINDOW) # f18 +15 leg: chochDir==swpDir (tanpa expiry) f18_leg = (cd != 0) & (cd == sweep_dir) f18_leg_basi = f18_leg & (age > SEQ_WINDOW) # kapan state f8 pertama kali menjadi 0 setelah pernah aktif (post-first-event) first_active = int(np.argmax(cd != 0)) if (cd != 0).any() else -1 zero_after_first = bool((cd[first_active:] == 0).any()) if first_active >= 0 else None result = { "scope": "XAUUSD M15 2017-2026 full-feed (runtime-semantics equal-allowed pivots)", "n_bars": int(n), "choch_state": { "onsets": onsets, "active_bars": int(active), "repetition_ratio": round(float(rep), 3), "zero_after_first_event": zero_after_first, "run_median": float(np.median(runs)) if len(runs) else None, "run_max": int(runs.max()) if len(runs) else None, "run_n": int(len(runs)), "age_active_median": float(np.median(age_active)) if len(age_active) else None, "age_active_p99": float(np.percentile(age_active, 99)) if len(age_active) else None, "age_active_max": int(age_active.max()) if len(age_active) else None, "note": "state f8 = last CHoCH direction (persisten, per FEATURE_CONTRACT); " "bukan duplicate emission detector." }, "f9_chochOK": { "true_bars": int(chochOK.sum()), "fresh_bars_le40": int(ok_fresh.sum()), "stale_bars_gt40": int(ok_basi.sum()), "stale_fraction": round(float(ok_basi.sum() / max(1, int(chochOK.sum()))), 4), "note": "f9 memakai CHoCH tanpa expiry -> event basi tetap 'confirm'" }, "f18_leg_choch_swp": { "true_bars": int(f18_leg.sum()), "stale_bars_gt40": int(f18_leg_basi.sum()), "stale_fraction": round(float(f18_leg_basi.sum() / max(1, int(f18_leg.sum()))), 4), "note": "f18 +15 (chochDir==swpDir) tanpa expiry" }, "seq_window": SEQ_WINDOW, } out_path = os.path.join(HERE, "output", "audit_choch_state.json") os.makedirs(os.path.dirname(out_path), exist_ok=True) with open(out_path, "w", encoding="utf-8") as f: json.dump(result, f, indent=2, default=str) print(json.dumps(result, indent=2, default=str)) print("[saved]", out_path) if __name__ == "__main__": main()