45 lines
1.9 KiB
Python
45 lines
1.9 KiB
Python
|
# -*- 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())}")
|