# -*- coding: utf-8 -*- """P3.1d TEMPORAL DIAGNOSTIC — SniperGold_ML. Diagnostic ONLY. Answers the fourth research question: "Apakah informasi yang hilang pada static feature memang bersifat temporal?" No neural sequence model is used. Evidence built from: * feature autocorrelation (lags 1..10) for continuous features * state persistence & transition matrices for discrete features * run-length of persistent states * LAGGED INFORMATION: univariate AUC of feature[t-k] vs label[t] (k = 0..5) -> if delayed information is retained, temporal dependency exists * label conditional persistence: P(Y_t=1 | Y_{t-k}=1) for k = 1..24 * state-age analysis: does P(Y=1) change with the age of the current HTF-bias / swing-trend run? (i.e., is "state age" itself informative?) All metrics are descriptive; nothing is promoted to a model. """ 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 CONT_FEATURES = [6, 13, 14, 15, 16, 17, 18] DISC_FEATURES = [0, 1, 2, 3, 4, 5, 7, 8, 9, 10, 11, 12] LAGS = [1, 2, 3, 5, 8, 12, 16, 24] 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) labeled = lab != 0 midx = np.where(labeled)[0] y = lab[midx] ypos = (y == 1).astype(int) report = {"provenance": P3.provenance(), "dataset_hash": P3.dataset_hash(F, lab, A, t)} # ---- 1) feature autocorrelation (continuous) ---- print("=== FEATURE AUTOCORRELATION (continuous, lags 1..10) ===") ac = {} for j in CONT_FEATURES: x = F[:, j].astype(float) xc = x - x.mean() var = (xc * xc).mean() row = {} for k in range(1, 11): row[str(k)] = float((xc[:-k] * xc[k:]).mean() / var) if var > 0 else 0.0 ac[j] = row print(f" {P3.FEAT_NAMES[j]:<16s} " + " ".join(f"l{k}:{row[str(k)]:+.3f}" for k in (1, 3, 5, 10))) report["feature_autocorr"] = {P3.FEAT_NAMES[j]: ac[j] for j in CONT_FEATURES} # ---- 2) discrete state persistence ---- print("\n=== DISCRETE STATE PERSISTENCE (P(state_t == state_{t-1})) ===") dp = {} for j in DISC_FEATURES: xs = F[:, j] dp[j] = float((xs[1:] == xs[:-1]).mean()) print(f" {P3.FEAT_NAMES[j]:<16s} persist={dp[j]:.4f}") report["discrete_persistence"] = {P3.FEAT_NAMES[j]: dp[j] for j in DISC_FEATURES} # ---- 3) run lengths of persistent states ---- print("\n=== RUN LENGTH (bars) of non-zero states ===") rl = {} for j in (0, 1, 2, 3, 4, 5, 7, 8): xs = F[:, j] runs = [] cur = 0 curv = 0 for vv in xs: if vv != 0 and vv == curv: cur += 1 else: if cur: runs.append(cur) cur = 1 if vv != 0 else 0 curv = vv if cur: runs.append(cur) rl[j] = {"mean": float(np.mean(runs)) if runs else 0.0, "median": float(np.median(runs)) if runs else 0.0, "p95": float(np.percentile(runs, 95)) if runs else 0.0, "max": float(np.max(runs)) if runs else 0.0} print(f" {P3.FEAT_NAMES[j]:<16s} run mean={rl[j]['mean']:.1f} " f"med={rl[j]['median']:.0f} p95={rl[j]['p95']:.0f} max={rl[j]['max']:.0f}") report["run_length"] = {P3.FEAT_NAMES[j]: rl[j] for j in rl} # ---- 4) LAGGED INFORMATION: AUC(feature[t-k], label[t]) ---- print("\n=== LAGGED INFORMATION: univariate AUC(feature[t-k] vs label[t]) ===") key_feats = [0, 5, 7, 9, 16, 18] lag_auc = {} for j in key_feats: row = {} for k in LAGS: # feature value k bars before each labeled bar xk = F[midx - k, j] valid = midx - k >= 0 if valid.sum() < 1000: row[str(k)] = None continue row[str(k)] = float(TM.auc(ypos[valid].astype(int), xk[valid])) lag_auc[j] = row print(f" {P3.FEAT_NAMES[j]:<16s} " + " ".join(f"k{k}:{(row[str(k)] if row[str(k)] is not None else float('nan')):.4f}" for k in LAGS)) report["lagged_auc"] = {P3.FEAT_NAMES[j]: lag_auc[j] for j in key_feats} # ---- 5) label conditional persistence ---- print("\n=== LABEL CONDITIONAL PERSISTENCE: P(Y_t=1 | Y_{t-k}=1) ===") lcp = {} li = midx # labeled bar indices (ascending) yv = y p1_uncond = float((yv == 1).mean()) for k in (1, 2, 3, 5, 8, 12, 16, 24): # find pairs (i, i-k) both labeled with Y_{i-k}=1 # labeled bars are not contiguous; use absolute time index mapping pos_idx = li[yv == 1] n_prev = 0 n_same = 0 # vectorized: for each labeled bar, check if bar-k bars ago was labeled +1 valid = li - k >= 0 prev_lab = np.full(len(li), 0) # map bar index -> label labmap = np.zeros(n, dtype=int) labmap[li] = yv prev = labmap[li - k] n_prev = int((prev == 1).sum()) n_same = int(((prev == 1) & (yv == 1)).sum()) p = float(n_same / n_prev) if n_prev else float("nan") lcp[str(k)] = {"n_prev_pos": n_prev, "P_same": p, "unconditional_P1": float(p1_uncond), "lift": float(p / p1_uncond) if n_prev else float("nan")} print(f" k={k:>2}: P(Y_t=1 | Y_{{t-{k}}}=1)={p:.4f} (uncond {p1_uncond:.4f}, " f"lift={p / p1_uncond:.3f})") report["label_cond_persistence"] = lcp # ---- 6) STATE-AGE ANALYSIS: P(Y=1) vs age of current run ---- print("\n=== STATE-AGE ANALYSIS: P(Y=1) by age of current HTF-bias run ===") age = {} for j in (0, 1, 5): xs = F[:, j] # run age at each bar (bars since run started) age_arr = np.zeros(n, dtype=int) cur_age = 0 prev = 0 for i in range(n): if xs[i] != 0 and xs[i] == prev: cur_age += 1 else: cur_age = 1 if xs[i] != 0 else 0 age_arr[i] = cur_age prev = xs[i] # P(Y=1 | age bucket) on labeled bars buckets = [(1, 3), (4, 10), (11, 30), (31, 100), (101, 10 ** 9)] rows = [] for lo, hi in buckets: m = (age_arr[midx] >= lo) & (age_arr[midx] <= hi) if m.sum() < 100: continue rows.append({"age_bucket": f"{lo}-{hi}", "n": int(m.sum()), "P1": float(ypos[m].mean())}) print(f" {P3.FEAT_NAMES[j]:<16s} age {lo:>3}-{hi:<4}: " f"P(Y=1)={ypos[m].mean():.4f} n={int(m.sum())}") age[j] = rows report["state_age"] = {P3.FEAT_NAMES[j]: age[j] for j in age} P3.save_json("temporal_diagnostic.json", report) print("\nTemporal diagnostic selesai. Output: ml/p3/output/temporal_diagnostic.json") if __name__ == "__main__": main()