forked from chiki2bum2/SniperGold_ML
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()
|