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