239 lines
10 KiB
Python
239 lines
10 KiB
Python
# -*- 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()
|