155 lines
6.1 KiB
Python
155 lines
6.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""P3.1c CANDIDATE-SETUP CONDITIONING AUDIT — SniperGold_ML.
|
|
|
|
Diagnostic ONLY. Answers:
|
|
"Apakah information density naik ketika ML hanya diberi SMC candidate setups?"
|
|
|
|
Design (section 12/13 of the P3 protocol):
|
|
D0 = ALL CLOSED BARS (labeled subset)
|
|
D1 = SMC CANDIDATE SETUP BARS ONLY
|
|
|
|
Candidate strata are defined ONLY from EXISTING feature semantics
|
|
(FEATURE_CONTRACT v1.0) as diagnostic proxies — no new setup rule is created
|
|
and nothing is promoted to production.
|
|
|
|
The SAME frozen P2.6 model (SniperGold_ML_p26_corrected.mqh) is evaluated on
|
|
the purged TEST split of each stratum. AUC(D1) vs AUC(D0) is the comparison;
|
|
class balance / entropy / outcome rate / MFE / MAE are reported alongside.
|
|
|
|
No retraining, no tuning, no threshold optimization.
|
|
"""
|
|
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
|
|
|
|
|
|
def strata_masks(F):
|
|
"""Candidate strata from existing feature semantics (diagnostic proxies)."""
|
|
f0, f1, f2 = F[:, 0], F[:, 1], F[:, 2]
|
|
f5, f7, f8, f9 = F[:, 5], F[:, 7], F[:, 8], F[:, 9]
|
|
f10, f11, f18 = F[:, 10], F[:, 11], F[:, 18]
|
|
sgn = np.sign
|
|
mtf_signs = np.column_stack([sgn(f0), sgn(f1), sgn(f2)])
|
|
mtf_all = (f0 != 0) & (f1 != 0) & (f2 != 0)
|
|
aligned = mtf_all & (np.all(mtf_signs == 1, axis=1) | np.all(mtf_signs == -1, axis=1))
|
|
conflicting = mtf_all & ~aligned
|
|
return {
|
|
"D0_all_bars": np.ones(len(F), dtype=bool),
|
|
"D1_any_event": (f7 != 0) | (f8 != 0) | (f9 == 1) | (f10 == 1) | (f11 == 1),
|
|
"BUY_candidate": (f9 == 1) & (f7 > 0),
|
|
"SELL_candidate": (f9 == 1) & (f7 < 0),
|
|
"BUY_strict": (f9 == 1) & (f7 > 0) & (f18 >= 60),
|
|
"SELL_strict": (f9 == 1) & (f7 < 0) & (f18 >= 60),
|
|
"MTF_aligned": aligned,
|
|
"MTF_conflicting": conflicting,
|
|
"HTF_bull": (f0 > 0) & (f1 > 0) & (f2 > 0),
|
|
"HTF_bear": (f0 < 0) & (f1 < 0) & (f2 < 0),
|
|
"Fuzzy_high": f18 >= 60,
|
|
"Fuzzy_low": f18 < 40,
|
|
"EQH_event": f10 == 1,
|
|
"EQL_event": f11 == 1,
|
|
}
|
|
|
|
|
|
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) # base 24-bar / 0.75 ATR
|
|
|
|
labeled = lab != 0
|
|
midx = np.where(labeled)[0]
|
|
y = lab[midx]
|
|
ypos = (y == 1).astype(int)
|
|
|
|
# ---- purged split (identical to P2.6) ----
|
|
tr_idx, te_idx, pg_idx, split_bar = P3.purged_split_idx(midx)
|
|
te_m = np.isin(midx, te_idx)
|
|
te_abs = midx[te_m] # absolute bar indices in test
|
|
yte = y[te_m]
|
|
yte_pos = (yte == 1).astype(int)
|
|
|
|
# ---- frozen model predictions on test ----
|
|
M = P3.parse_mqh_arrays()
|
|
Xte_raw = F[te_abs]
|
|
pL, pS = P3.frozen_forward(M, Xte_raw)
|
|
|
|
# ---- label decomposition on test bars (H=24) ----
|
|
dec = P3.label_decompose(c, h, l, A, te_abs, horizon=TM.H_LABEL,
|
|
thr_mult=TM.LABEL_ATR)
|
|
|
|
masks = strata_masks(F)
|
|
report = {"provenance": P3.provenance(),
|
|
"dataset_hash": P3.dataset_hash(F, lab, A, t),
|
|
"label": {"horizon": TM.H_LABEL, "thr_mult": TM.LABEL_ATR},
|
|
"test_n": int(len(te_abs)),
|
|
"test_auc_D0_long": float(TM.auc(yte_pos.astype(int), pL)),
|
|
"test_auc_D0_short": float(TM.auc((yte == -1).astype(int), pS)),
|
|
"strata": {}}
|
|
|
|
print("=== CANDIDATE-SETUP CONDITIONING (frozen P2.6 MLP, TEST split) ===")
|
|
print(f"D0 TEST: n={len(te_abs)} AUC_LONG={report['test_auc_D0_long']:.4f} "
|
|
f"AUC_SHORT={report['test_auc_D0_short']:.4f} "
|
|
f"(P2.6 offline 0.5105/0.5063)")
|
|
print(f"{'stratum':<18} {'n_full':>8} {'n_test':>7} {'P(+1)':>7} {'ent':>6} "
|
|
f"{'MFE':>6} {'MAE':>6} {'AUC_L':>7} {'AUC_S':>7} {'dAUC_L':>7}")
|
|
for name, mask in masks.items():
|
|
n_full = int(mask.sum())
|
|
mask_te = mask[te_abs]
|
|
n_te = int(mask_te.sum())
|
|
if n_te < 50:
|
|
print(f"{name:<18} {n_full:>8} {n_te:>7} (too few for AUC)")
|
|
report["strata"][name] = {"n_full": n_full, "n_test": n_te,
|
|
"note": "too few for AUC"}
|
|
continue
|
|
ysub = yte[mask_te]
|
|
p1 = float((ysub == 1).mean())
|
|
pm1 = float((ysub == -1).mean())
|
|
pz = 0.0
|
|
pvals = [p for p in (p1, pm1, pz) if p > 0]
|
|
ent = float(-sum(p * np.log(p) for p in pvals))
|
|
mfe = float(np.nanmedian(dec["mfe"][mask_te]))
|
|
mae = float(np.nanmedian(dec["mae"][mask_te]))
|
|
auc_l = float(TM.auc((ysub == 1).astype(int), pL[mask_te]))
|
|
auc_s = float(TM.auc((ysub == -1).astype(int), pS[mask_te]))
|
|
d_l = auc_l - report["test_auc_D0_long"]
|
|
print(f"{name:<18} {n_full:>8} {n_te:>7} {p1:>7.4f} {ent:>6.3f} "
|
|
f"{mfe:>6.2f} {mae:>6.2f} {auc_l:>7.4f} {auc_s:>7.4f} {d_l:>+7.4f}")
|
|
report["strata"][name] = {"n_full": n_full, "n_test": n_te,
|
|
"p_pos": p1, "p_neg": pm1,
|
|
"entropy_labeled": ent,
|
|
"mfe_med_atr": mfe, "mae_med_atr": mae,
|
|
"auc_long": auc_l, "auc_short": auc_s,
|
|
"d_auc_long_vs_D0": d_l}
|
|
|
|
# ---- event frequency / density (full series) ----
|
|
print("\n=== EVENT FREQUENCY & DENSITY (full series) ===")
|
|
freq = {}
|
|
for name, mask in masks.items():
|
|
on = mask
|
|
if on.any():
|
|
idx_on = np.where(on)[0]
|
|
gaps = np.diff(idx_on)
|
|
freq[name] = {"rate": float(on.mean()),
|
|
"n": int(on.sum()),
|
|
"median_gap_bars": float(np.median(gaps)) if len(gaps) else None}
|
|
for name, v in freq.items():
|
|
print(f" {name:<18} rate={v['rate']:.4f} n={v['n']:>7} "
|
|
f"med_gap={v['median_gap_bars']}")
|
|
report["event_frequency"] = freq
|
|
|
|
P3.save_json("candidate_setup_audit.json", report)
|
|
print("\nCandidate-setup audit selesai. Output: ml/p3/output/candidate_setup_audit.json")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|