Lesson 3.6: Attention Mechanisms and Transformers in Time-Series and Financial Markets

Introduction: The Limitations of Recurrent Architectures

While Long Short-Term Memory (LSTM) networks and Gated Recurrent Units (GRUs) successfully solved the vanishing gradient problem, they possess a fundamental architectural bottleneck: Sequential Processing. Because RNNs and LSTMs must process time-series data step-by-step (t₁ to t₂ to t₃), they cannot be parallelized across modern GPU clusters during training. Furthermore, compressing an entire multi-year financial history into a single fixed-size hidden state creates an information bottleneck, causing models to forget long-term events.

To eliminate sequential processing and capture complex relationships across long-horizon time series, Vaswani et al. introduced the Transformer architecture in 2017, powered entirely by Attention Mechanisms. This lesson deconstructs self-attention, multi-head attention, positional encodings, and their application to financial market forecasting.

Part 1: The Attention Mechanism

Instead of forcing a network to compress past time steps into a single hidden state, the attention mechanism allows the model to dynamically look backward across all previous time steps and assign weighted importance scores to the most relevant data points.

1. Queries, Keys, and Values (Q, K, V)

Analogous to a database retrieval system, the input sequence is projected into three matrices:

Queries (Q): What the model is currently looking for at time step t.
Keys (K): What information each historical time step contains.
Values (V): The actual contextual content of each time step.

2. Scaled Dot-Product Attention

The attention score is computed by taking the dot product of the Queries and Keys, scaling it by the square root of the key dimension (d_k) to prevent vanishing gradients, applying a softmax function to generate probability weights summing to 1, and multiplying by the Values:

Attention(Q, K, V) = softmax(QK^T / √d_k) × V

Financial Significance: When predicting tomorrow’s asset volatility, the attention mechanism can instantly link today’s market conditions directly to a similar macroeconomic shock that occurred five years ago, bypassing intermediate time steps entirely.

Part 2: Multi-Head Attention and Positional Encoding

1. Multi-Head Attention

Instead of performing a single attention calculation, Multi-Head Attention projects queries, keys, and values into multiple lower-dimensional subspaces. This allows the model to simultaneously attend to information from different representation subspaces—for example, one head tracking high-frequency order book imbalances, another tracking daily technical indicators, and a third tracking macroeconomic interest rate trends.

2. Positional Encoding

Because Transformers process entire time series simultaneously (non-sequentially), they possess zero inherent awareness of time order. To give the model temporal context, Positional Encodings (sine and cosine functions of varying frequencies) are added directly to the input embeddings, preserving the chronological sequence of financial events.

Part 3: Application of Transformers to Financial Time-Series

Transformers (such as Temporal Fusion Transformers – TFT, and PatchTST) have revolutionized quantitative finance:

1. Long-Horizon Forecasting

Transformers excel at capturing complex long-term dependencies across multi-year financial datasets without suffering from memory degradation.

2. Multi-Asset Cross-Attention

Unlike LSTMs that struggle with massive multivariate datasets, Transformer attention mechanisms can ingest thousands of parallel asset price series simultaneously, calculating cross-asset correlations and contagion risks in real-time.

1. Attention Mechanism Mathematical Deep-Dive

Scaled Dot-Product Attention Derivation:

Given:
- Query matrix Q ∈ ℝ^{n×d_k}
- Key matrix K ∈ ℝ^{m×d_k}
- Value matrix V ∈ ℝ^{m×d_v}

Attention Computation:
1. Compute similarity: S = QK^T ∈ ℝ^{n×m}
2. Scale: S_scaled = S / √d_k
3. Softmax: A = softmax(S_scaled) ∈ ℝ^{n×m}
4. Output: O = A × V ∈ ℝ^{n×d_v}

Where:
- n = Number of queries (sequence length for query)
- m = Number of keys (sequence length for key/value)
- d_k = Dimension of keys and queries
- d_v = Dimension of values

Attention Implementation:

import numpy as np
import torch
import torch.nn as nn
import torch.nn.functional as F

