Lesson 5.5: Transformers and Self-Attention for Financial Sequences

1. Learning Objectives

By the end of this lesson, you will be able to:

Understand the mathematical foundations of self-attention and its advantages over RNNs.
Derive the scaled dot-product attention mechanism from first principles.
Implement multi-head attention and positional encoding in PyTorch.
Build and train Transformer models for financial time series forecasting.
Apply Transformers to portfolio optimisation and volatility prediction.
Understand the computational complexity of Transformers and its implications for financial data.
Implement temporal fusion transformers for multi-horizon forecasting.
Understand the limitations of Transformers for financial applications.

2. Why Transformers? – The Limitations of RNNs

2.1 RNN Limitations

Sequential Processing: Cannot parallelise across time steps.
Vanishing Gradients: Long-range dependencies are difficult to capture.
Limited Context: Memory is compressed into a fixed-size hidden state.

2.2 Transformer Advantages

Parallel Processing: All positions are processed simultaneously.
Direct Connections: Any position can attend to any other position directly.
Long-Range Dependencies: No vanishing gradient through time.
Interpretability: Attention weights can be visualised and interpreted.

Financial Implication: Transformers can capture complex, long-range dependencies in financial data (e.g., market cycles, macroeconomic relationships) without the limitations of RNNs.

3. Self-Attention – The Core Mechanism

3.1 Scaled Dot-Product Attention

For a sequence of inputs, we compute three matrices:

Query (Q): What we are looking for.
Key (K): What we are matching against.
Value (V): What we are retrieving.

Mathematical Formulation:
Attention(Q, K, V) = softmax(Q K^T / sqrt(d_k)) V

Where:

Q ∈ R^{n x d_k}, K ∈ R^{m x d_k}, V ∈ R^{m x d_v}
d_k is the dimension of the key vectors.
Scaling by sqrt(d_k) prevents the dot products from becoming too large, which would push the softmax into regions of extremely small gradients.

Derivation of the Scaling Factor:
For a query q and keys k_1, ..., k_m with mean 0 and variance 1, q · k_i has mean 0 and variance d_k. Without scaling, the softmax would saturate. The scaling factor 1/sqrt(d_k) keeps the variance at 1.

3.2 Multi-Head Attention

Multi-head attention runs multiple attention mechanisms in parallel, allowing the model to focus on different aspects of the sequence.

MultiHead(Q, K, V) = Concat(head_1, ..., head_h) W_O

where:
head_i = Attention(Q W_i^Q, K W_i^K, V W_i^V)

3.3 Implementation

import torch
import torch.nn as nn
import torch.nn.functional as F
import math

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, n_heads, dropout=0.1):
        super(MultiHeadAttention, self).__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

        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.dropout = nn.Dropout(dropout)

    def forward(self, query, key, value, mask=None):
        batch_size = query.size(0)

        # Linear projections and split into heads
        Q = self.W_q(query).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)
        K = self.W_k(key).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)
        V = self.W_v(value).view(batch_size, -1, self.n_heads, self.d_k).transpose(1, 2)

        # Scaled dot-product attention
        scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k)

        if mask is not None:
            scores = scores.masked_fill(mask == 0, -1e9)

        attention_weights = F.softmax(scores, dim=-1)
        attention_weights = self.dropout(attention_weights)

        context = torch.matmul(attention_weights, V)

        # Concatenate heads
        context = context.transpose(1, 2).contiguous().view(batch_size, -1, self.d_model)

        # Final linear projection
        output = self.W_o(context)

        return output, attention_weights

4. Positional Encoding – Adding Sequence Order

Since Transformers have no inherent notion of order, we add positional encodings to the input embeddings.

4.1 Sinusoidal Positional Encoding
PE(pos, 2i) = sin(pos / 10000^{2i/d_model})
PE(pos, 2i+1) = cos(pos / 10000^{2i/d_model})

Where:

pos is the position in the sequence.
i is the dimension index.

Properties:

Each position has a unique encoding.
The encoding is deterministic (no learnable parameters).
The encoding allows the model to attend to relative positions because PE(pos + k) is a linear function of PE(pos).

4.2 Implementation

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=1000, dropout=0.1):
        super(PositionalEncoding, self).__init__()

        self.dropout = nn.Dropout(dropout)

        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
        div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))

        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)

        pe = pe.unsqueeze(0)  # (1, max_len, d_model)
        self.register_buffer('pe', pe)

    def forward(self, x):
        x = x + self.pe[:, :x.size(1), :]
        return self.dropout(x)

4.3 Learned Positional Encoding

class LearnedPositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=1000):
        super(LearnedPositionalEncoding, self).__init__()

        self.pos_embedding = nn.Parameter(torch.randn(1, max_len, d_model) * 0.01)

    def forward(self, x):
        return x + self.pos_embedding[:, :x.size(1), :]

5. The Transformer Block

5.1 Architecture

Transformer Block:
    Input: x
    → Multi-Head Self-Attention
    → Add & Norm (Residual connection + Layer Normalisation)
    → Feed Forward (Linear → ReLU → Linear)
    → Add & Norm (Residual connection + Layer Normalisation)
    → Output

5.2 Implementation

class TransformerBlock(nn.Module):
    def __init__(self, d_model, n_heads, d_ff, dropout=0.1):
        super(TransformerBlock, self).__init__()

        self.attention = MultiHeadAttention(d_model, n_heads, dropout)
        self.norm1 = nn.LayerNorm(d_model)
        self.dropout1 = nn.Dropout(dropout)

        self.ff = nn.Sequential(
            nn.Linear(d_model, d_ff),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(d_ff, d_model)
        )
        self.norm2 = nn.LayerNorm(d_model)
        self.dropout2 = nn.Dropout(dropout)

    def forward(self, x, mask=None):
        # Self-attention with residual connection
        attn_output, _ = self.attention(x, x, x, mask)
        x = self.norm1(x + self.dropout1(attn_output))

        # Feed forward with residual connection
        ff_output = self.ff(x)
        x = self.norm2(x + self.dropout2(ff_output))

        return x

6. Full Transformer for Financial Time Series

6.1 Complete Architecture

class FinancialTransformer(nn.Module):
    def __init__(self, input_dim, d_model, n_heads, num_layers, d_ff, output_dim, max_len=100, dropout=0.1):
        super(FinancialTransformer, self).__init__()

        # Input projection
        self.input_proj = nn.Linear(input_dim, d_model)

        # Positional encoding
        self.pos_encoding = PositionalEncoding(d_model, max_len, dropout)

        # Transformer encoder layers
        self.encoder_layers = nn.ModuleList([
            TransformerBlock(d_model, n_heads, d_ff, dropout)
            for _ in range(num_layers)
        ])

        # Output layer
        self.fc = nn.Linear(d_model, output_dim)
        self.dropout = nn.Dropout(dropout)

    def forward(self, x):
        # x: (batch_size, seq_len, input_dim)

        # Input projection
        x = self.input_proj(x)

        # Add positional encoding
        x = self.pos_encoding(x)

        # Pass through transformer layers
        for layer in self.encoder_layers:
            x = layer(x)

        # Global average pooling (or use last token)
        x = x.mean(dim=1)  # (batch_size, d_model)

        # Output
        x = self.dropout(x)
        x = self.fc(x)

        return x

6.2 Masking for Financial Sequences

def create_padding_mask(seq, pad_token=0):
    """
    Create a mask to ignore padding tokens.
    """
    # (batch_size, seq_len)
    mask = (seq != pad_token).unsqueeze(1).unsqueeze(2)
    return mask

def create_future_mask(size):
    """
    Create a mask to prevent attending to future positions.
    Used in decoder for autoregressive generation.
    """
    mask = torch.triu(torch.ones(size, size), diagonal=1).bool()
    return mask

7. Financial Applications of Transformers

7.1 Volatility Forecasting

class VolatilityTransformer(nn.Module):
    def __init__(self, input_dim=20, d_model=128, n_heads=8, num_layers=4, d_ff=256):
        super(VolatilityTransformer, self).__init__()

        self.input_proj = nn.Linear(input_dim, d_model)
        self.pos_encoding = PositionalEncoding(d_model)
        self.encoder_layers = nn.ModuleList([
            TransformerBlock(d_model, n_heads, d_ff)
            for _ in range(num_layers)
        ])
        self.fc = nn.Linear(d_model, 1)

    def forward(self, x):
        # x: (batch, seq_len, features)
        x = self.input_proj(x)
        x = self.pos_encoding(x)

        for layer in self.encoder_layers:
            x = layer(x)

        x = x[:, -1, :]  # Use last time step
        x = self.fc(x)
        return x

7.2 Portfolio Optimisation with Transformers

