146 lines
5 KiB
Python
146 lines
5 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""P3.1e REGIME DIAGNOSTIC — SniperGold_ML.
|
|
|
|
Diagnostic ONLY. Answers the fifth research question:
|
|
"Apakah relationship feature -> label berubah berdasarkan market state?"
|
|
|
|
Regime proxies use EXISTING semantics only (no new regime model):
|
|
* ATR percentile : rolling percentile rank of ATR over trailing 500 bars
|
|
* Realized vol : std of 20-bar log returns (trailing)
|
|
* Trend proxy : |f16 mom20_atr| magnitude (already a contract feature)
|
|
* Range proxy : f17 range_atr (already a contract feature)
|
|
|
|
Analysis:
|
|
* P(Y=1) / P(Y=-1) per regime tercile (low/med/high)
|
|
* feature x regime: univariate AUC of key features within each tercile
|
|
-> does the feature->label relationship change across regimes?
|
|
* regime x SMC state: P(Y=1) for selected combos
|
|
|
|
No MS-GARCH / HMM / regime gate is built.
|
|
"""
|
|
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
|
|
|
|
KEY_FEATS = [0, 5, 7, 9, 16, 18]
|
|
|
|
|
|
def terciles(x):
|
|
q33, q66 = np.quantile(x, [1 / 3, 2 / 3])
|
|
lo = x <= q33
|
|
hi = x >= q66
|
|
md = ~lo & ~hi
|
|
return lo, md, hi
|
|
|
|
|
|
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)
|
|
|
|
# ---- regime proxies ----
|
|
atr_pct = np.zeros(n)
|
|
for i in range(500, n):
|
|
atr_pct[i] = (A[i - 500:i] <= A[i]).mean()
|
|
atr_pct[:500] = 0.5
|
|
|
|
lr = np.zeros(n)
|
|
lr[1:] = np.log(c[1:] / np.maximum(c[:-1], 1e-12))
|
|
rv = np.zeros(n)
|
|
for i in range(20, n):
|
|
rv[i] = lr[i - 19:i + 1].std(ddof=1)
|
|
rv[:20] = np.nan
|
|
rv = np.nan_to_num(rv, nan=np.nanmedian(rv[20:]))
|
|
|
|
proxies = {
|
|
"ATR_percentile": atr_pct,
|
|
"RealizedVol20": rv,
|
|
"TrendAbs_mom20(f16)": np.abs(F[:, 16]),
|
|
"RangeATR(f17)": F[:, 17],
|
|
}
|
|
|
|
report = {"provenance": P3.provenance(),
|
|
"dataset_hash": P3.dataset_hash(F, lab, A, t),
|
|
"regimes": {}}
|
|
|
|
print("=== REGIME DIAGNOSTIC (tercile low/med/high) ===")
|
|
for pname, px in proxies.items():
|
|
lo, md, hi = terciles(px)
|
|
r = {}
|
|
print(f"\n[{pname}]")
|
|
for rname, m in (("low", lo), ("med", md), ("high", hi)):
|
|
ml = m[midx]
|
|
if ml.sum() < 100:
|
|
continue
|
|
p1 = float(ypos[ml].mean())
|
|
pm1 = float((y[ml] == -1).mean())
|
|
r[rname] = {"n": int(ml.sum()), "P1": p1, "Pm1": pm1,
|
|
"lift_P1": float(p1 / ypos.mean())}
|
|
print(f" {rname:>4}: n={int(ml.sum()):>6} P(+1)={p1:.4f} "
|
|
f"P(-1)={pm1:.4f} lift={p1 / ypos.mean():.3f}")
|
|
for j in KEY_FEATS:
|
|
xs = F[midx, j][ml]
|
|
if len(np.unique(xs)) < 3:
|
|
continue
|
|
a = TM.auc(ypos[ml].astype(int), xs)
|
|
r[rname]["auc_" + P3.FEAT_NAMES[j]] = float(a)
|
|
print(" univariate AUC by regime (long label):")
|
|
for fname in [P3.FEAT_NAMES[j] for j in KEY_FEATS]:
|
|
vals = []
|
|
for rname in ("low", "med", "high"):
|
|
vv = r.get(rname, {}).get("auc_" + fname)
|
|
vals.append(vv if vv is not None else float("nan"))
|
|
print(" %-16s low=%.4f med=%.4f high=%.4f" % (fname, vals[0], vals[1], vals[2]))
|
|
report["regimes"][pname] = r
|
|
|
|
# ---- regime x SMC state combos ----
|
|
print("\n=== REGIME x SMC STATE: P(Y=1) selected combos ===")
|
|
atr_lo, atr_md, atr_hi = terciles(atr_pct)
|
|
f9 = F[:, 9]
|
|
f7 = F[:, 7]
|
|
f18 = F[:, 18]
|
|
combos = {
|
|
"BUY_cand & ATR_low": (f9 == 1) & (f7 > 0) & atr_lo,
|
|
"BUY_cand & ATR_high": (f9 == 1) & (f7 > 0) & atr_hi,
|
|
"SELL_cand & ATR_low": (f9 == 1) & (f7 < 0) & atr_lo,
|
|
"SELL_cand & ATR_high": (f9 == 1) & (f7 < 0) & atr_hi,
|
|
"conf>=60 & ATR_low": (f18 >= 60) & atr_lo,
|
|
"conf>=60 & ATR_high": (f18 >= 60) & atr_hi,
|
|
"MTF_bull & ATR_low": (F[:, 0] > 0) & (F[:, 1] > 0) & (F[:, 2] > 0) & atr_lo,
|
|
"MTF_bull & ATR_high": (F[:, 0] > 0) & (F[:, 1] > 0) & (F[:, 2] > 0) & atr_hi,
|
|
}
|
|
comb = {}
|
|
for cname, mask in combos.items():
|
|
ml = mask[midx]
|
|
if ml.sum() < 100:
|
|
comb[cname] = {"n": int(ml.sum()), "note": "too few"}
|
|
print(f" {cname:<24} n={int(ml.sum())} (too few)")
|
|
continue
|
|
p1 = float(ypos[ml].mean())
|
|
pm1 = float((y[ml] == -1).mean())
|
|
comb[cname] = {"n": int(ml.sum()), "P1": p1, "Pm1": pm1,
|
|
"lift_P1": float(p1 / ypos.mean())}
|
|
print(f" {cname:<24} n={int(ml.sum()):>6} P(+1)={p1:.4f} "
|
|
f"P(-1)={pm1:.4f} lift={p1 / ypos.mean():.3f}")
|
|
report["regime_x_smc"] = comb
|
|
|
|
P3.save_json("regime_diagnostic.json", report)
|
|
print("\nRegime diagnostic selesai. Output: ml/p3/output/regime_diagnostic.json")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|