360 lines
14 KiB
Python
360 lines
14 KiB
Python
|
"""
|
|||
|
|
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)
|