class PortfolioTransformer(nn.Module):
    def __init__(self, n_assets, d_model=128, n_heads=8, num_layers=3):
        super(PortfolioTransformer, self).__init__()

        self.input_proj = nn.Linear(1, d_model)  # Each asset's return
        self.pos_encoding = PositionalEncoding(d_model)
        self.encoder_layers = nn.ModuleList([
            TransformerBlock(d_model, n_heads, 256)
            for _ in range(num_layers)
        ])

        # Output: portfolio weights
        self.fc = nn.Linear(d_model, n_assets)

    def forward(self, x):
        # x: (batch, seq_len, n_assets)
        # Transpose to (batch, seq_len, 1) for each asset
        # Or process all assets in parallel

        # Input projection
        x = self.input_proj(x.unsqueeze(-1))  # (batch, seq_len, n_assets, d_model)

        # Reshape to treat each asset as a token
        batch, seq_len, n_assets, d_model = x.shape
        x = x.view(batch, seq_len * n_assets, d_model)

        # Positional encoding
        x = self.pos_encoding(x)

        # Transformer
        for layer in self.encoder_layers:
            x = layer(x)

        # Aggregate over sequence
        x = x.view(batch, seq_len, n_assets, -1)
        x = x.mean(dim=1)  # (batch, n_assets, d_model)

        # Output weights
        weights = self.fc(x)  # (batch, n_assets)
        weights = F.softmax(weights, dim=-1)

        return weights

7.3 Temporal Fusion Transformer (TFT)

TFT is a specialised architecture for multi-horizon forecasting with static, known, and observed inputs.

class TemporalFusionTransformer(nn.Module):
    """
    Simplified Temporal Fusion Transformer for financial forecasting.
    """
    def __init__(self, input_dim, d_model, n_heads, num_layers, output_dim, horizon=5):
        super(TemporalFusionTransformer, self).__init__()

        self.horizon = horizon

        # Input projection
        self.input_proj = nn.Linear(input_dim, d_model)

        # Static covariate encoder (e.g., asset characteristics)
        self.static_encoder = nn.Linear(10, d_model)

        # LSTM for local processing
        self.lstm = nn.LSTM(d_model, d_model, num_layers=2, batch_first=True)

        # Self-attention
        self.attention = MultiHeadAttention(d_model, n_heads)
        self.norm = nn.LayerNorm(d_model)

        # Output
        self.fc = nn.Linear(d_model, output_dim)

    def forward(self, x, static_features):
        # x: (batch, seq_len, features)
        # static_features: (batch, static_dim)

        # Project inputs
        x = self.input_proj(x)

        # Add static context
        static_context = self.static_encoder(static_features).unsqueeze(1)
        x = x + static_context

        # LSTM for local patterns
        lstm_out, _ = self.lstm(x)

        # Self-attention for long-range dependencies
        attn_out, _ = self.attention(lstm_out, lstm_out, lstm_out)
        attn_out = self.norm(lstm_out + attn_out)

        # Multi-horizon output
        # Take last horizon steps
        forecasts = self.fc(attn_out[:, -self.horizon:, :])

        return forecasts

8. Computational Complexity Analysis

Model	Time Complexity	Memory Complexity
RNN/LSTM	`O(T * d²)`	`O(T * d)`
Transformer	`O(T² * d)`	`O(T²)`
Transformer (Sparse)	`O(T * sqrt(T) * d)`	`O(T * sqrt(T))`

Financial Implication: Transformers are quadratic in sequence length. For very long sequences (>1000), this becomes prohibitive. Use sparse attention variants (Longformer, BigBird) or limit sequence length to 100-200.

9. Limitations of Transformers in Finance

Limitation	Solution
Quadratic Complexity	Use sparse attention, windowed attention, or Linformer.
No Inductive Bias	Transformers have no built-in understanding of time. Positional encoding is a weak bias.
Data Hungry	Transformers require large datasets. Small financial datasets may not benefit.
Overfitting	More parameters lead to overfitting on noisy financial data. Use strong regularisation.
Interpretability	Attention weights are not always faithful explanations. Use SHAP/LIME for interpretability.

10. Summary for the AI Practitioner

Self-Attention allows direct connections between all positions in the sequence. Attention(Q,K,V) = softmax(QK^T/sqrt(d_k)) V.
Multi-Head Attention runs multiple attention heads in parallel, capturing different relationship types.
Positional Encoding adds sequence order information. Use sinusoidal or learned encodings.
Transformers are parallelisable and capture long-range dependencies better than RNNs.
Temporal Fusion Transformer is a specialised architecture for multi-horizon financial forecasting.
Computational complexity is O(T² * d). Use sparse attention for long sequences.
Transformers are data-hungry. Use them when you have large datasets (e.g., tick data, multiple assets).
Regularisation is mandatory. Use dropout, weight decay, and early stopping.

Diploma in Artificial Intelligence in Finance Lesson 5.5: Transformers and Self-Attention for Financial Sequences

Diploma in Artificial Intelligence in Finance