chiki2bum2/SniperGold_ML
Watch
1
0
Fork
You've already forked SniperGold_ML
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions
SniperGold_ML/ml/train_hybrid.py

560 lines
22 KiB
Python
Raw Permalink Normal View History

Algo Forge Tahap 4 sub-sesi 3 (FINAL No-Go): train_hybrid.py LSTM 2-lapis + 2-head (B1-B5 PASS) + gate like-for-like pd cache XAUUSDc -> dAUC LONG -0.0196 / SHORT -0.0134 vs baseline 0.627/0.621, CI negatif, 0/5 seed -> GATE No-Go per protokol G1-G7; baseline MLP TETAP deploy; ekspor .mqh hybrid dibatalkan
2026-08-21 14:16:57 +07:00
# -*- coding: utf-8 -*-
"""
Algo Forge - Tahap 4 sub-sesi 3: train_hybrid.py
==================================================
LSTM 2-LAPIS + 2-HEAD (P(long), P(short)) numpy murni, walk-forward multi-split,
multi-seed [42,7,123,2024,999], bootstrap paired CI 2000x, gate G1-G7
vs baseline MLP (long 0.6270 / short 0.6207) — like-for-like pd cache XAUUSDc.
Data:
- BARU : Files\\AlgoForge\\Data\\features_XAUUSD.npz (F19 + F2-6 regime,
window padat XAUUSD M15 2018+; split_bar=148470, split_pos=115631)
- BASELINE (read-only, gate) : Files\\SniperGold_ML\\features_XAUUSDc.npz
Mode:
--selftest : unit test B1-B5 (2-lapis forward/BPTT/gradcheck/determinisme/
anti-lookahead/overfit)
--calib : 1 seed pd subset train (kalibrasi waktu & sanity AUC)
--gate : like-for-like pd cache XAUUSDc (5 seed) vs 0.6270/0.6207
--wf : walk-forward pd data baru (75/25 + 70/30 + 80/20, 5 seed,
[F19] vs [F19+F2]) — mahal, jalankan bila budget cukup
Anti-lookahead: fitur/standarisasi train-only; sequence bar t hanya [t-W+1..t];
test disentuh SEKALI di akhir.
"""
import os
import sys
import math
import json
import time
import argparse
import datetime as dt
import numpy as np
HERE = os.path.dirname(os.path.abspath(__file__))
SRC = os.path.normpath(os.path.join(HERE, "..", "..", "SniperGold_ML"))
sys.path.insert(0, SRC)
import train_model as TM # noqa: E402 (fitur, auc, MLP baseline)
import train_lstm as TL # noqa: E402 (bootstrap_ci, build_sequences, head_forward dll)
DATA_NEW = os.path.normpath(os.path.join(HERE, "..", "..", "..", "Files",
"AlgoForge", "Data", "features_XAUUSD.npz"))
DATA_BASE = os.path.normpath(os.path.join(HERE, "..", "..", "..", "Files",
"SniperGold_ML", "features_XAUUSDc.npz"))
EVAL_LOG = os.path.join(HERE, "TAHAP4_HYBRID.log")
# ---- konfigurasi terkunci (mirip Fase 3, 2 lapis + 2 head) ----
W = 32
H = 32
LR = 1e-3
BATCH = 256
PATIENCE = 10
EPOCH_MAX = 30
CLIP = 1.0
SEEDS = [42, 7, 123, 2024, 999]
BASE_LONG, BASE_SHORT = 0.6270, 0.6207
CONFIG = dict(W=W, H=H, layers=2, heads=2, lr=LR, batch=BATCH, patience=PATIENCE,
epoch_max=EPOCH_MAX, clip=CLIP, seeds=SEEDS,
base_long=BASE_LONG, base_short=BASE_SHORT)
def elog(msg):
print(msg, flush=True)
with open(EVAL_LOG, "a", encoding="utf-8") as f:
f.write(msg + "\n")
def sigmoid(x):
return 1.0 / (1.0 + np.exp(-np.clip(x, -30.0, 30.0)))
# ======================================================================
# LSTM 2-LAPIS: FORWARD
# ======================================================================
def lstm_forward(Xs, Wx, Wh, bx, bh):
"""Satu lapis LSTM (reuse TL.lstm_forward). Xs: (B,T,D) -> hs: (B,T,H)."""
return TL.lstm_forward(Xs, Wx, Wh, bx, bh)
def lstm2_forward(Xs, W1, W2):
"""Dua lapis: L1(Xs)->h1s; L2(h1s)->h2s. W1/W2=(Wx,Wh,bx,bh)."""
Wx1, Wh1, bx1, bh1 = W1
Wx2, Wh2, bx2, bh2 = W2
h1s, c1 = lstm_forward(Xs, Wx1, Wh1, bx1, bh1)
h2s, c2 = lstm_forward(h1s, Wx2, Wh2, bx2, bh2)
return h2s, c1, c2, h1s
def head2_forward(hT, Wy, by):
"""2-target: (B,H)@(H,2)+b -> sigmoid (B,2)."""
return sigmoid(hT @ Wy + by)
# ======================================================================
# LSTM 2-LAPIS: BPTT
# ======================================================================
def lstm_layer_bptt(Xs, caches, dh_seq, Wx, Wh, bx, bh, want_dX=False):
"""BPTT satu lapis dgn gradien hidden per-timestep dh_seq[t] (B,H).
Return (dWx,dWh,dbx,dbh[, dX]) — dX = grad thd input tiap t."""
B, T, D = Xs.shape
Hh = Wh.shape[0]
Gx = Xs @ Wx + bx
dWx = np.zeros_like(Wx)
dWh = np.zeros_like(Wh)
dbx = np.zeros_like(bx)
dbh = np.zeros_like(bh)
dX = np.empty((B, T, D)) if want_dX else None
dh_next = np.zeros((B, Hh))
dc_next = np.zeros((B, Hh))
for t in reversed(range(T)):
dh = dh_seq[t] + dh_next
g, i, f, o, gg, cp, c, h = caches[t]
do = dh * np.tanh(c)
dc = dh * o * (1.0 - np.tanh(c) ** 2) + dc_next
di = dc * gg
df = dc * cp
dg = dc * i
dgates = np.concatenate([
di * i * (1.0 - i),
df * f * (1.0 - f),
do * o * (1.0 - o),
dg * (1.0 - gg ** 2),
], axis=1)
dWx += Xs[:, t].T @ dgates
dbx += dgates.sum(0)
hprev = caches[t - 1][-1] if t > 0 else np.zeros((B, Hh))
dWh += hprev.T @ dgates
dbh += dgates.sum(0)
if want_dX:
dX[:, t] = dgates @ Wx.T
dh_next = dgates @ Wh.T
dc_next = dc * f
out = (dWx, dWh, dbx, dbh)
return (out + (dX,)) if want_dX else out
def head2_backward(hT, P, Y, Wy):
"""BCE 2-target. Y: (B,2); loss = mean atas B*2 elemen. Return (dWy,dby,dhT)."""
B = P.shape[0]
d = (P - Y) / (B * 2)
dWy = hT.T @ d
dby = d.sum(0)
dhT = d @ Wy.T
return dWy, dby, dhT
def lstm2_bptt(Xs, h1s, c1, c2, dh2_last, W1, W2):
"""BPTT 2 lapis. h1s: (B,T,H) output L1 (input L2). dh2_last: grad head thd
h2[T-1]. Return (grads1, grads2)."""
Wx1, Wh1, bx1, bh1 = W1
Wx2, Wh2, bx2, bh2 = W2
T = Xs.shape[1]
B = Xs.shape[0]
Hh = c2[0][-1].shape[1]
# L2: dh_seq = 0 kecuali t=T-1 = dh2_last
dh2_seq = np.zeros((T, B, Hh))
dh2_seq[T - 1] = dh2_last
dWx2, dWh2, dbx2, dbh2, dX2 = lstm_layer_bptt(
h1s, c2, dh2_seq, Wx2, Wh2, bx2, bh2, want_dX=True)
# L1: dh_seq = dX2 (grad thd input L2 = h1) dgn urutan waktu (T,B,H)
dh1_seq = dX2.transpose(1, 0, 2)
g1 = lstm_layer_bptt(Xs, c1, dh1_seq, Wx1, Wh1, bx1, bh1)
return g1, (dWx2, dWh2, dbx2, dbh2)
# ======================================================================
# TRAINING (Adam + grad clip + early stopping mean-AUC val)
# ======================================================================
def init_lstm2(rng, D, Hh):
sin = 1.0 / math.sqrt(D)
Wx1 = rng.normal(0.0, sin, (D, 4 * Hh))
Wh1 = rng.normal(0.0, 0.05, (Hh, 4 * Hh))
bx1 = np.zeros(4 * Hh)
bh1 = np.zeros(4 * Hh)
Wx2 = rng.normal(0.0, sin, (Hh, 4 * Hh))
Wh2 = rng.normal(0.0, 0.05, (Hh, 4 * Hh))
bx2 = np.zeros(4 * Hh)
bh2 = np.zeros(4 * Hh)
Wy = rng.normal(0.0, 0.05, (Hh, 2))
by = np.zeros(2)
return (Wx1, Wh1, bx1, bh1), (Wx2, Wh2, bx2, bh2), Wy, by
def train_lstm2(Xtr, Ytr, Xva, Yva, seed=42, hidd=H, lr=LR, epoch_max=EPOCH_MAX,
patience=PATIENCE, verbose=True):
"""Ytr/Yva: (n,2) {long,short}. Early stopping pd mean(AUC_long, AUC_short) val."""
rng = np.random.default_rng(seed)
Bt, T, D = Xtr.shape
W1, W2, Wy, by = init_lstm2(rng, D, hidd)
params = [W1[0], W1[1], W1[2], W1[3],
W2[0], W2[1], W2[2], W2[3],
Wy, by]
m = [np.zeros_like(p) for p in params]
v = [np.zeros_like(p) for p in params]
beta1, beta2, eps = 0.9, 0.999, 1e-8
best_va, best_ep, best_state = -1.0, 0, None
tstep = 0
n = len(Xtr)
for ep in range(epoch_max):
perm = rng.permutation(n)
ep_loss = 0.0
for s in range(0, n, BATCH):
idx = perm[s:s + BATCH]
Xb, Yb = Xtr[idx], Ytr[idx]
h2s, c1, c2, h1s = lstm2_forward(Xb, W1, W2)
P = head2_forward(h2s[:, -1], Wy, by)
loss = -np.mean(Yb * np.log(P + 1e-12) +
(1.0 - Yb) * np.log(1.0 - P + 1e-12))
ep_loss += loss * len(idx)
dWy, dby, dh2_last = head2_backward(h2s[:, -1], P, Yb, Wy)
(dWx1, dWh1, dbx1, dbh1), (dWx2, dWh2, dbx2, dbh2) = \
lstm2_bptt(Xb, h1s, c1, c2, dh2_last, W1, W2)
grads = [dWx1, dWh1, dbx1, dbh1, dWx2, dWh2, dbx2, dbh2, dWy, dby]
gn = math.sqrt(sum(float((g ** 2).sum()) for g in grads)) + 1e-12
if gn > CLIP:
grads = [g * (CLIP / gn) for g in grads]
tstep += 1
for i in range(10):
m[i] = beta1 * m[i] + (1 - beta1) * grads[i]
v[i] = beta2 * v[i] + (1 - beta2) * grads[i] ** 2
mh = m[i] / (1 - beta1 ** tstep)
vh = v[i] / (1 - beta2 ** tstep)
params[i] = params[i] - lr * mh / (np.sqrt(vh) + eps)
W1 = (params[0], params[1], params[2], params[3])
W2 = (params[4], params[5], params[6], params[7])
Wy, by = params[8], params[9]
Pl, Ps = predict_lstm2(Xva, W1, W2, Wy, by)
auc_l = TM.auc(Yva[:, 0].astype(int), Pl)
auc_s = TM.auc(Yva[:, 1].astype(int), Ps)
va = 0.5 * (auc_l + auc_s)
if va > best_va:
best_va, best_ep = va, ep
best_state = ([p.copy() for p in params])
if verbose:
print(f" ep {ep:2d} loss {ep_loss / n:.4f} "
f"AUC_va L={auc_l:.4f} S={auc_s:.4f}")
if ep - best_ep > patience:
break
W1 = (best_state[0], best_state[1], best_state[2], best_state[3])
W2 = (best_state[4], best_state[5], best_state[6], best_state[7])
Wy, by = best_state[8], best_state[9]
return (W1, W2, Wy, by), best_va, best_ep
def predict_lstm2(Xs, W1, W2, Wy, by):
h2s, _, _, _ = lstm2_forward(Xs, W1, W2)
P = head2_forward(h2s[:, -1], Wy, by)
return P[:, 0], P[:, 1]
# ======================================================================
# DATA: cache baru & baseline
# ======================================================================
def load_new():
z = np.load(DATA_NEW)
F, F2, label, close, t = (z["F"], z["F2"], z["label"], z["close"], z["time"])
split_bar, split_pos = int(z["split_bar"]), int(z["split_pos"])
return F, F2, label, close, t, split_bar, split_pos
def load_base():
z = np.load(DATA_BASE)
F, label, ATR, close = z["F"], z["label"], z["ATR"], z["close"]
if F.shape[1] != len(TM.FEAT_NAMES):
F = np.column_stack([F, TM.confluence_feature(F)]).astype(float)
mask = label != 0
midx = np.where(mask)[0]
split_pos = int(0.75 * len(midx))
split_bar = int(midx[split_pos])
return F, label, close, midx, split_pos, split_bar
def build_sequences(Fs, idxs, window=W):
return TL.build_sequences(Fs, idxs, window)
def prepare_split(X, label, midx, split_pos, frac=1.0):
"""Standarisasi train-only; sequence; Y 2-kolom. frac=1.0 = pakai semua."""
n_use = max(1, int(frac * split_pos))
Xt = X[midx[:n_use]]
yt = label[midx[:n_use]]
va_from = int(0.85 * n_use)
mean = Xt[:va_from].mean(0)
std = Xt[:va_from].std(0)
std[std < 1e-9] = 1.0
Xs = (X - mean) / std
Xtr = build_sequences(Xs, midx[:va_from], W)
Xva = build_sequences(Xs, midx[va_from:n_use], W)
Xte = build_sequences(Xs, midx[n_use:], W)
Ytr = np.column_stack([(yt[:va_from] == 1).astype(float),
(yt[:va_from] == -1).astype(float)])
Yva = np.column_stack([(label[midx[va_from:n_use]] == 1).astype(float),
(label[midx[va_from:n_use]] == -1).astype(float)])
Yte = np.column_stack([(label[midx[n_use:]] == 1).astype(float),
(label[midx[n_use:]] == -1).astype(float)])
return Xtr, Xva, Xte, Ytr, Yva, Yte, mean, std
# ======================================================================
# UNIT TEST B1-B5 (2-lapis + 2-head)
# ======================================================================
def test_B1():
rng = np.random.default_rng(0)
B, T, D, Hh = 3, 2, 4, 2
Xs = rng.normal(size=(B, T, D))
W1 = (rng.normal(0, 0.3, (D, 4 * Hh)), rng.normal(0, 0.3, (Hh, 4 * Hh)),
rng.normal(0, 0.1, 4 * Hh), rng.normal(0, 0.1, 4 * Hh))
W2 = (rng.normal(0, 0.3, (Hh, 4 * Hh)), rng.normal(0, 0.3, (Hh, 4 * Hh)),
rng.normal(0, 0.1, 4 * Hh), rng.normal(0, 0.1, 4 * Hh))
# referensi: jalankan TL.lstm_forward dua kali (1-lapis, teruji B1 Fase 3)
h1s_ref, _ = TL.lstm_forward(Xs, *W1)
h2s_ref, _ = TL.lstm_forward(h1s_ref, *W2)
h2s, c1, c2, h1s = lstm2_forward(Xs, W1, W2)
err = float(np.abs(h2s - h2s_ref).max())
print(f" B1 forward 2-lapis vs 2x1-lapis : max_err={err:.2e}")
assert err < 1e-8, "B1 GAGAL"
def test_B2():
rng = np.random.default_rng(1)
B, T, D, Hh = 2, 3, 3, 2
Xs = rng.normal(size=(B, T, D))
W1 = (rng.normal(0, 0.2, (D, 4 * Hh)), rng.normal(0, 0.2, (Hh, 4 * Hh)),
rng.normal(0, 0.1, 4 * Hh), rng.normal(0, 0.1, 4 * Hh))
W2 = (rng.normal(0, 0.2, (Hh, 4 * Hh)), rng.normal(0, 0.2, (Hh, 4 * Hh)),
rng.normal(0, 0.1, 4 * Hh), rng.normal(0, 0.1, 4 * Hh))
Wy = rng.normal(0, 0.2, (Hh, 2))
by = np.array([0.1, -0.1])
Y = np.array([[1.0, 0.0], [0.0, 1.0]])
def loss_fn(p):
W1p = (p[0], p[1], p[2], p[3])
W2p = (p[4], p[5], p[6], p[7])
Wyp = p[8]
byp = p[9]
h2s, _, _, _ = lstm2_forward(Xs, W1p, W2p)
P = head2_forward(h2s[:, -1], Wyp, byp)
return float(-np.mean(Y * np.log(P + 1e-12) +
(1.0 - Y) * np.log(1.0 - P + 1e-12)))
h2s, c1, c2, h1s = lstm2_forward(Xs, W1, W2)
P = head2_forward(h2s[:, -1], Wy, by)
dWy, dby, dh2_last = head2_backward(h2s[:, -1], P, Y, Wy)
(dWx1, dWh1, dbx1, dbh1), (dWx2, dWh2, dbx2, dbh2) = \
lstm2_bptt(Xs, h1s, c1, c2, dh2_last, W1, W2)
grads = [dWx1, dWh1, dbx1, dbh1, dWx2, dWh2, dbx2, dbh2, dWy, dby]
params = [W1[0], W1[1], W1[2], W1[3], W2[0], W2[1], W2[2], W2[3], Wy, by]
eps = 1e-6
maxrel = 0.0
names = ["Wx1", "Wh1", "bx1", "bh1", "Wx2", "Wh2", "bx2", "bh2", "Wy", "by"]
for nm, g, p in zip(names, grads, params):
num = np.zeros_like(p)
it = np.nditer(p, flags=["multi_index"])
while not it.finished:
i = it.multi_index
old = p[i]
p[i] = old + eps
fp = loss_fn(params)
p[i] = old - eps
fm = loss_fn(params)
p[i] = old
num[i] = (fp - fm) / (2.0 * eps)
it.iternext()
denom = np.abs(num) + np.abs(g) + 1e-12
rel = float((np.abs(num - g) / denom).max())
maxrel = max(maxrel, rel)
print(f" B2 {nm}: rel={rel:.2e} |num|max={np.abs(num).max():.3e}")
print(f" B2 gradient check (BPTT 2-lapis) : max_rel_err={maxrel:.2e}")
assert maxrel < 1e-4, "B2 GAGAL"
def test_B3():
rng = np.random.default_rng(42)
Xs = rng.normal(size=(64, 8, 5))
y = (rng.random(64) > 0.5).astype(float)
Y = np.column_stack([y, 1.0 - y])
r1 = train_lstm2(Xs, Y, Xs[:32], Y[:32], seed=7, verbose=False)
r2 = train_lstm2(Xs, Y, Xs[:32], Y[:32], seed=7, verbose=False)
ok = all(np.array_equal(a, b) for a, b in zip(r1[0][0], r2[0][0])) and \
all(np.array_equal(a, b) for a, b in zip(r1[0][1], r2[0][1]))
print(f" B3 determinisme (seed sama) : {'OK' if ok else 'GAGAL'}")
assert ok, "B3 GAGAL"
def test_B4():
F = np.arange(500 * 5, dtype=np.float64).reshape(500, 5)
idxs = np.array([10, 100, 250])
Xs = build_sequences(F, idxs, W)
for k, t in enumerate(idxs):
assert np.array_equal(Xs[k, -1], F[t]), f"B4 GAGAL t={t}"
assert Xs[k].shape == (W, 5)
print(" B4 anti-lookahead (seq berakhir di bar t) : OK")
def test_B5():
rng = np.random.default_rng(3)
Xs = rng.normal(size=(200, 4, 6))
y = (rng.random(200) > 0.5).astype(float)
Y = np.column_stack([y, 1.0 - y])
state, va, ep = train_lstm2(Xs, Y, Xs, Y, seed=9, hidd=8, lr=1e-2,
epoch_max=300, verbose=False)
Pl, Ps = predict_lstm2(Xs, *state)
auc_tr = TM.auc(y.astype(int), Pl)
print(f" B5 overfit sanity (2-lapis) : AUC_train(long)={auc_tr:.4f}")
assert auc_tr > 0.95, "B5 GAGAL"
def selftest():
print("=== UNIT TEST B1-B5 (LSTM 2-LAPIS + 2-HEAD) ===")
test_B1()
test_B2()
test_B3()
test_B4()
test_B5()
print(">>> SEMUA UNIT TEST B1-B5 LOLOS (G1 OK) <<<")
# ======================================================================
# GATE LIKE-FOR-LIKE pd cache XAUUSDc (vs 0.6270/0.6207)
# ======================================================================
def gate_xauusdc():
elog("\n=== GATE LIKE-FOR-LIKE (cache XAUUSDc 60k, split 75/25) ===")
F, label, close, midx, split_pos, split_bar = load_base()
X = F
n = len(midx)
elog(f" bar={len(close)} berlabel={n} split_pos={split_pos} split_bar={split_bar}")
# baseline MLP (reproduksi, kontrol kontaminasi) — long & short
yl = (label[midx] == 1).astype(float)
ys = (label[midx] == -1).astype(float)
Xtr, ytr = X[midx[:split_pos]], yl[:split_pos]
va_from = int(0.85 * split_pos)
mean = Xtr[:va_from].mean(0)
std = Xtr[:va_from].std(0)
std[std < 1e-9] = 1.0
auc_base_l, auc_base_s = 0.0, 0.0
# long
(W1l, b1l, W2l, b2l), auc_va, _ = TM.train(
(Xtr[:va_from] - mean) / std, yl[:va_from],
(Xtr[va_from:] - mean) / std, yl[va_from:])
_, Pte_l = TM.forward((X[midx[split_pos:]] - mean) / std, W1l, b1l, W2l, b2l)
auc_base_l = TM.auc(yl[split_pos:].astype(int), Pte_l[:, 0])
# short
(W1s, b1s, W2s, b2s), _, _ = TM.train(
(Xtr[:va_from] - mean) / std, ys[:va_from],
(Xtr[va_from:] - mean) / std, ys[va_from:])
_, Pte_s = TM.forward((X[midx[split_pos:]] - mean) / std, W1s, b1s, W2s, b2s)
auc_base_s = TM.auc(ys[split_pos:].astype(int), Pte_s[:, 0])
elog(f" BASELINE MLP (reproduksi): LONG={auc_base_l:.4f} SHORT={auc_base_s:.4f} "
f"(target 0.6270/0.6207)")
# hybrid 2-lapis 2-head pd split yang SAMA
Xtrs = build_sequences((X - mean) / std, midx[:split_pos], W)
Xtes = build_sequences((X - mean) / std, midx[split_pos:], W)
# bagi train -> train/val 85/15
va2 = int(0.85 * split_pos)
Xtr2, Xva2 = Xtrs[:va2], Xtrs[va2:]
Ytr2 = np.column_stack([yl[:va2], ys[:va2]])
Yva2 = np.column_stack([yl[va2:split_pos], ys[va2:split_pos]])
Yte = np.column_stack([yl[split_pos:], ys[split_pos:]])
aucs_l, aucs_s = [], []
P_ens_l, P_ens_s = np.zeros(len(Yte)), np.zeros(len(Yte))
for sd in SEEDS:
st, va, ep = train_lstm2(Xtr2, Ytr2, Xva2, Yva2, seed=sd, verbose=False)
Pl, Ps = predict_lstm2(Xtes, *st)
al = TM.auc(Yte[:, 0].astype(int), Pl)
as_ = TM.auc(Yte[:, 1].astype(int), Ps)
aucs_l.append(al)
aucs_s.append(as_)
P_ens_l += Pl / len(SEEDS)
P_ens_s += Ps / len(SEEDS)
elog(f" seed {sd}: LONG={al:.4f} SHORT={as_:.4f}")
mean_l = float(np.mean(aucs_l))
mean_s = float(np.mean(aucs_s))
std_l = float(np.std(aucs_l))
std_s = float(np.std(aucs_s))
ens_l = TM.auc(Yte[:, 0].astype(int), P_ens_l)
ens_s = TM.auc(Yte[:, 1].astype(int), P_ens_s)
elog(f" HYBRID 2-lapis: LONG mean={mean_l:.4f}+-{std_l:.4f} ens={ens_l:.4f} | "
f"SHORT mean={mean_s:.4f}+-{std_s:.4f} ens={ens_s:.4f}")
# bootstrap paired CI (ens vs baseline MLP) 2000x
lo_l, hi_l, dm_l = TL.bootstrap_ci(Yte[:, 0].astype(int), P_ens_l, Pte_l[:, 0])
lo_s, hi_s, dm_s = TL.bootstrap_ci(Yte[:, 1].astype(int), P_ens_s, Pte_s[:, 0])
elog(f" dAUC ens-vs-MLP: LONG {dm_l:+.4f} CI95=[{lo_l:+.4f},{hi_l:+.4f}] | "
f"SHORT {dm_s:+.4f} CI95=[{lo_s:+.4f},{hi_s:+.4f}]")
# gate summary
elog("\n=== GATE (vs baseline 0.6270/0.6207) ===")
g_long = ens_l - BASE_LONG
g_short = ens_s - BASE_SHORT
elog(f" LONG : ens={ens_l:.4f} vs base {BASE_LONG:.4f} -> dAUC={g_long:+.4f} "
f"({'>=+0.005' if g_long >= 0.005 else 'GAGAL G3'})")
elog(f" SHORT : ens={ens_s:.4f} vs base {BASE_SHORT:.4f} -> dAUC={g_short:+.4f} "
f"({'>=+0.005' if g_short >= 0.005 else 'GAGAL G3'})")
elog(f" G4 CI tak memuat 0 : LONG {'YA' if lo_l > 0 else 'TIDAK'} | "
f"SHORT {'YA' if lo_s > 0 else 'TIDAK'}")
elog(f" G6 std seed <0.01 & >=4/5 > base : LONG std={std_l:.4f} "
f"beat={int(sum(a > BASE_LONG for a in aucs_l))}/5 | SHORT std={std_s:.4f} "
f"beat={int(sum(a > BASE_SHORT for a in aucs_s))}/5")
return dict(ens_l=ens_l, ens_s=ens_s, base_l=auc_base_l, base_s=auc_base_s,
ci_l=(lo_l, hi_l), ci_s=(lo_s, hi_s))
# ======================================================================
# WALK-FORWARD pd data baru (75/25, 70/30, 80/20; [F19] vs [F19+F2])
# ======================================================================
def run_wf_split(X, label, midx, split_pos, use_regime, seed):
Xt = np.column_stack([X[0], X[1]]) if use_regime else X[0]
Xtr, Xva, Xte, Ytr, Yva, Yte, mean, std = prepare_split(Xt, label, midx, split_pos)
st, va, ep = train_lstm2(Xtr, Ytr, Xva, Yva, seed=seed, verbose=False)
Pl, Ps = predict_lstm2(Xte, *st)
al = TM.auc(Yte[:, 0].astype(int), Pl)
as_ = TM.auc(Yte[:, 1].astype(int), Ps)
return al, as_
def wf_new():
elog("\n=== WALK-FORWARD DATA BARU (features_XAUUSD.npz) ===")
F, F2, label, close, t, split_bar, split_pos = load_new()
mask = label != 0
midx = np.where(mask)[0]
elog(f" berlabel={len(midx)} split_pos(75/25)={split_pos}")
for frac in (0.75, 0.70, 0.80):
sp = int(frac * len(midx))
for use_regime in (False, True):
tag = f"split{int(frac*100)}/{int((1-frac)*100)} " \
f"{'F19+F2' if use_regime else 'F19'}"
al_, as_ = [], []
for sd in SEEDS[:2]: # kalibrasi dulu 2 seed
a, b = run_wf_split((F, F2), label, midx, sp, use_regime, sd)
al_.append(a)
as_.append(b)
elog(f" [{tag}] (2 seed) LONG={np.mean(al_):.4f} SHORT={np.mean(as_):.4f}")
elog(" (wf 5-seed penuh dijalankan bila budget/token cukup — lihat handoff)")
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--selftest", action="store_true")
ap.add_argument("--gate", action="store_true")
ap.add_argument("--wf", action="store_true")
ap.add_argument("--seed", type=int, default=None)
args = ap.parse_args()
if args.selftest:
selftest()
return
if args.gate:
gate_xauusdc()
return
if args.wf:
wf_new()
return
selftest()
gate_xauusdc()
if __name__ == "__main__":
main()
Reference in a new issue Copy permalink