# -*- coding: utf-8 -*- """P3.2.2 FORMAL EVENT DE-OVERLAP AUDIT — SniperGold_ML. Diagnostic ONLY. No model training, no TP/SL optimization. Formal event clustering of the EXISTING SMC event stream (no semantics change): * overlap window H = outcome horizon (8 / 16 / 24 bars) — robustness, no "best H" * cluster = maximal run of events with gap <= H * lead event = earliest eligible event of each cluster (kept) * FOLLOW_ON = all later events of a cluster (excluded from the primary de-overlapped dataset, but never deleted — reported separately) * verify: next_selected - selected > H for the de-overlapped stream Per family (E_BUY / E_SELL / E_EQH / E_EQL / strict variants) and per H: * original / clusters / lead / follow-on counts * median & max cluster size, retention rate, follow-on rate * ALL vs LEAD outcome comparison (WIN/LOSS/UNRESOLVED/AMBIGUOUS, MFE, MAE, median time-to-TP, median time-to-SL) under the primary candidate combo 1.5/0.75 ATR and the symmetric sensitivity 1.0/1.0 ATR. Entry semantics: close of the closed decision bar (unchanged from P3.2.1). """ import os import sys import datetime as dt import numpy as np HERE = os.path.dirname(os.path.abspath(__file__)) if HERE not in sys.path: sys.path.insert(0, HERE) import p3_common as P3 import label_event_audit as LEA HORIZONS = [8, 16, 24] COMBOS = [(1.5, 0.75), (1.0, 1.0)] # primary candidate + symmetric sensitivity FAMILIES = ["E_BUY", "E_SELL", "E_EQH", "E_EQL", "E_BUY_STRICT", "E_SELL_STRICT"] def cluster_events(E, H): """Cluster events with gap <= H. Returns (lead, follow_on, clusters).""" if len(E) == 0: return np.array([], dtype=int), np.array([], dtype=int), [] leads = [] clusters = [] cur = [E[0]] for e in E[1:]: if e - cur[-1] <= H: cur.append(e) else: clusters.append(np.array(cur, dtype=int)) leads.append(cur[0]) cur = [e] clusters.append(np.array(cur, dtype=int)) leads.append(cur[0]) leads = np.array(leads, dtype=int) follow = np.concatenate([c[1:] for c in clusters]) if len(clusters) else np.array([], dtype=int) return leads, follow, clusters def outcome_stats(E, H, combo, h, l, c, A, direction): """WIN/LOSS/UNRESOLVED/AMBIGUOUS + MFE/MAE + median times for event set.""" if len(E) == 0: return None E = E[E + H < len(c)] if len(E) == 0: return None ent = c[E] Ae = np.maximum(A[E], 1e-12) mfe, mae, t_mfe, t_mae = LEA.mfe_mae_times(ent, E, H, Ae, h, l, direction) tp_j, sl_j = LEA.tp_sl_scan(ent, E, H, combo[0], combo[1], Ae, h, l, direction) oc = LEA.classify_outcome(tp_j, sl_j) n = len(oc) t_tp = tp_j[np.isfinite(tp_j)] t_sl = sl_j[np.isfinite(sl_j)] return { "n": int(n), "P_win": float((oc == "WIN").mean()), "P_loss": float((oc == "LOSS").mean()), "P_unresolved": float((oc == "UNRESOLVED").mean()), "P_ambiguous": float((oc == "AMBIGUOUS").mean()), "MFE_median_atr": float(np.nanmedian(mfe)), "MAE_median_atr": float(np.nanmedian(mae)), "median_t_TP": float(np.median(t_tp)) if len(t_tp) else None, "median_t_SL": float(np.median(t_sl)) if len(t_sl) else None, } def main(): t, o, h, l, c, v, htf = P3.load_data() F = P3.load_F() A = P3.atr_series(h, l, c) n = len(c) lab = P3.make_label(c, A) idx, direction = LEA.event_masks(F, n) F8 = F[:, 8] F5 = F[:, 5] provenance = P3.provenance() provenance["P3_1_P3_2_1_CHECKPOINT_SHA"] = "dc1faa92f36ad22d4062315a5654965a08a006dc" provenance["script_sha"] = P3.sha256_file(os.path.abspath(__file__)) provenance["dataset_hash"] = P3.dataset_hash(F, lab, A, t) provenance["event_semantics"] = "existing FEATURE_CONTRACT v1.0 (f9/f7/f10/f11)" provenance["entry_semantics"] = "close of closed decision bar" provenance["invalidation_semantics"] = "opposite structure break (f8 or f5 flip)" print("=== P3.2.2 FORMAL DE-OVERLAP AUDIT ===") for H in HORIZONS: report = {"provenance": provenance, "overlap_window_H": H, "combo": [list(x) for x in COMBOS], "families": {}} print("\n--- H=%d ---" % H) for k in FAMILIES: E = idx[k] d = direction[k] leads, follow, clusters = cluster_events(E, H) # verify de-overlap ok_verify = bool(np.all(np.diff(leads) > H)) if len(leads) > 1 else True sizes = np.array([len(c_) for c_ in clusters]) if clusters else np.array([0]) fam = { "n_original": int(len(E)), "n_clusters": int(len(clusters)), "n_lead": int(len(leads)), "n_follow_on": int(len(follow)), "median_cluster_size": float(np.median(sizes)) if len(sizes) else None, "max_cluster_size": float(np.max(sizes)) if len(sizes) else None, "retention_rate": float(len(leads) / len(E)) if len(E) else None, "follow_on_rate": float(len(follow) / len(E)) if len(E) else None, "deoverlap_verified_next_gt_H": bool(ok_verify), "outcomes": {}, } for combo in COMBOS: key = "%s/%s" % (combo[0], combo[1]) fam["outcomes"][key] = { "ALL": outcome_stats(E, H, combo, h, l, c, A, d), "LEAD": outcome_stats(leads, H, combo, h, l, c, A, d), } report["families"][k] = fam oa = fam["outcomes"]["1.5/0.75"] print(" %-14s orig=%6d lead=%5d follow=%6d ret=%.4f med_cl=%5.0f " "max_cl=%5.0f verify=%s" % ( k, fam["n_original"], fam["n_lead"], fam["n_follow_on"], fam["retention_rate"], fam["median_cluster_size"], fam["max_cluster_size"], ok_verify)) if oa["ALL"] and oa["LEAD"]: print(" ALL : W=%.3f L=%.3f U=%.3f | MFE=%.2f MAE=%.2f | " "tTP=%s tSL=%s" % ( oa["ALL"]["P_win"], oa["ALL"]["P_loss"], oa["ALL"]["P_unresolved"], oa["ALL"]["MFE_median_atr"], oa["ALL"]["MAE_median_atr"], ("%.1f" % oa["ALL"]["median_t_TP"]) if oa["ALL"]["median_t_TP"] else "-", ("%.1f" % oa["ALL"]["median_t_SL"]) if oa["ALL"]["median_t_SL"] else "-")) print(" LEAD: W=%.3f L=%.3f U=%.3f | MFE=%.2f MAE=%.2f | " "tTP=%s tSL=%s" % ( oa["LEAD"]["P_win"], oa["LEAD"]["P_loss"], oa["LEAD"]["P_unresolved"], oa["LEAD"]["MFE_median_atr"], oa["LEAD"]["MAE_median_atr"], ("%.1f" % oa["LEAD"]["median_t_TP"]) if oa["LEAD"]["median_t_TP"] else "-", ("%.1f" % oa["LEAD"]["median_t_SL"]) if oa["LEAD"]["median_t_SL"] else "-")) P3.save_json("p3_2_deoverlap_h%d.json" % H, report) print("\nP3.2.2 de-overlap audit selesai: p3_2_deoverlap_h8/16/24.json") if __name__ == "__main__": main()