# -*- coding: utf-8 -*- """P3.2.1 EVENT-BASED LABEL FORENSIC — SniperGold_ML. Diagnostic ONLY. No model training, no threshold optimization. Product question being measured (P3.2 §4): SMC candidate setup occurs -> would this setup be valid to trade? -> did the expected outcome occur before invalidation? This script measures the SETUP VALIDATION question on an event-based dataset D_EVENT built ONLY from EXISTING feature semantics (FEATURE_CONTRACT v1.0) — no new setup rule is created. Event families (direction = expected trade direction): E_BUY : f9==1 & f7>0 (CHoCH confirms bullish sweep) dir=+1 E_SELL : f9==1 & f7<0 (CHoCH confirms bearish sweep) dir=-1 E_BUY_STRICT : E_BUY & f18>=60 dir=+1 E_SELL_STRICT: E_SELL & f18>=60 dir=-1 E_EQH : f10==1 (equal-high swept -> bearish bias) dir=-1 E_EQL : f11==1 (equal-low swept -> bullish bias) dir=+1 Entry semantics (P3.2 §6): A = close of the closed decision bar (primary; closed-bar causality, consistent with the existing label and the runtime architecture). Robustness note: |close[i]-open[i+1]|/ATR reported (entry B comparison). Outcome families (P3.2 §7): L1 Directional return : sign-aligned forward return over horizon L2 MFE / MAE : excursions in ATR units + time-to-excursion L3 TP-before-SL : WIN / LOSS / UNRESOLVED / AMBIGUOUS (semantic TP/SL distances in ATR; ambiguous when both hit on the same bar — tick order unknown) L4 Time-to-outcome : time_to_TP, time_to_SL, censoring Horizons (M15, market-structure motivated, P3.2 §8): 4 / 8 / 16 / 24 bars (1h / 2h / 4h / 6h real time). Also: invalidation audit (opposite structure break before TP), event overlap audit (isolated / overlapping / clustered), and a frozen-feature information audit under candidate labels (does information reappear when the label is aligned with the event?). """ 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 train_model as TM HORIZONS = [4, 8, 16, 24] TP_SL_COMBOS = [(1.0, 0.5), (1.5, 0.75), (2.0, 1.0), (1.0, 1.0)] PRIMARY_COMBO = (1.5, 0.75) WARMUP = 100 # skip bars where structure features are warmup/zero def event_masks(F, n): f7, f9 = F[:, 7], F[:, 9] f10, f11, f18 = F[:, 10], F[:, 11], F[:, 18] m = { "E_BUY": (f9 == 1) & (f7 > 0), "E_SELL": (f9 == 1) & (f7 < 0), "E_BUY_STRICT": (f9 == 1) & (f7 > 0) & (f18 >= 60), "E_SELL_STRICT": (f9 == 1) & (f7 < 0) & (f18 >= 60), "E_EQH": (f10 == 1), "E_EQL": (f11 == 1), } direction = {"E_BUY": 1, "E_SELL": -1, "E_BUY_STRICT": 1, "E_SELL_STRICT": -1, "E_EQH": -1, "E_EQL": 1} idx = {} for k, mask in m.items(): e = np.where(mask)[0] e = e[e >= WARMUP] idx[k] = e return idx, direction def mfe_mae_times(entry, E, H, A_e, h, l, direction): """Vectorized MFE/MAE (ATR units) + first-bar time to final excursion.""" n_e = len(E) cmax = np.full(n_e, -np.inf) cmin = np.full(n_e, np.inf) for j in range(1, H + 1): cmax = np.maximum(cmax, h[E + j]) cmin = np.minimum(cmin, l[E + j]) if direction > 0: mfe = (cmax - entry) / A_e mae = (entry - cmin) / A_e else: mfe = (entry - cmin) / A_e mae = (cmax - entry) / A_e t_mfe = np.full(n_e, np.nan) t_mae = np.full(n_e, np.nan) cmax2 = np.full(n_e, -np.inf) cmin2 = np.full(n_e, np.inf) for j in range(1, H + 1): cmax2 = np.maximum(cmax2, h[E + j]) cmin2 = np.minimum(cmin2, l[E + j]) hit = (cmax2 == cmax) & np.isnan(t_mfe) t_mfe[hit] = j hit = (cmin2 == cmin) & np.isnan(t_mae) t_mae[hit] = j return mfe, mae, t_mfe, t_mae def tp_sl_scan(entry, E, H, tp_atr, sl_atr, A_e, h, l, direction): """First-hit TP/SL scan -> (tp_j, sl_j); inf when not reached in H.""" n_e = len(E) tp_j = np.full(n_e, np.inf) sl_j = np.full(n_e, np.inf) cmax = np.full(n_e, -np.inf) cmin = np.full(n_e, np.inf) tp_d = tp_atr * A_e sl_d = sl_atr * A_e if direction > 0: for j in range(1, H + 1): cmax = np.maximum(cmax, h[E + j]) cmin = np.minimum(cmin, l[E + j]) tp_j = np.where((cmax - entry >= tp_d) & np.isinf(tp_j), j, tp_j) sl_j = np.where((entry - cmin >= sl_d) & np.isinf(sl_j), j, sl_j) else: for j in range(1, H + 1): cmax = np.maximum(cmax, h[E + j]) cmin = np.minimum(cmin, l[E + j]) tp_j = np.where((entry - cmin >= tp_d) & np.isinf(tp_j), j, tp_j) sl_j = np.where((cmax - entry >= sl_d) & np.isinf(sl_j), j, sl_j) return tp_j, sl_j def classify_outcome(tp_j, sl_j): """WIN / LOSS / UNRESOLVED / AMBIGUOUS from first-hit bars.""" out = np.full(len(tp_j), "UNRESOLVED", dtype=object) win = (tp_j < sl_j) loss = (sl_j < tp_j) amb = (tp_j == sl_j) & np.isfinite(tp_j) out[win] = "WIN" out[loss] = "LOSS" out[amb] = "AMBIGUOUS" return out def opposite_structure_break(E, H, F8, F5, direction): """First bar j in 1..H where structure flips against the trade direction. Uses existing semantics: f8 (choch_dir) or f5 (chart_bias) sign opposite to the expected direction. Returns first-flip bar (inf if none).""" n_e = len(E) flip = np.full(n_e, np.inf) for j in range(1, H + 1): if direction > 0: bad = (F8[E + j] < 0) | (F5[E + j] < 0) else: bad = (F8[E + j] > 0) | (F5[E + j] > 0) flip = np.where(bad & np.isinf(flip), j, flip) return flip def event_gap_audit(E, H): """Median gap, P(next event within H bars), cluster sizes, strata.""" if len(E) < 2: return {"n": len(E), "note": "too few"} gaps = np.diff(E) within = (gaps <= H) clusters = [] cur = 1 for w in within: if w: cur += 1 else: clusters.append(cur) cur = 1 clusters.append(cur) prev_gap = np.concatenate([[10 ** 9], gaps]) next_gap = np.concatenate([gaps, [10 ** 9]]) isolated = (prev_gap > H) & (next_gap > H) return { "n": len(E), "median_gap_bars": float(np.median(gaps)), "mean_gap_bars": float(gaps.mean()), "P_next_within_H": float(within.mean()), "cluster_count": len(clusters), "median_cluster_size": float(np.median(clusters)), "max_cluster_size": float(np.max(clusters)), "n_isolated": int(isolated.sum()), "n_overlapping": int((~isolated).sum()), "P_isolated": float(isolated.mean()), } def mi_discrete(x, y): """Simple contingency MI (nats) for low-cardinality x vs binary y.""" xv = np.unique(x) if xv.size > 32 or xv.size < 2: return 0.0 m = len(y) n1 = int(y.sum()) if n1 == 0 or n1 == m: return 0.0 p1 = n1 / m mi = 0.0 for v in xv: nv = int((x == v).sum()) if nv == 0: continue n1v = int((y[x == v] == 1).sum()) pv = nv / m if n1v > 0: mi += pv * (n1v / nv) * np.log((n1v / nv) / p1) if n1v < nv: mi += pv * ((nv - n1v) / nv) * np.log(((nv - n1v) / nv) / (1 - p1)) return float(max(mi, 0.0)) 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) # reference only idx, direction = event_masks(F, n) F8 = F[:, 8] F5 = F[:, 5] summary = {"provenance": P3.provenance(), "script_sha": P3.sha256_file(os.path.abspath(__file__)), "dataset_hash": P3.dataset_hash(F, lab, A, t), "event_definition": { "E_BUY": "f9==1 & f7>0 (CHoCH confirms bullish sweep)", "E_SELL": "f9==1 & f7<0 (CHoCH confirms bearish sweep)", "E_BUY_STRICT": "E_BUY & f18>=60", "E_SELL_STRICT": "E_SELL & f18>=60", "E_EQH": "f10==1 (equal-high swept, bearish bias)", "E_EQL": "f11==1 (equal-low swept, bullish bias)"}, "entry_semantics": "A = close of closed decision bar", "horizons": HORIZONS, "tp_sl_combos": [list(x) for x in TP_SL_COMBOS], "events": {}, "overlap": {}, "feature_info_candidate_labels": {}} # ---- entry robustness (A vs B) ---- e_all = np.concatenate(list(idx.values())) gap_atr = np.abs(c[e_all] - o[np.minimum(e_all + 1, n - 1)]) / np.maximum(A[e_all], 1e-12) summary["entry_robustness"] = { "n": int(len(e_all)), "median_abs_close_to_next_open_atr": float(np.median(gap_atr)), "p95_abs_close_to_next_open_atr": float(np.percentile(gap_atr, 95))} # ---- per-event detail file (primary combo, H=16) ---- events_file = {"entry_semantics": "close of decision bar", "horizon_bars": 16, "combo": list(PRIMARY_COMBO), "events": {}} for k in ("E_BUY", "E_SELL", "E_EQH", "E_EQL"): E = idx[k] d = direction[k] rows = [] H = 16 if len(E) and (E + H < n).any(): E2 = E[E + H < n] entry = c[E2] A_e = np.maximum(A[E2], 1e-12) mfe, mae, t_mfe, t_mae = mfe_mae_times(entry, E2, H, A_e, h, l, d) tp_j, sl_j = tp_sl_scan(entry, E2, H, PRIMARY_COMBO[0], PRIMARY_COMBO[1], A_e, h, l, d) out = classify_outcome(tp_j, sl_j) for ri in range(len(E2)): rows.append({ "i": int(E2[ri]), "ts": dt.datetime.fromtimestamp(int(t[E2[ri]]), dt.timezone.utc).isoformat(), "dir": d, "fwd_atr": float((c[E2[ri] + H] - entry[ri]) / A_e[ri] * d), "mfe_atr": float(mfe[ri]), "mae_atr": float(mae[ri]), "t_mfe": None if np.isnan(t_mfe[ri]) else int(t_mfe[ri]), "t_mae": None if np.isnan(t_mae[ri]) else int(t_mae[ri]), "outcome": str(out[ri]), "tp_bar": int(tp_j[ri]) if np.isfinite(tp_j[ri]) else None, "sl_bar": int(sl_j[ri]) if np.isfinite(sl_j[ri]) else None}) events_file["events"][k] = rows P3.save_json("p3_2_label_events.json", events_file) # ---- per family x horizon x combo summary ---- for k in idx: E = idx[k] d = direction[k] fam = {"n": len(E), "direction": d} summary["events"][k] = fam if len(E) == 0: continue valid = E + max(HORIZONS) < n E2 = E[valid] fam["n_with_future"] = int(E2.size) if E2.size == 0: continue entry2 = c[E2] A2 = np.maximum(A[E2], 1e-12) per_h = {} for H in HORIZONS: mfe, mae, t_mfe, t_mae = mfe_mae_times(entry2, E2, H, A2, h, l, d) fwd = (c[E2 + H] - entry2) / A2 * d hh = {"n": int(len(E2)), "L1": {"mean_fwd_atr": float(fwd.mean()), "median_fwd_atr": float(np.median(fwd)), "P_dir_correct": float((fwd > 0).mean()), "P_dir_wrong": float((fwd < 0).mean())}, "L2": {"MFE_median_atr": float(np.nanmedian(mfe)), "MFE_mean_atr": float(np.nanmean(mfe)), "MAE_median_atr": float(np.nanmedian(mae)), "MAE_mean_atr": float(np.nanmean(mae)), "t_MFE_median": float(np.nanmedian(t_mfe)), "t_MAE_median": float(np.nanmedian(t_mae))}, "L3": {}, "L4": {}} for tp_a, sl_a in TP_SL_COMBOS: tp_j, sl_j = tp_sl_scan(entry2, E2, H, tp_a, sl_a, A2, h, l, d) out = classify_outcome(tp_j, sl_j) n_w = int((out == "WIN").sum()) n_l = int((out == "LOSS").sum()) n_u = int((out == "UNRESOLVED").sum()) n_a = int((out == "AMBIGUOUS").sum()) t_tp = tp_j[np.isfinite(tp_j)] t_sl = sl_j[np.isfinite(sl_j)] hh["L3"][f"{tp_a}/{sl_a}"] = { "WIN": n_w, "LOSS": n_l, "UNRESOLVED": n_u, "AMBIGUOUS": n_a, "P_win": float(n_w / len(out)), "P_loss": float(n_l / len(out)), "P_unresolved": float(n_u / len(out)), "P_ambiguous": float(n_a / len(out)), "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} # invalidation vs primary combo flip = opposite_structure_break(E2, H, F8, F5, d) tp_j, sl_j = tp_sl_scan(entry2, E2, H, PRIMARY_COMBO[0], PRIMARY_COMBO[1], A2, h, l, d) hh["invalidation"] = { "P_opposite_break_within_H": float(np.isfinite(flip).mean()), "P_opposite_break_before_TP": float( (np.isfinite(flip) & (flip < tp_j)).mean())} per_h[str(H)] = hh fam["horizons"] = per_h fam["overlap_H16"] = event_gap_audit(E, 16) # ---- feature information under candidate labels (frozen features) ---- print("\n=== FEATURE INFORMATION UNDER CANDIDATE LABELS (frozen features) ===") feats = {} for H in (8, 16, 24): for tp_a, sl_a in ((1.0, 0.5), (1.5, 0.75), (2.0, 1.0)): sub = {} for famk, dd in (("E_BUY", 1), ("E_SELL", -1)): Ef = idx[famk] Ef = Ef[Ef + H < n] if len(Ef) < 500: continue ent = c[Ef] Ae = np.maximum(A[Ef], 1e-12) tp_j, sl_j = tp_sl_scan(ent, Ef, H, tp_a, sl_a, Ae, h, l, dd) out = classify_outcome(tp_j, sl_j) keep = (out == "WIN") | (out == "LOSS") if keep.sum() < 300: continue y = (out[keep] == "WIN").astype(int) Xs = F[Ef[keep]] aucs = [] mis = [] for j in range(P3.NF): xj = Xs[:, j] if np.unique(xj).size < 2: continue a = TM.auc(y, xj) if a == a: aucs.append(a) mis.append(mi_discrete(xj, y)) devs = [abs(a - 0.5) for a in aucs] sub[famk] = {"n_win": int(y.sum()), "n_loss": int((y == 0).sum()), "max_abs_auc_dev": float(max(devs)) if devs else None, "mean_abs_auc_dev": float(np.mean(devs)) if devs else None, "max_mi": float(max(mis)) if mis else None} key = "H%d_TP%.1f_SL%.1f" % (H, tp_a, sl_a) feats[key] = sub print(" %s: %s" % (key, " | ".join( "%s n=%d max|dev|=%.4f maxMI=%.4f" % ( k, v["n_win"] + v["n_loss"], v["max_abs_auc_dev"] or 0.0, v["max_mi"] or 0.0) for k, v in sub.items()))) summary["feature_info_candidate_labels"] = feats P3.save_json("p3_2_label_summary.json", summary) print("\nP3.2.1 selesai: ml/p3/output/p3_2_label_events.json + p3_2_label_summary.json") if __name__ == "__main__": main()