# -*- 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()