""" AHAD QUANT Forex V5 — TransformerGRU Hybrid Model Combines a Bidirectional GRU (short-range pattern capture) with a Transformer Encoder (long-range self-attention) for binary classification. Architecture ─────────────────────────────────────────────────────────────── Input : (B, T, 62) Layer 1: Input projection — Linear(62, D_MODEL) + GELU → (B, T, D_MODEL) Layer 2: Positional Encoding (sinusoidal, fixed) → (B, T, D_MODEL) Layer 3: Bidirectional GRU — hidden=GRU_HIDDEN, 2 layers — captures local sequential dependencies in both directions → (B, T, D_MODEL) [GRU_HIDDEN*2 projected back to D_MODEL] Layer 4: Transformer Encoder — N_ATT_LAYERS × (MHA + FFN + LayerNorm) — long-range temporal attention on top of GRU hidden states → (B, T, D_MODEL) Layer 5: Aggregation — mean pooling + last token concatenated → (B, D_MODEL*2) Layer 6: Classification head Linear → GELU → Dropout → Linear(1) — logit Output : (B,) logit — apply sigmoid for probability Training util functions ─────────────────────── train_tgru(X_seq_tr, y_tr, X_seq_va, y_va, ...) → TransformerGRU predict_tgru_proba(model, X_seq_tensor) → np.ndarray (B,) """ import math import time import numpy as np import torch import torch.nn as nn import torch.nn.functional as F from torch.utils.data import DataLoader, TensorDataset # ─── Hyper-parameters (can be overridden via kwargs in train_tgru) ─────────── D_MODEL = 128 # main embedding dimension GRU_HIDDEN = 64 # GRU hidden per direction (×2 bidir = D_MODEL) N_ATT_LAYERS = 3 # Transformer encoder depth N_HEADS = 4 # attention heads (D_MODEL must be divisible) FFN_DIM = 256 # feedforward expansion in Transformer DROPOUT = 0.10 BATCH_SIZE = 512 LR = 5e-4 WEIGHT_DECAY = 1e-4 MAX_EPOCHS = 30 PATIENCE = 6 GRAD_CLIP = 1.0 # ─── Positional Encoding (sinusoidal) ──────────────────────────────────────── class SinusoidalPositionalEncoding(nn.Module): """ Fixed sinusoidal positional encoding added to embeddings. PE(pos, 2i) = sin(pos / 10000^(2i/d_model)) PE(pos, 2i+1) = cos(pos / 10000^(2i/d_model)) """ def __init__(self, d_model: int, max_len: int = 512, dropout: float = 0.1): super().__init__() self.drop = nn.Dropout(dropout) pe = torch.zeros(max_len, d_model) pos = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) div = torch.exp( torch.arange(0, d_model, 2, dtype=torch.float) * (-math.log(10000.0) / d_model) ) pe[:, 0::2] = torch.sin(pos * div) pe[:, 1::2] = torch.cos(pos * div) self.register_buffer("pe", pe.unsqueeze(0)) # (1, max_len, d_model) def forward(self, x: torch.Tensor) -> torch.Tensor: """x : (B, T, d_model)""" return self.drop(x + self.pe[:, : x.size(1)]) # ─── Transformer Encoder Block ─────────────────────────────────────────────── class TransformerEncoderBlock(nn.Module): """ Standard Pre-LN Transformer encoder block: x → LayerNorm → MHA → residual → LayerNorm → FFN → residual Pre-LayerNorm (before attention) provides more stable training. """ def __init__(self, d_model: int, n_heads: int, ffn_dim: int, dropout: float = DROPOUT): super().__init__() self.norm1 = nn.LayerNorm(d_model) self.attn = nn.MultiheadAttention(d_model, n_heads, dropout=dropout, batch_first=True) self.norm2 = nn.LayerNorm(d_model) self.ffn = nn.Sequential( nn.Linear(d_model, ffn_dim), nn.GELU(), nn.Dropout(dropout), nn.Linear(ffn_dim, d_model), nn.Dropout(dropout), ) def forward(self, x: torch.Tensor) -> torch.Tensor: # Self-attention sub-layer h, _ = self.attn(self.norm1(x), self.norm1(x), self.norm1(x)) x = x + h # Feed-forward sub-layer x = x + self.ffn(self.norm2(x)) return x # ─── Main model ────────────────────────────────────────────────────────────── class TransformerGRU(nn.Module): """ TransformerGRU for binary classification on (B, T, F) time-series. Output: (B,) raw logit — use with BCEWithLogitsLoss. """ def __init__( self, num_features : int = 62, seq_len : int = 168, d_model : int = D_MODEL, gru_hidden : int = GRU_HIDDEN, n_att_layers : int = N_ATT_LAYERS, n_heads : int = N_HEADS, ffn_dim : int = FFN_DIM, dropout : float = DROPOUT, ): super().__init__() assert d_model == gru_hidden * 2, ( f"D_MODEL ({d_model}) must equal GRU_HIDDEN*2 ({gru_hidden*2}) " "so bidirectional GRU output aligns with d_model" ) self.d_model = d_model self.seq_len = seq_len # ── 1. Input projection ────────────────────────────────────────────── self.input_proj = nn.Sequential( nn.Linear(num_features, d_model), nn.GELU(), nn.Dropout(dropout), ) # ── 2. Positional encoding ─────────────────────────────────────────── self.pos_enc = SinusoidalPositionalEncoding(d_model, max_len=seq_len + 2, dropout=dropout) # ── 3. Bidirectional GRU ───────────────────────────────────────────── # Input: (B, T, d_model) → Output: (B, T, gru_hidden*2 = d_model) self.gru = nn.GRU( input_size = d_model, hidden_size = gru_hidden, num_layers = 2, batch_first = True, bidirectional = True, dropout = dropout, ) self.gru_norm = nn.LayerNorm(d_model) # Residual projection (input_proj output → d_model already, residual fits) # ── 4. Transformer encoder ─────────────────────────────────────────── self.transformer = nn.ModuleList([ TransformerEncoderBlock(d_model, n_heads, ffn_dim, dropout) for _ in range(n_att_layers) ]) self.final_norm = nn.LayerNorm(d_model) # ── 5. Classification head ─────────────────────────────────────────── # Aggregate: mean pool + last token → concat → d_model*2 self.head = nn.Sequential( nn.Linear(d_model * 2, d_model), nn.GELU(), nn.Dropout(dropout), nn.Linear(d_model, 1), ) def forward(self, x: torch.Tensor) -> torch.Tensor: """ x : (B, T, num_features) → (B,) logit """ # 1. Input projection + positional encoding h = self.input_proj(x) # (B, T, d_model) h = self.pos_enc(h) # 2. Bidirectional GRU — add residual gru_out, _ = self.gru(h) # (B, T, d_model) [bidir → gru_h*2] h = self.gru_norm(h + gru_out) # add+norm residual # 3. Transformer encoder for block in self.transformer: h = block(h) # (B, T, d_model) h = self.final_norm(h) # 4. Aggregation: mean pool + last token mean_pool = h.mean(dim=1) # (B, d_model) last_tok = h[:, -1, :] # (B, d_model) pooled = torch.cat([mean_pool, last_tok], dim=-1) # (B, d_model*2) # 5. Head return self.head(pooled).squeeze(-1) # (B,) # ─── Training utilities ────────────────────────────────────────────────────── def _make_loader(X: np.ndarray, y: np.ndarray, batch_size: int, shuffle: bool) -> DataLoader: X_t = torch.FloatTensor(X) y_t = torch.FloatTensor(y) return DataLoader(TensorDataset(X_t, y_t), batch_size=batch_size, shuffle=shuffle, num_workers=0, pin_memory=torch.cuda.is_available()) def train_tgru( X_seq_tr : np.ndarray, y_tr : np.ndarray, X_seq_va : np.ndarray, y_va : np.ndarray, *, # Architecture overrides d_model : int = D_MODEL, gru_hidden : int = GRU_HIDDEN, n_att_layers : int = N_ATT_LAYERS, n_heads : int = N_HEADS, ffn_dim : int = FFN_DIM, dropout : float = DROPOUT, # Training overrides batch_size : int = BATCH_SIZE, lr : float = LR, weight_decay : float = WEIGHT_DECAY, max_epochs : int = MAX_EPOCHS, patience : int = PATIENCE, grad_clip : float = GRAD_CLIP, ) -> "TransformerGRU": """ Train a TransformerGRU on pre-normalised sequence data. Parameters ---------- X_seq_tr / X_seq_va : (N, T, 62) float32 — already normalised y_tr / y_va : (N,) float32 binary labels Returns ------- Best model (lowest val loss) as a CPU-resident TransformerGRU. """ device = torch.device("cuda" if torch.cuda.is_available() else "cpu") print(f" [TGRU] device={device} | " f"train={len(y_tr):,} val={len(y_va):,} samples") _, T, F = X_seq_tr.shape model = TransformerGRU( num_features = F, seq_len = T, d_model = d_model, gru_hidden = gru_hidden, n_att_layers = n_att_layers, n_heads = n_heads, ffn_dim = ffn_dim, dropout = dropout, ).to(device) n_params = sum(p.numel() for p in model.parameters() if p.requires_grad) print(f" [TGRU] parameters: {n_params:,}") optimiser = torch.optim.AdamW(model.parameters(), lr=lr, weight_decay=weight_decay) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimiser, T_max=max_epochs) criterion = nn.BCEWithLogitsLoss() scaler_amp = torch.cuda.amp.GradScaler(enabled=torch.cuda.is_available()) train_loader = _make_loader(X_seq_tr, y_tr, batch_size, shuffle=True) val_loader = _make_loader(X_seq_va, y_va, batch_size * 2, shuffle=False) best_val_loss = float("inf") best_state = None no_improve = 0 t0 = time.time() for epoch in range(1, max_epochs + 1): # ── Train ──────────────────────────────────────────────────────────── model.train() train_loss = 0.0 for X_b, y_b in train_loader: X_b, y_b = X_b.to(device), y_b.to(device) optimiser.zero_grad() with torch.cuda.amp.autocast(enabled=torch.cuda.is_available()): logits = model(X_b) loss = criterion(logits, y_b) scaler_amp.scale(loss).backward() scaler_amp.unscale_(optimiser) nn.utils.clip_grad_norm_(model.parameters(), grad_clip) scaler_amp.step(optimiser) scaler_amp.update() train_loss += loss.item() * len(y_b) train_loss /= len(y_tr) # ── Validate ───────────────────────────────────────────────────────── model.eval() val_loss = 0.0 correct = 0 with torch.no_grad(): for X_b, y_b in val_loader: X_b, y_b = X_b.to(device), y_b.to(device) logits = model(X_b) val_loss += criterion(logits, y_b).item() * len(y_b) correct += ((logits > 0).float() == y_b).sum().item() val_loss /= len(y_va) val_acc = correct / len(y_va) scheduler.step() elapsed = time.time() - t0 print(f" [TGRU] epoch {epoch:3d}/{max_epochs} " f"train={train_loss:.4f} val={val_loss:.4f} " f"val_acc={val_acc:.4f} ({elapsed:.0f}s)") # ── Early stopping ─────────────────────────────────────────────────── if val_loss < best_val_loss - 1e-5: best_val_loss = val_loss best_state = {k: v.cpu().clone() for k, v in model.state_dict().items()} no_improve = 0 else: no_improve += 1 if no_improve >= patience: print(f" [TGRU] early stopping at epoch {epoch}") break model.load_state_dict(best_state) model.cpu() print(f" [TGRU] training complete — best val_loss={best_val_loss:.4f}") return model @torch.no_grad() def predict_tgru_proba( model : TransformerGRU, X_seq_t : torch.Tensor, batch_size: int = 1024, device : str = "cpu", ) -> np.ndarray: """ Run inference on a (N, T, F) float tensor. Returns probability array of shape (N,). """ model.eval() model.to(device) probs = [] for start in range(0, len(X_seq_t), batch_size): chunk = X_seq_t[start : start + batch_size].to(device) logits = model(chunk) probs.append(torch.sigmoid(logits).cpu().numpy()) model.cpu() return np.concatenate(probs).astype(np.float32)