# -*- coding: utf-8 -*- """P3.2.2 SURVIVAL / TIME-TO-EVENT DIAGNOSTIC — SniperGold_ML. Diagnostic ONLY. No production survival model, no TP/SL optimization. Input: LEAD events (de-overlapped, overlap window H=16) of the EXISTING SMC event stream. Competing risks TP vs SL are kept SEPARATE (never merged). Outputs: * cumulative incidence by bar t (1..48): P(TP by t), P(SL by t), P(ambiguous by t), P(no event by t) [competing risks separated] * horizon report H in {8,16,24,48}: TP prob, SL prob, censoring rate, median event time, median TP time, median SL time * MFE/MAE conditional on final status (WIN/LOSS/CENSORED/AMBIGUOUS) * invalidation diagnostic: opposite structure break (secondary outcome only) - TP before invalidation / SL before invalidation / invalidation first * serial dependence: outcome autocorrelation, run length, same-direction clustering, label persistence — ALL events vs LEAD events Primary candidate combo: TP=1.5 ATR, SL=0.75 ATR (semantic 2:1, candidate only). Symmetric sensitivity 1.0/1.0 reported for horizon/incidence. """ import os import sys 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 import deoverlap_audit as DOA FAMILIES = ["E_BUY", "E_SELL", "E_EQH", "E_EQL"] PRIMARY = (1.5, 0.75) SYMMETRIC = (1.0, 1.0) HMAX = 48 REPORT_H = [8, 16, 24, 48] def incidence(leads, Hmax, combo, h, l, c, A, direction): """Cumulative incidence arrays over t=1..Hmax (competing risks separate).""" E = leads[leads + Hmax < len(c)] if len(E) == 0: return None ent = c[E] Ae = np.maximum(A[E], 1e-12) tp_j, sl_j = LEA.tp_sl_scan(ent, E, Hmax, combo[0], combo[1], Ae, h, l, direction) n = len(E) tp_inc = np.zeros(Hmax + 1) sl_inc = np.zeros(Hmax + 1) am_inc = np.zeros(Hmax + 1) for tt in range(1, Hmax + 1): tp_inc[tt] = ((tp_j <= tt) & (tp_j < sl_j)).mean() sl_inc[tt] = ((sl_j <= tt) & (sl_j < tp_j)).mean() am_inc[tt] = ((tp_j == sl_j) & (tp_j <= tt)).mean() no_inc = 1.0 - tp_inc - sl_inc - am_inc return {"E": E, "tp_j": tp_j, "sl_j": sl_j, "tp_inc": tp_inc, "sl_inc": sl_inc, "am_inc": am_inc, "no_inc": no_inc} def serial_dependence(E, H, combo, h, l, c, A, direction): """Outcome serial dependence on temporal event order (resolved only).""" if len(E) < 50: return None E = E[E + H < len(c)] ent = c[E] Ae = np.maximum(A[E], 1e-12) 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) resolved = (oc == "WIN") | (oc == "LOSS") if resolved.sum() < 50: return None y = (oc[resolved] == "WIN").astype(int) if len(y) > 1: ac1 = float(np.corrcoef(y[:-1], y[1:])[0, 1]) if np.ptp(y) > 0 else 0.0 same = float((y[1:] == y[:-1]).mean()) else: ac1, same = 0.0, 0.0 # run length of WIN (1) and LOSS (0) runs = [] cur = 1 for i in range(1, len(y)): if y[i] == y[i - 1]: cur += 1 else: runs.append(cur) cur = 1 runs.append(cur) # same-direction event clustering: fraction of consecutive events same dir dirs = direction * np.ones(len(E), dtype=int) same_dir = float((dirs[1:] == dirs[:-1]).mean()) if len(dirs) > 1 else 0.0 return {"n_resolved": int(len(y)), "outcome_autocorr_lag1": float(ac1), "P_same_outcome_consec": float(same), "mean_run_len": float(np.mean(runs)), "max_run_len": float(np.max(runs)), "P_same_dir_consec": float(same_dir)} 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" provenance["entry_semantics"] = "close of closed decision bar" provenance["deoverlap"] = "lead event per cluster, overlap window H=16" provenance["invalidation_semantics"] = "opposite structure break (f8 or f5 flip)" report = {"provenance": provenance, "primary_combo": list(PRIMARY), "symmetric_sensitivity": list(SYMMETRIC), "families": {}} print("=== P3.2.2 SURVIVAL / TIME-TO-EVENT (LEAD events, de-overlap H16) ===") for k in FAMILIES: E = idx[k] d = direction[k] leads, follow, clusters = DOA.cluster_events(E, 16) fam = {"n_original": int(len(E)), "n_lead": int(len(leads)), "direction": d} # ---- primary incidence ---- inc = incidence(leads, HMAX, PRIMARY, h, l, c, A, d) if inc is None: fam["note"] = "no lead with future" report["families"][k] = fam continue E_l = inc["E"] ent = c[E_l] Ae = np.maximum(A[E_l], 1e-12) tp_j, sl_j = inc["tp_j"], inc["sl_j"] oc = LEA.classify_outcome(tp_j, sl_j) first_j = np.minimum(tp_j, sl_j) first_j_fin = first_j[np.isfinite(first_j)] t_tp = tp_j[(tp_j < sl_j) & np.isfinite(tp_j)] t_sl = sl_j[(sl_j < tp_j) & np.isfinite(sl_j)] fam["cumulative_incidence"] = { "bars": list(range(1, HMAX + 1)), "TP": [float(x) for x in inc["tp_inc"][1:]], "SL": [float(x) for x in inc["sl_inc"][1:]], "AMBIGUOUS": [float(x) for x in inc["am_inc"][1:]], "NO_EVENT": [float(x) for x in inc["no_inc"][1:]], } fam["horizon_report"] = {} for H in REPORT_H: fam["horizon_report"][str(H)] = { "P_TP_by_H": float(inc["tp_inc"][H]), "P_SL_by_H": float(inc["sl_inc"][H]), "P_ambiguous_by_H": float(inc["am_inc"][H]), "P_censored_by_H": float(inc["no_inc"][H]), "median_event_time": float(np.median(first_j_fin[first_j_fin <= H])) if (first_j_fin <= H).any() else None, "median_TP_time": float(np.median(t_tp[t_tp <= H])) if (t_tp <= H).any() else None, "median_SL_time": float(np.median(t_sl[t_sl <= H])) if (t_sl <= H).any() else None, } # ---- MFE/MAE conditional on final status (primary combo) ---- mfe, mae, t_mfe, t_mae = LEA.mfe_mae_times(ent, E_l, HMAX, Ae, h, l, d) fam["mfe_mae_by_status"] = {} for st in ("WIN", "LOSS", "UNRESOLVED", "AMBIGUOUS"): m = oc == st if m.sum() == 0: continue fam["mfe_mae_by_status"][st] = { "n": int(m.sum()), "MFE_median_atr": float(np.nanmedian(mfe[m])), "MAE_median_atr": float(np.nanmedian(mae[m])), "MFE_mean_atr": float(np.nanmean(mfe[m])), "MAE_mean_atr": float(np.nanmean(mae[m])), "median_MFE_over_MAE_ratio": float( np.nanmedian(mfe[m] / np.maximum(mae[m], 1e-9))), } # ---- invalidation (secondary outcome only) ---- flip = LEA.opposite_structure_break(E_l, HMAX, F8, F5, d) inv_first = np.isfinite(flip) & (flip < first_j) tp_before_inv = np.isfinite(tp_j) & (tp_j < sl_j) & ( ~np.isfinite(flip) | (tp_j < flip)) sl_before_inv = np.isfinite(sl_j) & (sl_j < tp_j) & ( ~np.isfinite(flip) | (sl_j < flip)) fam["invalidation"] = { "P_invalidation_within_HMAX": float(np.isfinite(flip).mean()), "P_invalidation_before_TP_SL": float(inv_first.mean()), "P_TP_before_invalidation": float(tp_before_inv.mean()), "P_SL_before_invalidation": float(sl_before_inv.mean()), "median_t_invalidation": float(np.median(flip[np.isfinite(flip)])) if np.isfinite(flip).any() else None, } # ---- serial dependence: ALL vs LEAD (H=16, primary combo) ---- fam["serial_dependence"] = { "ALL": serial_dependence(idx[k], 16, PRIMARY, h, l, c, A, d), "LEAD": serial_dependence(leads, 16, PRIMARY, h, l, c, A, d), } report["families"][k] = fam hr = fam["horizon_report"] print("\n[%s] lead=%d (orig %d) dir=%+d" % (k, fam["n_lead"], fam["n_original"], d)) for H in ("8", "16", "24", "48"): x = hr[H] print(" H=%-2s TP=%.3f SL=%.3f cens=%.3f amb=%.3f | med_ev=%s " "med_TP=%s med_SL=%s" % ( H, x["P_TP_by_H"], x["P_SL_by_H"], x["P_censored_by_H"], x["P_ambiguous_by_H"], ("%.1f" % x["median_event_time"]) if x["median_event_time"] else "-", ("%.1f" % x["median_TP_time"]) if x["median_TP_time"] else "-", ("%.1f" % x["median_SL_time"]) if x["median_SL_time"] else "-")) mb = fam["mfe_mae_by_status"] for st in ("WIN", "LOSS", "UNRESOLVED", "AMBIGUOUS"): if st in mb: print(" %-11s n=%-6d MFE_med=%.2f MAE_med=%.2f ratio=%.2f" % ( st, mb[st]["n"], mb[st]["MFE_median_atr"], mb[st]["MAE_median_atr"], mb[st]["median_MFE_over_MAE_ratio"])) sd = fam["serial_dependence"] if sd["ALL"] and sd["LEAD"]: print(" serial ALL : ac1=%.4f same_out=%.4f run=%.1f same_dir=%.4f" % ( sd["ALL"]["outcome_autocorr_lag1"], sd["ALL"]["P_same_outcome_consec"], sd["ALL"]["mean_run_len"], sd["ALL"]["P_same_dir_consec"])) print(" serial LEAD: ac1=%.4f same_out=%.4f run=%.1f same_dir=%.4f" % ( sd["LEAD"]["outcome_autocorr_lag1"], sd["LEAD"]["P_same_outcome_consec"], sd["LEAD"]["mean_run_len"], sd["LEAD"]["P_same_dir_consec"])) P3.save_json("p3_2_survival.json", report) print("\nP3.2.2 survival diagnostic selesai: ml/p3/output/p3_2_survival.json") if __name__ == "__main__": main()