377 lines
15 KiB
Python
377 lines
15 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""P3.1a FEATURE INFORMATION AUDIT — SniperGold_ML.
|
|
|
|
Diagnostic ONLY. Answers the first research question:
|
|
"Apakah 19 feature runtime-consistent memiliki information content terhadap
|
|
label 24-bar / 0.75 x ATR saat ini?"
|
|
|
|
Outputs (per feature):
|
|
* descriptive stats: mean, std, min, max, NaN/Inf, unique, zero-rate,
|
|
variance, percentiles, distribution
|
|
* label-conditioned stats (positive vs negative label): mean diff, median
|
|
diff, point-biserial, rank-biserial, Cohen's d, univariate AUC, MI
|
|
* classification: INFORMATIVE / WEAK / NEAR-DEGENERATE / CONSTANT / UNKNOWN
|
|
|
|
Special audits:
|
|
* F10/F11 EQH/EQL (frequency, time distribution, label conditioning)
|
|
* HTF f0-f2 + f18 (direction balance, transition, run length)
|
|
* Structure f3-f9, f12-f17 (event frequency, sparsity, clustering, label assoc)
|
|
|
|
Feature-group diagnostic (section 17) uses the SAME per-feature metrics,
|
|
aggregated per family — a diagnostic proxy, NOT feature selection.
|
|
|
|
No model training, no threshold tuning, no AUC hunting.
|
|
"""
|
|
import os
|
|
import sys
|
|
import json
|
|
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
|
|
|
|
H_LABEL = TM.H_LABEL
|
|
LABEL_ATR = TM.LABEL_ATR
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# metrics
|
|
# ----------------------------------------------------------------------
|
|
def point_biserial(x, y):
|
|
"""y binary 0/1."""
|
|
if np.ptp(x) == 0:
|
|
return 0.0
|
|
return float(np.corrcoef(x, y)[0, 1])
|
|
|
|
|
|
def cohens_d(x1, x0):
|
|
n1, n0 = len(x1), len(x0)
|
|
if n1 < 2 or n0 < 2:
|
|
return 0.0
|
|
s1, s0 = x1.std(ddof=1), x0.std(ddof=1)
|
|
sp = np.sqrt(((n1 - 1) * s1 ** 2 + (n0 - 1) * s0 ** 2) / (n1 + n0 - 2))
|
|
if sp <= 0:
|
|
return 0.0
|
|
return float((x1.mean() - x0.mean()) / sp)
|
|
|
|
|
|
def rank_biserial(x, y):
|
|
"""Mann-Whitney U based: r = 1 - 2*U/(n1*n0)."""
|
|
from scipy.stats import mannwhitneyu
|
|
x1 = x[y == 1]
|
|
x0 = x[y == 0]
|
|
if len(x1) == 0 or len(x0) == 0:
|
|
return 0.0
|
|
try:
|
|
U, _ = mannwhitneyu(x1, x0, alternative="two-sided")
|
|
except ValueError:
|
|
return 0.0
|
|
return float(1.0 - 2.0 * U / (len(x1) * len(x0)))
|
|
|
|
|
|
def mutual_info(x, y, bins=16):
|
|
"""Histogram MI between continuous x and binary y (nats)."""
|
|
m = len(x)
|
|
if m < bins * 10:
|
|
return 0.0
|
|
q = np.quantile(x, np.linspace(0, 1, bins + 1))
|
|
q[0] -= 1e-12
|
|
q[-1] += 1e-12
|
|
xd = np.clip(np.digitize(x, q[1:-1]), 0, bins - 1)
|
|
n1 = int(y.sum())
|
|
n0 = m - n1
|
|
if n1 == 0 or n0 == 0:
|
|
return 0.0
|
|
p1, p0 = n1 / m, n0 / m
|
|
mi = 0.0
|
|
for b in range(bins):
|
|
nb = int((xd == b).sum())
|
|
if nb == 0:
|
|
continue
|
|
n1b = int((xd == b)[y == 1].sum())
|
|
n0b = nb - n1b
|
|
pb = nb / m
|
|
if n1b > 0:
|
|
mi += pb * (n1b / nb) * np.log((n1b / nb) / p1)
|
|
if n0b > 0:
|
|
mi += pb * (n0b / nb) * np.log((n0b / nb) / p0)
|
|
return float(max(mi, 0.0))
|
|
|
|
|
|
def univariate_auc(x, ypos):
|
|
"""AUC of feature x vs binary long label (ypos=1..0)."""
|
|
return TM.auc(ypos.astype(int), x)
|
|
|
|
|
|
def run_lengths(seq, value):
|
|
"""Mean/max run length of `value` in int seq; (0,0) if absent."""
|
|
runs, cur = [], 0
|
|
for v in seq:
|
|
if v == value:
|
|
cur += 1
|
|
else:
|
|
if cur:
|
|
runs.append(cur)
|
|
cur = 0
|
|
if cur:
|
|
runs.append(cur)
|
|
return (float(np.mean(runs)) if runs else 0.0,
|
|
float(np.max(runs)) if runs else 0.0)
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
# classification (diagnostic thresholds; effect sizes, not p-values)
|
|
# ----------------------------------------------------------------------
|
|
def classify(feat):
|
|
"""feat: dict with unique, zero_rate, auc_dev, r_pb, d, r_rb, mi."""
|
|
if feat["unique"] <= 1:
|
|
return "CONSTANT"
|
|
if feat["zero_rate"] >= 0.99 or feat["unique"] <= 2 and feat["zero_rate"] >= 0.95:
|
|
return "NEAR-DEGENERATE"
|
|
auc_dev = abs(feat["auc_dev"])
|
|
r = abs(feat["r_pb"])
|
|
d = abs(feat["d"])
|
|
rb = abs(feat["r_rb"])
|
|
strong = (auc_dev >= 0.02) or (r >= 0.02) or (rb >= 0.10) or (d >= 0.20)
|
|
weak = (auc_dev >= 0.008) or (r >= 0.008) or (rb >= 0.04) or (d >= 0.08)
|
|
if strong:
|
|
return "INFORMATIVE"
|
|
if weak:
|
|
return "WEAK"
|
|
if feat["unique"] <= 3:
|
|
return "NEAR-DEGENERATE"
|
|
return "UNKNOWN"
|
|
|
|
|
|
# ----------------------------------------------------------------------
|
|
def main():
|
|
t, o, h, l, c, v, htf = P3.load_data()
|
|
F = P3.load_F()
|
|
A = P3.atr_series(h, l, c)
|
|
lab = P3.make_label(c, A)
|
|
n = len(c)
|
|
|
|
labeled = lab != 0
|
|
midx = np.where(labeled)[0]
|
|
y = lab[midx]
|
|
ypos = (y == 1).astype(int)
|
|
X = F[midx]
|
|
print(f"bars={n} labeled={len(midx)} pos={int(ypos.sum())} neg={int((ypos == 0).sum())}")
|
|
|
|
hours = np.array([dt.datetime.fromtimestamp(int(tt), dt.timezone.utc).hour
|
|
for tt in t])
|
|
years = np.array([dt.datetime.fromtimestamp(int(tt), dt.timezone.utc).year
|
|
for tt in t])
|
|
|
|
report = {"provenance": P3.provenance(),
|
|
"dataset_hash": P3.dataset_hash(F, lab, A, t),
|
|
"label": {"horizon": H_LABEL, "thr_mult": LABEL_ATR,
|
|
"n_labeled": int(len(midx)),
|
|
"n_pos": int(ypos.sum()),
|
|
"n_neg": int((ypos == 0).sum())},
|
|
"features": [], "groups": {}, "special": {}}
|
|
|
|
print("\n=== PER-FEATURE DESCRIPTIVE + INFORMATION AUDIT ===")
|
|
hdr = (f"{'id':>2} {'name':<16} {'mean':>8} {'std':>8} {'min':>9} {'max':>9} "
|
|
f"{'nan':>4} {'inf':>4} {'uniq':>6} {'zero%':>6} {'auc':>7} {'r_pb':>7} "
|
|
f"{'d':>6} {'r_rb':>7} {'mi':>7} class")
|
|
print(hdr)
|
|
for j, nm in enumerate(P3.FEAT_NAMES):
|
|
x = F[:, j]
|
|
xs = X[:, j]
|
|
desc = {
|
|
"id": j, "name": nm,
|
|
"mean": float(x.mean()), "std": float(x.std(ddof=1)),
|
|
"min": float(x.min()), "max": float(x.max()),
|
|
"nan": int(np.isnan(x).sum()), "inf": int(np.isinf(x).sum()),
|
|
"unique": int(np.unique(x).size),
|
|
"zero_rate": float((x == 0).mean()),
|
|
"var": float(x.var()),
|
|
"pct": {q: float(np.quantile(x, q)) for q in (0.01, 0.05, 0.25, 0.5, 0.75, 0.95, 0.99)},
|
|
}
|
|
r_pb = point_biserial(xs, ypos)
|
|
d = cohens_d(xs[ypos == 1], xs[ypos == 0])
|
|
r_rb = rank_biserial(xs, ypos)
|
|
aucv = univariate_auc(xs, ypos)
|
|
mi = mutual_info(xs, ypos)
|
|
info = {"auc": float(aucv), "auc_dev": float(aucv - 0.5),
|
|
"r_pb": float(r_pb), "d": float(d), "r_rb": float(r_rb),
|
|
"mi": float(mi),
|
|
"mean_pos": float(xs[ypos == 1].mean()),
|
|
"mean_neg": float(xs[ypos == 0].mean()),
|
|
"median_pos": float(np.median(xs[ypos == 1])),
|
|
"median_neg": float(np.median(xs[ypos == 0]))}
|
|
cls = classify({**desc, **info})
|
|
desc.update(info)
|
|
desc["class"] = cls
|
|
report["features"].append(desc)
|
|
print(f"{j:>2} {nm:<16} {desc['mean']:>8.4f} {desc['std']:>8.4f} "
|
|
f"{desc['min']:>9.3f} {desc['max']:>9.3f} "
|
|
f"{desc['nan']:>4} {desc['inf']:>4} {desc['unique']:>6} "
|
|
f"{desc['zero_rate']*100:>5.1f}% {info['auc']:>7.4f} {r_pb:>7.4f} "
|
|
f"{d:>6.3f} {r_rb:>7.4f} {mi:>7.4f} {cls}")
|
|
|
|
# ---- special audit: F10/F11 EQH/EQL ----
|
|
print("\n=== SPECIAL: F10/F11 EQH/EQL ===")
|
|
sp = {}
|
|
for j in (10, 11):
|
|
nm = P3.FEAT_NAMES[j]
|
|
x = F[:, j]
|
|
on = x == 1
|
|
n_on = int(on.sum())
|
|
freq = n_on / n
|
|
# year distribution
|
|
yr = {}
|
|
for yy in sorted(set(years)):
|
|
m = (years == yy)
|
|
yr[str(yy)] = {"n": int((on & m).sum()), "rate": float((on & m).mean())}
|
|
# hour distribution
|
|
hr = {}
|
|
for hh in range(24):
|
|
m = (hours == hh)
|
|
hr[str(hh)] = float((on & m).mean())
|
|
# label conditioning on labeled subset
|
|
on_l = on[midx]
|
|
p1_on = float(ypos[on_l].mean()) if on_l.any() else float("nan")
|
|
p1_off = float(ypos[~on_l].mean()) if (~on_l).any() else float("nan")
|
|
n_on_l = int(on_l.sum())
|
|
n_off_l = int((~on_l).sum())
|
|
# co-occurrence with other SMC events (f7 sweep, f8 choch, f9 confirm)
|
|
co = {}
|
|
for k in (7, 8, 9):
|
|
co[P3.FEAT_NAMES[k]] = float((F[on, k] != 0).mean()) if n_on else 0.0
|
|
ent = {"name": nm, "n_on": n_on, "freq": freq,
|
|
"year_dist": yr, "hour_rate": hr,
|
|
"label": {"P1_on": p1_on, "P1_off": p1_off,
|
|
"n_on_l": n_on_l, "n_off_l": n_off_l},
|
|
"cooccurrence": co,
|
|
"unconditional_P1": float(ypos.mean())}
|
|
sp[j] = ent
|
|
print(f" {nm}: freq={freq:.4f} n_on={n_on} | P1(on)={p1_on:.4f} "
|
|
f"P1(off)={p1_off:.4f} (uncond {ypos.mean():.4f}) "
|
|
f"| cooccur sweep/choch/confirm={co}")
|
|
report["special"]["eqh_eql"] = sp
|
|
|
|
# ---- special audit: HTF f0-f2 + f18 ----
|
|
print("\n=== SPECIAL: HTF (f0-f2) + Confluence (f18) ===")
|
|
htf_aud = {}
|
|
for j in (0, 1, 2, 18):
|
|
nm = P3.FEAT_NAMES[j]
|
|
x = F[:, j]
|
|
if j <= 2:
|
|
vp = int((x == 1).sum()) / n
|
|
vn = int((x == -1).sum()) / n
|
|
v0 = int((x == 0).sum()) / n
|
|
# transition matrix (labeled subset order preserved via midx)
|
|
xs = x[midx]
|
|
tr = {}
|
|
for a in (-1, 0, 1):
|
|
row = {}
|
|
for b in (-1, 0, 1):
|
|
mask = (xs[:-1] == a) & (xs[1:] == b)
|
|
row[str(b)] = int(mask.sum())
|
|
tr[str(a)] = row
|
|
persist = float((xs[1:] == xs[:-1]).mean())
|
|
rl1 = run_lengths(xs, 1)
|
|
rlm1 = run_lengths(xs, -1)
|
|
# label conditioning
|
|
p1_pos = float(ypos[xs == 1].mean()) if (xs == 1).any() else float("nan")
|
|
p1_neg = float(ypos[xs == -1].mean()) if (xs == -1).any() else float("nan")
|
|
p1_zero = float(ypos[xs == 0].mean()) if (xs == 0).any() else float("nan")
|
|
ent = {"name": nm, "p_pos": vp, "p_neg": vn, "p_zero": v0,
|
|
"persistence": persist,
|
|
"run_len_pos": rl1, "run_len_neg": rlm1,
|
|
"transition": tr,
|
|
"label": {"P1_state+1": p1_pos, "P1_state-1": p1_neg,
|
|
"P1_state0": p1_zero,
|
|
"unconditional_P1": float(ypos.mean())}}
|
|
print(f" {nm}: +{vp:.3f} 0={v0:.3f} -{vn:.3f} persist={persist:.3f} "
|
|
f"| P1(+1)={p1_pos:.4f} P1(-1)={p1_neg:.4f} P1(0)={p1_zero:.4f}")
|
|
else:
|
|
xs = x[midx]
|
|
hi = xs >= 50
|
|
lo = xs < 25
|
|
midb = (xs >= 25) & (xs < 50)
|
|
ent = {"name": nm, "mean": float(x.mean()), "std": float(x.std(ddof=1)),
|
|
"frac_ge50": float(hi.mean()), "frac_lt25": float(lo.mean()),
|
|
"P1_ge50": float(ypos[hi].mean()) if hi.any() else float("nan"),
|
|
"P1_lt25": float(ypos[lo].mean()) if lo.any() else float("nan"),
|
|
"P1_25_50": float(ypos[midb].mean()) if midb.any() else float("nan"),
|
|
"unconditional_P1": float(ypos.mean())}
|
|
print(f" {nm}: mean={x.mean():.2f} frac>=50={hi.mean():.3f} "
|
|
f"| P1(>=50)={ent['P1_ge50']:.4f} P1(<25)={ent['P1_lt25']:.4f}")
|
|
htf_aud[j] = ent
|
|
report["special"]["htf_conf"] = htf_aud
|
|
|
|
# ---- special audit: structure features event analysis ----
|
|
print("\n=== SPECIAL: STRUCTURE FEATURES (f3-f9, f12-f17) ===")
|
|
struct_aud = {}
|
|
for j in list(range(3, 10)) + list(range(12, 18)):
|
|
nm = P3.FEAT_NAMES[j]
|
|
x = F[:, j]
|
|
active = x != 0
|
|
n_active = int(active.sum())
|
|
# event sparsity: mean gap between active bars (on full series)
|
|
idx_on = np.where(active)[0]
|
|
if len(idx_on) > 1:
|
|
gaps = np.diff(idx_on)
|
|
mean_gap = float(gaps.mean())
|
|
med_gap = float(np.median(gaps))
|
|
max_gap = float(gaps.max())
|
|
else:
|
|
mean_gap = med_gap = max_gap = float("nan")
|
|
# clustering: fraction of active bars within 5 bars of another active bar
|
|
clust = 0.0
|
|
if len(idx_on) > 1:
|
|
d5 = (np.diff(idx_on) <= 5).sum()
|
|
clust = float(d5 / max(1, len(idx_on) - 1))
|
|
# label association
|
|
act_l = active[midx]
|
|
p1_act = float(ypos[act_l].mean()) if act_l.any() else float("nan")
|
|
p1_inact = float(ypos[~act_l].mean()) if (~act_l).any() else float("nan")
|
|
ent = {"name": nm, "n_active": n_active, "active_rate": float(active.mean()),
|
|
"mean_gap": mean_gap, "med_gap": med_gap, "max_gap": max_gap,
|
|
"cluster5": clust,
|
|
"P1_active": p1_act, "P1_inactive": p1_inact,
|
|
"unconditional_P1": float(ypos.mean()),
|
|
"diff_P1": (p1_act - p1_inact) if not np.isnan(p1_act) and not np.isnan(p1_inact) else float("nan")}
|
|
struct_aud[j] = ent
|
|
print(f" {nm}: active={n_active} ({active.mean()*100:.2f}%) "
|
|
f"gap_mean={mean_gap:.1f} cluster5={clust:.3f} "
|
|
f"| P1(act)={p1_act:.4f} P1(inact)={p1_inact:.4f} "
|
|
f"diff={ent['diff_P1']:+.4f}")
|
|
report["special"]["structure"] = struct_aud
|
|
|
|
# ---- feature-group diagnostic (proxy) ----
|
|
print("\n=== FEATURE-GROUP DIAGNOSTIC (proxy, bukan selection) ===")
|
|
groups = {}
|
|
for gname, idxs in P3.FEAT_GROUPS.items():
|
|
rows = [report["features"][j] for j in idxs]
|
|
auc_devs = [abs(r["auc_dev"]) for r in rows]
|
|
r_pbs = [abs(r["r_pb"]) for r in rows]
|
|
mis = [r["mi"] for r in rows]
|
|
g = {"features": [P3.FEAT_NAMES[j] for j in idxs],
|
|
"max_abs_auc_dev": float(max(auc_devs)),
|
|
"mean_abs_auc_dev": float(np.mean(auc_devs)),
|
|
"max_abs_r_pb": float(max(r_pbs)),
|
|
"mean_abs_r_pb": float(np.mean(r_pbs)),
|
|
"sum_mi": float(np.sum(mis)),
|
|
"n_informative": int(sum(1 for r in rows if r["class"] == "INFORMATIVE"))}
|
|
groups[gname] = g
|
|
print(f" {gname:14s} max|auc-0.5|={g['max_abs_auc_dev']:.4f} "
|
|
f"mean|auc-0.5|={g['mean_abs_auc_dev']:.4f} "
|
|
f"max|r_pb|={g['max_abs_r_pb']:.4f} sum_mi={g['sum_mi']:.4f} "
|
|
f"n_inf={g['n_informative']}")
|
|
report["groups"] = groups
|
|
|
|
P3.save_json("feature_audit.json", report)
|
|
print("\nFeature audit selesai. Output: ml/p3/output/feature_audit.json")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|