Lesson 5.3: Convolutional Neural Networks (CNNs) for Financial Alternative Data

1. Learning Objectives

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

Understand the mathematical foundations of convolution operations and their application to financial data.
Derive the forward and backward passes for convolutional layers.
Implement 1D CNNs for financial time series (technical indicators, order flow).
Implement 2D CNNs for alternative data (satellite imagery, document processing).
Apply CNNs to anomaly detection in financial time series.
Understand the architecture of ResNet and its application to financial data.
Implement transfer learning with pre-trained CNNs for financial image data.
Understand the limitations of CNNs for financial applications.

2. The Mathematical Foundations of Convolution

2.1 Continuous Convolution
For two continuous functions f(t) and g(t):
(f * g)(t) = ∫_{-∞}^{∞} f(τ) g(t - τ) dτ

2.2 Discrete Convolution (1D)
For discrete sequences x[n] and w[n]:
(x * w)[n] = Σ_{k=-∞}^{∞} x[k] w[n - k]

2.3 Discrete Convolution (2D)
For a 2D image X and a kernel W:
(X * W)[i, j] = Σ_{m} Σ_{n} X[i+m, j+n] W[m, n]

2.4 The Convolution Theorem
The Fourier transform of a convolution is the pointwise product of Fourier transforms:
F{f * g} = F{f} · F{g}

Financial Implication: Convolutions are linear operations that can be applied efficiently in the frequency domain. This is used in high-frequency trading for fast feature extraction.

2.5 Cross-Correlation vs. Convolution
In deep learning, what is called “convolution” is actually cross-correlation (no flipping):
(X ★ W)[i, j] = Σ_{m} Σ_{n} X[i+m, j+n] W[m, n]

This is what CNNs implement. The difference is irrelevant for learned filters.

3. 1D Convolutional Neural Networks for Financial Time Series

3.1 Architecture

Input: (batch_size, seq_len, n_features)
Conv1D: (batch_size, seq_len - kernel_size + 1, n_filters)
Pooling: (batch_size, seq_len / pool_size, n_filters)
Flatten: (batch_size, seq_len * n_filters)
Dense: (batch_size, output_dim)

3.2 1D Convolution Implementation

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

class Conv1DFinancial(nn.Module):
    def __init__(self, input_channels, seq_length, n_filters=64, kernel_size=3, pool_size=2):
        super(Conv1DFinancial, self).__init__()

        self.conv1 = nn.Conv1d(
            in_channels=input_channels,
            out_channels=n_filters,
            kernel_size=kernel_size,
            padding='same'
        )
        self.bn1 = nn.BatchNorm1d(n_filters)
        self.pool1 = nn.MaxPool1d(pool_size)

        self.conv2 = nn.Conv1d(
            in_channels=n_filters,
            out_channels=n_filters * 2,
            kernel_size=kernel_size,
            padding='same'
        )
        self.bn2 = nn.BatchNorm1d(n_filters * 2)
        self.pool2 = nn.MaxPool1d(pool_size)

        # Calculate flattened size
        self.flatten_size = self._calculate_flatten_size(input_channels, seq_length, n_filters, kernel_size, pool_size)

        self.fc1 = nn.Linear(self.flatten_size, 128)
        self.dropout = nn.Dropout(0.3)
        self.fc2 = nn.Linear(128, 1)

    def _calculate_flatten_size(self, input_channels, seq_length, n_filters, kernel_size, pool_size):
        # After conv1 and pool1
        size = seq_length // pool_size
        # After conv2 and pool2
        size = size // pool_size
        # Channels: n_filters * 2
        return size * (n_filters * 2)

    def forward(self, x):
        # Input: (batch, seq_len, features) -> (batch, features, seq_len)
        x = x.permute(0, 2, 1)

        x = self.conv1(x)
        x = self.bn1(x)
        x = F.relu(x)
        x = self.pool1(x)

        x = self.conv2(x)
        x = self.bn2(x)
        x = F.relu(x)
        x = self.pool2(x)

        # Flatten
        x = x.reshape(x.size(0), -1)

        x = F.relu(self.fc1(x))
        x = self.dropout(x)
        x = self.fc2(x)

        return x.squeeze()

3.3 Financial Application – Technical Indicator Extraction