class ScaledDotProductAttention(nn.Module):
    """
    Scaled Dot-Product Attention mechanism
    """
    def __init__(self, dropout=0.1):
        super().__init__()
        self.dropout = nn.Dropout(dropout)
        self.softmax = nn.Softmax(dim=-1)
    
    def forward(self, Q, K, V, mask=None):
        """
        Forward pass for scaled dot-product attention
        
        Parameters:
        - Q: Queries (batch_size, n_queries, d_k)
        - K: Keys (batch_size, n_keys, d_k)
        - V: Values (batch_size, n_keys, d_v)
        - mask: Optional attention mask
        
        Returns:
        - Output (batch_size, n_queries, d_v)
        - Attention weights (batch_size, n_queries, n_keys)
        """
        d_k = Q.size(-1)
        
        # Compute scores
        scores = torch.matmul(Q, K.transpose(-2, -1)) / np.sqrt(d_k)
        
        # Apply mask if provided
        if mask is not None:
            scores = scores.masked_fill(mask == 0, -1e9)
        
        # Apply softmax
        attention_weights = self.softmax(scores)
        attention_weights = self.dropout(attention_weights)
        
        # Apply attention to values
        output = torch.matmul(attention_weights, V)
        
        return output, attention_weights

2. Multi-Head Attention Deep-Dive

Multi-Head Attention Mathematics:

Multi-Head Attention:
MultiHead(Q, K, V) = Concat(head₁, ..., head_h) × W_O

Where:
head_i = Attention(Q × W_Q^i, K × W_K^i, V × W_V^i)

Dimensions:
- W_Q^i ∈ ℝ^{d_model × d_k}
- W_K^i ∈ ℝ^{d_model × d_k}
- W_V^i ∈ ℝ^{d_model × d_v}
- W_O ∈ ℝ^{h × d_v × d_model}

Where:
- h = Number of heads
- d_model = Model dimension
- d_k = d_model / h (for each head)
- d_v = d_model / h (for each head)

Multi-Head Attention Implementation:

class MultiHeadAttention(nn.Module):
    """
    Multi-Head Attention mechanism
    """
    def __init__(self, d_model, n_heads, dropout=0.1):
        super().__init__()
        assert d_model % n_heads == 0, "d_model must be divisible by n_heads"
        
        self.d_model = d_model
        self.n_heads = n_heads
        self.d_k = d_model // n_heads
        
        # Linear projections for Q, K, V
        self.W_Q = nn.Linear(d_model, d_model)
        self.W_K = nn.Linear(d_model, d_model)
        self.W_V = nn.Linear(d_model, d_model)
        self.W_O = nn.Linear(d_model, d_model)
        
        self.attention = ScaledDotProductAttention(dropout)
        self.dropout = nn.Dropout(dropout)
    
    def forward(self, Q, K, V, mask=None):
        """
        Forward pass for multi-head attention
        """
        batch_size = Q.size(0)
        
        # Linear projections and split into heads
        Q = self.W_Q(Q).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)
        K = self.W_K(K).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)
        V = self.W_V(V).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)
        
        # Apply attention
        attention_output, attention_weights = self.attention(Q, K, V, mask)
        
        # Concatenate heads
        attention_output = attention_output.transpose(1, 2).contiguous().view(
            batch_size, -1, self.d_model
        )
        
        # Final linear projection
        output = self.W_O(attention_output)
        
        return output, attention_weights

3. Positional Encoding Deep-Dive

Sinusoidal Positional Encoding:

For position pos and dimension i:
PE_{(pos, 2i)} = sin(pos / 10000^{2i/d_model})
PE_{(pos, 2i+1)} = cos(pos / 10000^{2i/d_model})

Where:
- pos = Position in sequence (0 to seq_len-1)
- i = Dimension index (0 to d_model/2-1)
- d_model = Model dimension

Positional Encoding Implementation:

class PositionalEncoding(nn.Module):
    """
    Sinusoidal positional encoding for transformers
    """
    def __init__(self, d_model, max_len=5000, dropout=0.1):
        super().__init__()
        self.dropout = nn.Dropout(dropout)
        
        # Create positional encoding matrix
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
        
        # Compute division terms
        div_term = torch.exp(
            torch.arange(0, d_model, 2).float() * (-np.log(10000.0) / d_model)
        )
        
        # Apply sin to even indices and cos to odd indices
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        
        # Add batch dimension
        pe = pe.unsqueeze(0)
        
        # Register as buffer (not a trainable parameter)
        self.register_buffer('pe', pe)
    
    def forward(self, x):
        """
        Add positional encoding to input embeddings
        """
        x = x + self.pe[:, :x.size(1), :]
        return self.dropout(x)

