395 lines
16 KiB
Python
395 lines
16 KiB
Python
# -*- 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()
|