SniperGold_ML/ml/diag_returns.py

# -*- coding: utf-8 -*-
"""Diagnosa distribusi return M15 XAUUSD — cari outlier & penyebab GARCH/HMM kolaps."""
import os
import numpy as np
import datetime as dt

BASE = os.path.normpath(os.path.join(os.path.dirname(os.path.abspath(__file__)),
                                     "..", "..", "..", "Files", "AlgoForge", "Data"))
z = np.load(os.path.join(BASE, "XAUUSD_M15.npz"))
t = z["time"].astype(np.int64)
c = z["close"].astype(np.float64)

r = np.zeros(len(c))
r[1:] = np.log(np.maximum(c[1:], 1e-12) / np.maximum(c[:-1], 1e-12))

def stats(name, rr, tt):
    print(f"--- {name}: n={len(rr)}")
    print(f"  mean={rr.mean():+.6f} std={rr.std():.6f} min={rr.min():.6f} max={rr.max():.6f}")
    for q in (0.001, 0.01, 0.05, 0.5, 0.95, 0.99, 0.999):
        print(f"  q{q:.3f}={np.quantile(rr, q):+.6f}", end="")
    print()
    n_ext = int((np.abs(rr) > 0.01).sum())
    print(f"  |r|>1%: {n_ext} | |r|>5%: {int((np.abs(rr) > 0.05).sum())} "
          f"| |r|>10%: {int((np.abs(rr) > 0.10).sum())}")
    if n_ext:
        idx = np.where(np.abs(rr) > 0.01)[0]
        for i in idx[:8]:
            print(f"    outlier t={dt.datetime.fromtimestamp(int(tt[i]), dt.timezone.utc)} "
                  f"r={rr[i]:+.4f} close={c[i]:.2f}->{c[min(i+1, len(c)-1)]:.2f}")

stats("SEMUA", r, t)
m2017 = t < dt.datetime(2018, 1, 1).timestamp()
stats("2017 (warmup, sparse)", r[m2017], t[m2017])
md = (t >= dt.datetime(2018, 1, 1).timestamp()) & (t < dt.datetime(2024, 8, 1).timestamp())
stats("2018..2024-07 (train dense)", r[md], t[md])
mtest = t >= dt.datetime(2024, 8, 1).timestamp()
stats("2024-08.. (test dense)", r[mtest], t[mtest])

# gap bars: return antar bar dgn jeda waktu > 30 menit (weekend/gap)
d = np.diff(t)
gap_idx = np.where(d > 1800)[0]
print(f"bar dgn gap >30m: {len(gap_idx)}")
if len(gap_idx):
    gr = r[gap_idx + 1]
    print(f"  return setelah gap: std={gr.std():.6f} |r|>1%: {int((np.abs(gr) > 0.01).sum())}")