4. Transformer Architecture Deep-Dive

Transformer Encoder Layer:

class TransformerEncoderLayer(nn.Module):
    """
    Single Transformer Encoder Layer
    """
    def __init__(self, d_model, n_heads, d_ff, dropout=0.1):
        super().__init__()
        
        # Multi-head self-attention
        self.self_attention = MultiHeadAttention(d_model, n_heads, dropout)
        
        # Feed-forward network
        self.ffn = nn.Sequential(
            nn.Linear(d_model, d_ff),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(d_ff, d_model)
        )
        
        # Layer normalization
        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)
        
        # Dropout
        self.dropout = nn.Dropout(dropout)
    
    def forward(self, x, mask=None):
        """
        Forward pass through encoder layer
        """
        # Self-attention with residual connection
        attn_output, _ = self.self_attention(x, x, x, mask)
        x = self.norm1(x + self.dropout(attn_output))
        
        # Feed-forward with residual connection
        ffn_output = self.ffn(x)
        x = self.norm2(x + self.dropout(ffn_output))
        
        return x

Complete Transformer:

class Transformer(nn.Module):
    """
    Complete Transformer model for time-series forecasting
    """
    def __init__(self, d_model, n_heads, n_layers, d_ff, input_dim, output_dim, max_len=5000, dropout=0.1):
        super().__init__()
        
        # Input projection
        self.input_projection = nn.Linear(input_dim, d_model)
        
        # Positional encoding
        self.positional_encoding = PositionalEncoding(d_model, max_len, dropout)
        
        # Encoder layers
        self.encoder_layers = nn.ModuleList([
            TransformerEncoderLayer(d_model, n_heads, d_ff, dropout)
            for _ in range(n_layers)
        ])
        
        # Output projection
        self.output_projection = nn.Linear(d_model, output_dim)
        
        # Layer normalization
        self.norm = nn.LayerNorm(d_model)
    
    def forward(self, x, mask=None):
        """
        Forward pass through transformer
        """
        # Input projection
        x = self.input_projection(x)
        
        # Add positional encoding
        x = self.positional_encoding(x)
        
        # Pass through encoder layers
        for layer in self.encoder_layers:
            x = layer(x, mask)
        
        # Layer norm
        x = self.norm(x)
        
        # Output projection (use last time step)
        x = self.output_projection(x[:, -1, :])
        
        return x

5. Time-Series Transformer Models

Temporal Fusion Transformer (TFT):

class TemporalFusionTransformer(nn.Module):
    """
    Temporal Fusion Transformer for time-series forecasting
    """
    def __init__(self, d_model, n_heads, n_layers, d_ff, input_dim, output_dim, static_dim, max_len=5000):
        super().__init__()
        
        # Static covariate encoder
        self.static_encoder = nn.Linear(static_dim, d_model)
        
        # Time-series encoder
        self.time_encoder = nn.Linear(input_dim, d_model)
        
        # Transformer encoder
        self.encoder = Transformer(d_model, n_heads, n_layers, d_ff, d_model, d_model, max_len)
        
        # Decoder (for forecasting)
        self.decoder = nn.Sequential(
            nn.Linear(d_model * 2, d_model),
            nn.ReLU(),
            nn.Linear(d_model, output_dim)
        )
    
    def forward(self, time_series, static_features, mask=None):
        """
        Forward pass for TFT
        """
        # Encode time series
        time_encoded = self.time_encoder(time_series)
        
        # Encode static features (broadcast to all time steps)
        static_encoded = self.static_encoder(static_features)
        static_encoded = static_encoded.unsqueeze(1).expand(-1, time_series.size(1), -1)
        
        # Combine time and static features
        combined = torch.cat([time_encoded, static_encoded], dim=-1)
        
        # Pass through encoder
        encoded = self.encoder(combined, mask)
        
        # Decode to forecast
        forecast = self.decoder(encoded)
        
        return forecast

PatchTST (Patch Time Series Transformer):