def conv_technical_indicators(prices, window_size=60):
    """
    Use 1D CNN to extract technical features from price data.
    """
    # Input: (batch, seq_len, features)
    # Features: Open, High, Low, Close, Volume

    class TechnicalCNN(nn.Module):
        def __init__(self, input_channels, seq_length):
            super(TechnicalCNN, self).__init__()
            # Multi-scale convolutions
            self.conv_short = nn.Conv1d(input_channels, 32, kernel_size=3, padding='same')
            self.conv_medium = nn.Conv1d(input_channels, 32, kernel_size=7, padding='same')
            self.conv_long = nn.Conv1d(input_channels, 32, kernel_size=14, padding='same')

            # Combine features
            self.fc = nn.Linear(32 * 3 * seq_length, 64)
            self.output = nn.Linear(64, 10)  # 10 technical indicators

        def forward(self, x):
            x = x.permute(0, 2, 1)

            short = F.relu(self.conv_short(x))
            medium = F.relu(self.conv_medium(x))
            long = F.relu(self.conv_long(x))

            combined = torch.cat([short, medium, long], dim=1)
            combined = combined.reshape(combined.size(0), -1)

            features = F.relu(self.fc(combined))
            indicators = self.output(features)

            return indicators

    model = TechnicalCNN(5, window_size)
    return model

4. 2D Convolutional Neural Networks for Alternative Data

4.1 Satellite Imagery – Retail Foot Traffic Prediction

class SatelliteCNN(nn.Module):
    def __init__(self, num_classes=1):
        super(SatelliteCNN, self).__init__()

        # Feature extraction layers
        self.conv1 = nn.Conv2d(3, 32, kernel_size=3, padding=1)
        self.bn1 = nn.BatchNorm2d(32)
        self.pool1 = nn.MaxPool2d(2)

        self.conv2 = nn.Conv2d(32, 64, kernel_size=3, padding=1)
        self.bn2 = nn.BatchNorm2d(64)
        self.pool2 = nn.MaxPool2d(2)

        self.conv3 = nn.Conv2d(64, 128, kernel_size=3, padding=1)
        self.bn3 = nn.BatchNorm2d(128)
        self.pool3 = nn.MaxPool2d(2)

        self.conv4 = nn.Conv2d(128, 256, kernel_size=3, padding=1)
        self.bn4 = nn.BatchNorm2d(256)
        self.pool4 = nn.MaxPool2d(2)

        # Global average pooling (instead of flatten)
        self.gap = nn.AdaptiveAvgPool2d(1)

        # Fully connected layers
        self.fc1 = nn.Linear(256, 128)
        self.dropout = nn.Dropout(0.3)
        self.fc2 = nn.Linear(128, num_classes)

    def forward(self, x):
        x = F.relu(self.bn1(self.conv1(x)))
        x = self.pool1(x)

        x = F.relu(self.bn2(self.conv2(x)))
        x = self.pool2(x)

        x = F.relu(self.bn3(self.conv3(x)))
        x = self.pool3(x)

        x = F.relu(self.bn4(self.conv4(x)))
        x = self.pool4(x)

        x = self.gap(x)
        x = x.reshape(x.size(0), -1)

        x = F.relu(self.fc1(x))
        x = self.dropout(x)
        x = self.fc2(x)

        return x

4.2 Financial Application – Parking Lot Occupancy Detection

def parking_lot_analysis(satellite_image):
    """
    Analyse satellite imagery for parking lot occupancy.
    """
    # Preprocessing
    # 1. Load image
    # 2. Normalise pixel values
    # 3. Resize to fixed size

    class ParkingLotCNN(nn.Module):
        def __init__(self):
            super(ParkingLotCNN, self).__init__()
            # Use pre-trained ResNet as feature extractor
            import torchvision.models as models
            self.backbone = models.resnet18(pretrained=True)
            self.backbone.fc = nn.Identity()  # Remove classification head

            # Custom head for occupancy prediction
            self.fc1 = nn.Linear(512, 128)
            self.fc2 = nn.Linear(128, 1)

        def forward(self, x):
            features = self.backbone(x)
            features = F.relu(self.fc1(features))
            occupancy = torch.sigmoid(self.fc2(features))
            return occupancy

    model = ParkingLotCNN()
    return model

4.3 Transfer Learning with Pre-trained CNNs

import torchvision.models as models
import torchvision.transforms as transforms

def transfer_learning_finance(num_classes=1):
    """
    Use pre-trained ResNet for financial image data.
    """
    # Load pre-trained model
    model = models.resnet50(pretrained=True)

    # Freeze all layers
    for param in model.parameters():
        param.requires_grad = False

    # Replace the final layer
    num_features = model.fc.in_features
    model.fc = nn.Sequential(
        nn.Linear(num_features, 256),
        nn.ReLU(),
        nn.Dropout(0.3),
        nn.Linear(256, num_classes)
    )

    # Unfreeze the last few layers
    for param in list(model.parameters())[-10:]:
        param.requires_grad = True

    return model

