SniperGold_ML/ml/p3/candidate_setup_audit.py

# -*- 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()