class PatchTST(nn.Module):
    """
    Patch Time-Series Transformer for efficient forecasting
    """
    def __init__(self, d_model, n_heads, n_layers, d_ff, input_dim, output_dim, patch_len, stride):
        super().__init__()
        
        self.patch_len = patch_len
        self.stride = stride
        
        # Patch projection
        self.patch_projection = nn.Linear(patch_len, d_model)
        
        # Positional encoding for patches
        self.positional_encoding = PositionalEncoding(d_model, max_len=1000)
        
        # Transformer encoder
        self.encoder = Transformer(d_model, n_heads, n_layers, d_ff, d_model, d_model)
        
        # Output projection
        self.output_projection = nn.Linear(d_model, output_dim)
    
    def forward(self, x):
        """
        Forward pass for PatchTST
        """
        batch_size, seq_len, features = x.shape
        
        # Patch the sequence
        patches = []
        for i in range(0, seq_len - self.patch_len + 1, self.stride):
            patch = x[:, i:i+self.patch_len, :]
            patches.append(patch)
        
        # Stack patches
        patches = torch.stack(patches, dim=1)  # (batch, n_patches, patch_len, features)
        
        # Project patches
        patch_embeddings = self.patch_projection(patches)  # (batch, n_patches, d_model)
        
        # Add positional encoding
        patch_embeddings = self.positional_encoding(patch_embeddings)
        
        # Pass through transformer
        encoded = self.encoder(patch_embeddings)
        
        # Output
        output = self.output_projection(encoded.mean(dim=1))
        
        return output

6. Transformer vs LSTM Comparison

Feature	LSTM	Transformer
Processing	Sequential	Parallel
Long-Term Dependencies	Limited by vanishing gradients	Direct attention
Training Speed	Slow (sequential)	Fast (parallel)
Memory Usage	O(T × d_h)	O(T² × d_model)
Explainability	Low (black box)	High (attention weights)
Performance on Long Sequences	Degrades	Maintains
Parameter Efficiency	High	Lower
Pre-training Potential	Limited	Excellent

7. Financial Time-Series with Transformers

class FinancialTimeSeriesTransformer:
    """
    Transformer for financial time-series forecasting
    """
    def __init__(self, d_model=64, n_heads=8, n_layers=6, d_ff=128, input_dim=10, output_dim=1):
        self.model = Transformer(d_model, n_heads, n_layers, d_ff, input_dim, output_dim)
        self.optimizer = torch.optim.Adam(self.model.parameters(), lr=0.001)
        self.criterion = nn.MSELoss()
    
    def fit(self, X_train, y_train, X_val, y_val, epochs=100, batch_size=32):
        """
        Train transformer on financial data
        """
        train_loader = torch.utils.data.DataLoader(
            torch.utils.data.TensorDataset(X_train, y_train),
            batch_size=batch_size,
            shuffle=True
        )
        
        val_loader = torch.utils.data.DataLoader(
            torch.utils.data.TensorDataset(X_val, y_val),
            batch_size=batch_size,
            shuffle=False
        )
        
        for epoch in range(epochs):
            # Training
            self.model.train()
            train_loss = 0
            for X_batch, y_batch in train_loader:
                self.optimizer.zero_grad()
                y_pred = self.model(X_batch)
                loss = self.criterion(y_pred, y_batch)
                loss.backward()
                torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
                self.optimizer.step()
                train_loss += loss.item()
            
            # Validation
            self.model.eval()
            val_loss = 0
            with torch.no_grad():
                for X_batch, y_batch in val_loader:
                    y_pred = self.model(X_batch)
                    loss = self.criterion(y_pred, y_batch)
                    val_loss += loss.item()
            
            if epoch % 10 == 0:
                print(f"Epoch {epoch}: Train Loss = {train_loss/len(train_loader):.4f}, Val Loss = {val_loss/len(val_loader):.4f}")
    
    def predict(self, X_test):
        """
        Make predictions
        """
        self.model.eval()
        with torch.no_grad():
            predictions = self.model(X_test)
        return predictions.numpy()

Certificate in FinTech Lesson 3.6: Attention Mechanisms and Transformers in Time-Series and Financial Markets

Certificate in FinTech