185 lines
6.9 KiB
Python
185 lines
6.9 KiB
Python
# -*- 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()
|