# -*- coding: utf-8 -*- """P3.1b LABEL FORENSIC AUDIT — SniperGold_ML. Diagnostic ONLY. Answers the second research question: "Apakah label 24-bar / 0.75 x ATR sesuai dengan tujuan SMC?" Contents: * base label (24-bar, 0.75 ATR): class balance, entropy, outcome rate * label decomposition: future return, MFE, MAE, time-to-hit, TP-before-SL, SL-before-TP, no-hit — all without look-ahead * horizon sensitivity (descriptive): 12 / 24 / 36 / 48 bars -> does SMC information persist across horizons? * label temporal persistence: autocorrelation, transition matrix, P(label_t | label_{t-1}) * label x SMC state table (diagnostic link between label and setup concept) NOT a parameter optimization; no horizon/threshold is promoted to production. """ import os import sys import json 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 = [12, 24, 36, 48] BASE_H = TM.H_LABEL # 24 BASE_T = TM.LABEL_ATR # 0.75 def entropy2(p1, p0): p = [x for x in (p1, p0) if x > 0] if not p: return 0.0 return float(-sum(x * np.log(x) for x in p)) def entropy3(p1, p0, pz): p = [x for x in (p1, p0, pz) if x > 0] return float(-sum(x * np.log(x) for x in p)) 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) report = {"provenance": P3.provenance(), "dataset_hash": P3.dataset_hash(F, P3.make_label(c, A), A, t), "base_label": {"horizon": BASE_H, "thr_mult": BASE_T}} # ===================================================================== # 1) BASE LABEL decomposition # ===================================================================== print("=== BASE LABEL (24-bar, 0.75 x ATR) — class balance & entropy ===") lab = P3.make_label(c, A) p1 = float((lab == 1).mean()) pm1 = float((lab == -1).mean()) pz = float((lab == 0).mean()) labeled = lab != 0 n_lab = int(labeled.sum()) n_pos = int((lab == 1).sum()) n_neg = int((lab == -1).sum()) e2 = entropy2(p1 / (p1 + pm1), pm1 / (p1 + pm1)) e3 = entropy3(p1, pm1, pz) print(f" P(+1)={p1:.4f} P(-1)={pm1:.4f} P(0)={pz:.4f} " f"| labeled={n_lab} ({n_lab/n*100:.1f}%) pos={n_pos} neg={n_neg}") print(f" 2-class entropy (labeled only)={e2:.4f} | 3-class entropy={e3:.4f}") report["base_class"] = {"p_pos": p1, "p_neg": pm1, "p_zero": pz, "n_labeled": n_lab, "n_pos": n_pos, "n_neg": n_neg, "entropy_labeled": e2, "entropy_3class": e3} print("\n=== LABEL DECOMPOSITION (H=24, thr=0.75 ATR) — all bars with future ===") dec = P3.label_decompose(c, h, l, A, np.arange(n - BASE_H), horizon=BASE_H, thr_mult=BASE_T) # summary on ALL decision bars (includes flat outcomes) fwd_atr = dec["fwd"] / A[:n - BASE_H] mfe = dec["mfe"] mae = dec["mae"] tp = dec["tp_before_sl"] sl = dec["sl_before_tp"] nh = dec["no_hit"] t_hit = dec["t_hit"] stat = lambda x: dict(mean=float(np.nanmean(x)), median=float(np.nanmedian(x)), p25=float(np.nanpercentile(x, 25)), p75=float(np.nanpercentile(x, 75)), std=float(np.nanstd(x))) all_sum = {"n": int(len(fwd_atr)), "fwd_atr": stat(fwd_atr), "mfe_atr": stat(mfe), "mae_atr": stat(mae), "P_tp_before_sl": float(tp.mean()), "P_sl_before_tp": float(sl.mean()), "P_no_hit": float(nh.mean()), "time_to_hit_median": float(np.nanmedian(t_hit)) if np.isfinite(t_hit).any() else None} print(f" n={all_sum['n']}") print(f" fwd/ATR: mean={all_sum['fwd_atr']['mean']:.3f} med={all_sum['fwd_atr']['median']:.3f}") print(f" MFE/ATR: mean={all_sum['mfe_atr']['mean']:.3f} med={all_sum['mfe_atr']['median']:.3f}") print(f" MAE/ATR: mean={all_sum['mae_atr']['mean']:.3f} med={all_sum['mae_atr']['median']:.3f}") print(f" TP-before-SL={all_sum['P_tp_before_sl']:.4f} SL-before-TP={all_sum['P_sl_before_tp']:.4f} " f"no-hit={all_sum['P_no_hit']:.4f} | median time-to-hit={all_sum['time_to_hit_median']}") # direction-conditioned dir_sum = {} for dname, mask in (("long", lab[:n - BASE_H] == 1), ("short", lab[:n - BASE_H] == -1), ("flat", lab[:n - BASE_H] == 0)): if not mask.any(): dir_sum[dname] = None continue sub = dict(n=int(mask.sum()), fwd_atr=stat(fwd_atr[mask]), mfe_atr=stat(mfe[mask]), mae_atr=stat(mae[mask]), P_tp=float(tp[mask].mean()), P_sl=float(sl[mask].mean()), P_no_hit=float(nh[mask].mean()), time_to_hit_median=float(np.nanmedian(t_hit[mask])) if np.isfinite(t_hit[mask]).any() else None) dir_sum[dname] = sub print(f" [{dname}] n={sub['n']} MFE med={sub['mfe_atr']['median']:.2f} " f"MAE med={sub['mae_atr']['median']:.2f} " f"TP={sub['P_tp']:.3f} SL={sub['P_sl']:.3f} nohit={sub['P_no_hit']:.3f} " f"t_hit_med={sub['time_to_hit_median']}") report["decomposition_H24"] = {"all": all_sum, "by_direction": dir_sum} # ===================================================================== # 2) HORIZON SENSITIVITY (descriptive, no promotion) # ===================================================================== print("\n=== HORIZON SENSITIVITY (thr fixed = 0.75 x ATR) ===") horiz = {} for H in HORIZONS: labH = P3.make_label(c, A, horizon=H, thr_mult=BASE_T) pH1 = float((labH == 1).mean()) pHm1 = float((labH == -1).mean()) pHz = float((labH == 0).mean()) decH = P3.label_decompose(c, h, l, A, np.arange(n - H), horizon=H, thr_mult=BASE_T) # label persistence: autocorrelation of 3-state label (lag 1) s3 = labH.astype(float) ac1 = float(np.corrcoef(s3[:-1], s3[1:])[0, 1]) if np.ptp(s3) > 0 else 0.0 ent_lab = entropy2(pH1 / (pH1 + pHm1), pHm1 / (pH1 + pHm1)) ent3 = entropy3(pH1, pHm1, pHz) e = {"horizon": H, "p_pos": float(pH1), "p_neg": float(pHm1), "p_zero": float(pHz), "n_labeled": int((labH != 0).sum()), "outcome_rate": float((labH != 0).mean()), "entropy_labeled": ent_lab, "entropy_3class": ent3, "autocorr3state_lag1": ac1, "fwd_atr_med": float(np.nanmedian(decH["fwd"] / A[:n - H])), "mfe_med": float(np.nanmedian(decH["mfe"])), "mae_med": float(np.nanmedian(decH["mae"])), "P_tp_before_sl": float(decH["tp_before_sl"].mean()), "P_sl_before_tp": float(decH["sl_before_tp"].mean()), "P_no_hit": float(decH["no_hit"].mean()), "t_hit_median": float(np.nanmedian(decH["t_hit"])) if np.isfinite(decH["t_hit"]).any() else None} horiz[str(H)] = e print(f" H={H:>2}: P+={pH1:.4f} P-={pHm1:.4f} P0={pHz:.4f} " f"| ent3={ent3:.4f} ac1={ac1:.4f} | MFE med={e['mfe_med']:.2f} " f"MAE med={e['mae_med']:.2f} | TP={e['P_tp_before_sl']:.3f} " f"SL={e['P_sl_before_tp']:.3f} nohit={e['P_no_hit']:.3f} " f"t_hit_med={e['t_hit_median']}") report["horizon_sensitivity"] = horiz # ===================================================================== # 3) LABEL TEMPORAL PERSISTENCE (full 3-state series) # ===================================================================== print("\n=== LABEL TEMPORAL PERSISTENCE (3-state, lag-1 transition) ===") lab = P3.make_label(c, A) # base 24 s3 = lab.astype(int) tr = {} for a in (-1, 0, 1): row = {} for b in (-1, 0, 1): row[str(b)] = int(((s3[:-1] == a) & (s3[1:] == b)).sum()) tr[str(a)] = row # conditional probabilities from transition counts trp = {} for a in (-1, 0, 1): tot = sum(tr[str(a)].values()) trp[str(a)] = {k: (v / tot if tot else 0.0) for k, v in tr[str(a)].items()} # labeled-only persistence (ignore flats) lm = lab != 0 li = np.where(lm)[0] same_dir = float((lab[li[1:]] == lab[li[:-1]]).mean()) if len(li) > 1 else 0.0 p1_given1 = float((lab[li[1:]][lab[li[:-1]] == 1] == 1).mean()) if (lab[li[:-1]] == 1).any() else float("nan") p_neg_given_neg = float((lab[li[1:]][lab[li[:-1]] == -1] == -1).mean()) if (lab[li[:-1]] == -1).any() else float("nan") print(f" transition P(state_t | state_{'{t-1}'}):") for a in (-1, 0, 1): print(f" from {a:+d}: " + " ".join(f"->{b:+d}:{trp[str(a)][str(b)]:.3f}" for b in (-1, 0, 1))) print(f" labeled-only same-direction persistence={same_dir:.4f} " f"(P(+1|+1)={p1_given1:.4f} P(-1|-1)={p_neg_given_neg:.4f})") report["label_persistence"] = {"transition_counts": tr, "transition_prob": trp, "labeled_same_dir": same_dir, "P_pos_given_pos": p1_given1, "P_neg_given_neg": p_neg_given_neg} # ===================================================================== # 4) LABEL x SMC STATE (diagnostic link label <-> setup concept) # ===================================================================== print("\n=== LABEL x SMC STATE (H=24, thr=0.75) ===") lab = P3.make_label(c, A) y = lab p1_uncond = float((y == 1).mean()) rows = [] # f5 chart bias direction for nm, col, conds in ( ("f5_chart_bias", 5, {"bull": (F[:, 5] > 0), "bear": (F[:, 5] < 0)}), ("f7_sweep", 7, {"bull_grab": (F[:, 7] > 0), "bear_grab": (F[:, 7] < 0)}), ("f8_choch", 8, {"bull_choch": (F[:, 8] > 0), "bear_choch": (F[:, 8] < 0)}), ("f9_confirm", 9, {"confirm_on": (F[:, 9] == 1)}), ("f18_conf", 18, {"conf>=60": (F[:, 18] >= 60), "conf>=70": (F[:, 18] >= 70), "conf>=80": (F[:, 18] >= 80)}), ("f10_eqh", 10, {"eqh_on": (F[:, 10] == 1)}), ("f11_eql", 11, {"eql_on": (F[:, 11] == 1)}), ): for cname, mask in conds.items(): if not mask.any(): continue r = dict(feature=nm, condition=cname, n=int(mask.sum()), p_pos=float((y[mask] == 1).mean()), p_neg=float((y[mask] == -1).mean()), lift_pos=float((y[mask] == 1).mean() / max(p1_uncond, 1e-12))) rows.append(r) print(f" {nm:11s} {cname:12s} n={r['n']:>6d} P(+1)={r['p_pos']:.4f} " f"P(-1)={r['p_neg']:.4f} lift={r['lift_pos']:.3f}") report["label_x_smc_state"] = {"unconditional_p_pos": float(p1_uncond), "rows": rows} P3.save_json("label_audit.json", report) print("\nLabel audit selesai. Output: ml/p3/output/label_audit.json") if __name__ == "__main__": main()