# Data augmentation for satellite images
train_transform = transforms.Compose([
    transforms.RandomHorizontalFlip(),
    transforms.RandomVerticalFlip(),
    transforms.RandomRotation(10),
    transforms.ColorJitter(brightness=0.2, contrast=0.2),
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])

val_transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225])
])

5. 1D CNN for Anomaly Detection in Financial Time Series

5.1 Autoencoder Architecture

class Conv1DAutoencoder(nn.Module):
    def __init__(self, input_channels, seq_length, latent_dim=16):
        super(Conv1DAutoencoder, self).__init__()

        # Encoder
        self.encoder = nn.Sequential(
            nn.Conv1d(input_channels, 32, kernel_size=3, padding=1),
            nn.ReLU(),
            nn.MaxPool1d(2),
            nn.Conv1d(32, 64, kernel_size=3, padding=1),
            nn.ReLU(),
            nn.MaxPool1d(2),
            nn.Conv1d(64, 128, kernel_size=3, padding=1),
            nn.ReLU(),
            nn.AdaptiveAvgPool1d(1)
        )

        # Latent space
        self.fc_latent = nn.Linear(128, latent_dim)

        # Decoder
        self.fc_decode = nn.Linear(latent_dim, 128 * (seq_length // 4))

        self.decoder = nn.Sequential(
            nn.ConvTranspose1d(128, 64, kernel_size=4, stride=2, padding=1),
            nn.ReLU(),
            nn.ConvTranspose1d(64, 32, kernel_size=4, stride=2, padding=1),
            nn.ReLU(),
            nn.Conv1d(32, input_channels, kernel_size=3, padding=1)
        )

    def forward(self, x):
        # Input: (batch, seq_len, features) -> (batch, features, seq_len)
        x = x.permute(0, 2, 1)

        # Encode
        encoded = self.encoder(x)
        encoded = encoded.reshape(encoded.size(0), -1)
        latent = self.fc_latent(encoded)

        # Decode
        decoded = self.fc_decode(latent)
        decoded = decoded.reshape(decoded.size(0), 128, -1)
        decoded = self.decoder(decoded)

        # Output: (batch, features, seq_len) -> (batch, seq_len, features)
        decoded = decoded.permute(0, 2, 1)

        return decoded

def detect_anomalies_conv1d(data, threshold=3.0):
    """
    Detect anomalies using reconstruction error.
    """
    model = Conv1DAutoencoder(input_channels=data.shape[-1], seq_length=data.shape[1])
    model.eval()

    with torch.no_grad():
        reconstructed = model(torch.FloatTensor(data))
        reconstruction_error = torch.mean((data - reconstructed)**2, dim=(1, 2))

    anomaly_mask = reconstruction_error > (reconstruction_error.mean() + threshold * reconstruction_error.std())
    return anomaly_mask.numpy()

6. Dilated Convolutions – Increasing Receptive Field

Dilated convolutions expand the receptive field without increasing the number of parameters.

6.1 Mathematical Formulation
For a dilation rate d:
(X * W)[i, j] = Σ_{m} Σ_{n} X[i + d*m, j + d*n] W[m, n]

6.2 Implementation

class DilatedConv1D(nn.Module):
    def __init__(self, input_channels, n_filters, dilations=[1, 2, 4, 8]):
        super(DilatedConv1D, self).__init__()

        self.convs = nn.ModuleList()
        for dilation in dilations:
            self.convs.append(
                nn.Conv1d(input_channels, n_filters, kernel_size=3, dilation=dilation, padding=dilation)
            )

        self.fc = nn.Linear(n_filters * len(dilations), 1)

    def forward(self, x):
        x = x.permute(0, 2, 1)

        outputs = []
        for conv in self.convs:
            out = F.relu(conv(x))
            out = F.adaptive_avg_pool1d(out, 1)
            outputs.append(out)

        combined = torch.cat(outputs, dim=1).reshape(x.size(0), -1)
        return self.fc(combined).squeeze()

7. Summary for the AI Practitioner

Convolutions are linear operations that extract local patterns. In finance, they capture short-term price movements and technical patterns.
1D CNNs are ideal for time series data (price sequences, order flow). They are faster and have fewer parameters than LSTMs for short sequences.
2D CNNs process alternative data (satellite images, document scans). Use pre-trained models (ResNet) with transfer learning.
Dilated convolutions increase receptive field without increasing parameters. Useful for capturing multi-scale patterns in financial data.
Anomaly detection with convolutional autoencoders identifies unusual market behaviour and potential fraud.
Data augmentation (rotation, flipping, jittering) improves generalisation for limited datasets.
CNNs are not a panacea: They assume local stationarity and translational invariance, which may not hold in financial markets.