Graph Engineering for Quants : The Complete Masterclass
Financial markets are, at their core, relational systems. A stock does not move in isolation - it moves because its largest customer cut orders, because its sector peers reported earnings, because a sovereign credit event repriced its debt covenants, or because a shared institutional holder is liquidating across a correlated book. The return signal is not in the stock's own time series. It is in the web of relationships surrounding it.
Traditional quantitative models have always known this, but they have approximated it with blunt instruments: sector dummies, correlation matrices, factor loadings on common indices. These approximations collapse the true relational structure of markets into a handful of linear scalars and then lose everything else. Graph Neural Networks (GNNs) change this. They operate directly on the relational structure - stocks as nodes, their economic relationships as edges - and learn to propagate signals through the network in ways that capture momentum spillovers, supply chain contagion, volatility transmission, and cross-market dependencies that flat time-series models fundamentally cannot see.
This article covers the full spectrum: the mathematical foundations, every major GNN architecture used in production quantitative finance, five distinct use cases from stock prediction to fraud detection, and complete runnable PyTorch Geometric code for each one. Every result cited is from peer-reviewed research or documented production deployments.
about me : I am Venus (open-source-believer, so spitting out internal secrets on X), a Senior Quant Systems Architect and Backend Engineer experienced in building startups from 0→1 and scaling products from 1→100 across AI, cloud, and fintech x defi infrastructure. dm's are open to connect. Let's get back to article.
Why Graphs Belong in the Quant Toolkit
The empirical case is now clear and consistent across multiple research groups and time periods.
A deep GNN model integrating correlation structure and attention mechanism on CSI 300 constituent stocks from 2018 to 2023 achieved 78.9% directional accuracy with an RMSE of 0.0182 - a 23.5% improvement over traditional methods, with prediction accuracy staying above 75.8% during typical volatility periods.
A graph attention-based heterogeneous multi-agent deep reinforcement learning framework tested on S&P 500, NASDAQ 100, and Russell 2000 datasets achieved 16.8% annualized returns, a 1.34 Sharpe ratio, and 8.2% maximum drawdown, significantly outperforming traditional mean-variance optimization.
Research directly testing GCN and GAT architectures against traditional ML for systemic risk classification on financial networks reported a 94% MCC improvement for GNNs - the strongest published evidence that graph structure improves financial risk detection.
BlackRock has deployed turnover-constrained GNNs across $50 billion in ETF assets, with risk-adjusted returns improving by 15–393% through attention-based volatility scaling and cost-aware regularization reducing turnover by 20–40% without compromising performance.
These are not academic curiosities. They reflect a fundamental insight: inter-stock dependencies, supply chain connections, and the diffusion of market sentiment all constitute a structured graphical information system. Relying solely on traditional sequence models fails to capture the underlying interactions and structured dependencies among stocks. This has become a key factor limiting prediction accuracy in large-scale quantitative systems.
The Mathematical Foundation
Before writing any code, you need the math. A graph is defined as G = (V, E), where V is the set of nodes and E is the set of edges. In finance, nodes are typically stocks, companies, or financial institutions. Edges represent the relationships between them: correlation, supply chain links, shared ownership, regulatory exposure, or sector membership.
Each node v ∈ V carries a feature vector x_v ∈ ℝ^d. The adjacency matrix A ∈ ℝ^{|V|×|V|} encodes the edge structure. The degree matrix D is diagonal with D_{ii} = Σ_j A_{ij}.
2.1 Graph Convolutional Networks (GCN)
The GCN layer, introduced by Kipf and Welling (2017), performs the following update:
H^{(l+1)} = \sigma\!\left(\tilde{D}^{-1/2}\, \tilde{A}\, \tilde{D}^{-1/2}\, H^{(l)}\, W^{(l)}\right)
where à = A + I is the adjacency matrix with added self-loops, D̃ is the corresponding degree matrix, H^{(l)} is the node feature matrix at layer l, W^{(l)} is the learnable weight matrix, and σ is an activation function. The symmetric normalization D̃^{-1/2} à D̃^{-1/2} ensures that aggregation is normalized by both source and target node degree, preventing scale issues when nodes have wildly different numbers of neighbors - which is common in financial networks where major index components have many more correlated peers than small-caps.
2.2 Graph Attention Networks (GAT)
GCN treats all neighbors equally. GAT, introduced by Veličković et al. (2018), computes attention weights between nodes:
\alpha_{ij} = \frac{\exp\!\left(\text{LeakyReLU}\!\left(\mathbf{a}^\top [\mathbf{W} h_i \,\|\, \mathbf{W} h_j]\right)\right)}{\sum_{k \in \mathcal{N}(i)} \exp\!\left(\text{LeakyReLU}\!\left(\mathbf{a}^\top [\mathbf{W} h_i \,\|\, \mathbf{W} h_k]\right)\right)}
The node update is then:
h_i^{(l+1)} = \sigma\!\left(\sum_{j \in \mathcal{N}(i)} \alpha_{ij}\, W\, h_j^{(l)}\right)
For stock prediction, this is transformative. The attention weight α_{ij} learns which relationships are actually predictive - a momentum spillover from semiconductor suppliers to device manufacturers matters more than a spurious correlation between two stocks that happened to co-move last quarter. The model discovers these weights from data rather than imposing them by hand.
2.3 Temporal Graph Networks
Static GNNs assume the graph structure is fixed. Financial networks are not. Correlations shift across regimes, supply chains restructure, and institutional ownership changes. Temporal Graph Networks (TGN) extend the GNN framework to continuous-time dynamic graphs where edges arrive as timestamped events. The TGN memory module maintains a state for each node that is updated whenever the node is involved in an interaction:
s_i(t) = \text{mem}\!\left(s_i(t^-),\; \text{msg}(s_i(t^-), s_j(t^-), \Delta t, e_{ij}(t))\right)
where s_i(t) is the memory state of node i at time t, msg is a message function, and e_{ij}(t) is the edge feature at time t. The node embedding is then computed by a graph attention aggregator on top of the memory states.
Use Case 1: Stock Return Prediction with Sector and Supply Chain Graphs
The foundational quant application. Stocks in the same sector and supply chain exhibit correlated return dynamics. A GNN encodes these relationships into the embedding space and uses them to predict next-period returns.
Graph construction: Nodes are stocks. Two types of edges are commonly used: sector edges connecting all stocks within the same GICS sector, and supply chain edges connecting companies with documented customer-supplier relationships (sourced from FactSet, Bloomberg Supply Chain, or SEC filings). Edge weights can be set uniformly (binary graph) or by relationship strength (revenue exposure percentage for supply chain edges, correlation coefficient for sector edges).
Inter-industry linkages, supply chain connections between companies, and the diffusion of market sentiment all constitute a structured graphical information system. GNNs can extract interaction information embedded in the graph structure through iterative updates of node features, capturing potential correlations that are often overlooked by traditional models.
"""
Stock Return Prediction with Graph Attention Network
Uses sector + correlation-based adjacency matrix
Dependencies: torch, torch_geometric, pandas, numpy, yfinance
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
import pandas as pd
import yfinance as yf
from torch_geometric.data import Data, DataLoader
from torch_geometric.nn import GATConv, GCNConv
from torch_geometric.utils import dense_to_sparse
from sklearn.preprocessing import StandardScaler
from sklearn.model_selection import TimeSeriesSplit
# ============================================================
# 1. Data Preparation and Graph Construction
# ============================================================
def fetch_price_data(tickers: list, start: str, end: str) -> pd.DataFrame:
"""
Fetch adjusted close prices for a list of tickers.
Returns a DataFrame with tickers as columns.
"""
raw = yf.download(tickers, start=start, end=end,
auto_adjust=True, progress=False)["Close"]
raw = raw.dropna(how="all", axis=1)
raw = raw.ffill().dropna()
return raw
def compute_features(prices: pd.DataFrame,
lookback: int = 20) -> pd.DataFrame:
"""
Compute node features for each stock at each time step.
Features: 20-day return, 5-day return, 20-day volatility,
RSI(14), volume-adjusted momentum proxy,
normalized price relative to 52-week high/low.
"""
features = pd.DataFrame(index=prices.index)
log_ret = np.log(prices / prices.shift(1))
for col in prices.columns:
r = log_ret[col]
features[f"{col}_ret20"] = r.rolling(lookback).sum()
features[f"{col}_ret5"] = r.rolling(5).sum()
features[f"{col}_vol20"] = r.rolling(lookback).std()
# RSI (14-day)
delta = r.diff()
gain = delta.clip(lower=0).rolling(14).mean()
loss = (-delta.clip(upper=0)).rolling(14).mean()
rs = gain / (loss + 1e-8)
features[f"{col}_rsi14"] = 100 - (100 / (1 + rs))
# 52-week range position
high52 = prices[col].rolling(252, min_periods=20).max()
low52 = prices[col].rolling(252, min_periods=20).min()
features[f"{col}_range_pos"] = (
(prices[col] - low52) / (high52 - low52 + 1e-8)
)
return features.dropna()
def build_correlation_adjacency(returns: pd.DataFrame,
threshold: float = 0.3,
window: int = 60) -> np.ndarray:
"""
Build adjacency matrix from rolling Pearson correlations.
Edge exists if |corr| > threshold.
Returns: binary adjacency matrix of shape (n_stocks, n_stocks).
Note: adjacency is computed on training data only to avoid
look-ahead bias when used in rolling walk-forward.
"""
corr_matrix = returns.tail(window).corr().values
adj = (np.abs(corr_matrix) > threshold).astype(float)
np.fill_diagonal(adj, 1.0) # self-loops
return adj
def build_sector_adjacency(tickers: list,
sector_map: dict) -> np.ndarray:
"""
Build adjacency matrix from sector membership.
Two stocks in the same GICS sector are connected.
sector_map: {ticker: sector_string}
"""
n = len(tickers)
adj = np.zeros((n, n))
for i, t_i in enumerate(tickers):
for j, t_j in enumerate(tickers):
if i == j:
adj[i, j] = 1.0 # self-loop
elif sector_map.get(t_i) == sector_map.get(t_j):
adj[i, j] = 1.0
return adj
def combine_adjacency(adj_corr: np.ndarray,
adj_sector: np.ndarray,
w_corr: float = 0.5,
w_sector: float = 0.5) -> np.ndarray:
"""Combine correlation and sector adjacency matrices."""
combined = w_corr * adj_corr + w_sector * adj_sector
# Binarize: edge exists if combined weight > 0.3
return (combined > 0.3).astype(float)
def build_graph_dataset(features_np: np.ndarray,
adj: np.ndarray,
targets: np.ndarray,
n_stocks: int,
n_features_per_stock: int) -> list:
"""
Convert feature arrays into PyTorch Geometric Data objects.
One Data object per time step.
features_np: (T, n_stocks * n_features_per_stock)
adj: (n_stocks, n_stocks)
targets: (T, n_stocks) — next-period returns
"""
edge_index, _ = dense_to_sparse(
torch.FloatTensor(adj)
) # (2, num_edges)
dataset = []
for t in range(len(features_np)):
# Node features: (n_stocks, n_features_per_stock)
x = torch.FloatTensor(
features_np[t].reshape(n_stocks, n_features_per_stock)
)
y = torch.FloatTensor(targets[t]) # (n_stocks,)
data = Data(x=x, edge_index=edge_index, y=y)
dataset.append(data)
return dataset
# ============================================================
# 2. Graph Attention Network Architecture
# ============================================================
class StockGAT(nn.Module):
"""
Multi-layer Graph Attention Network for stock return prediction.
Architecture:
Input node features → GAT Layer 1 → GAT Layer 2 →
Residual connection → LayerNorm → MLP head →
Per-node return predictions
"""
def __init__(self,
in_channels: int,
hidden_channels: int = 64,
out_channels: int = 1,
n_heads: int = 4,
dropout: float = 0.2):
super().__init__()
self.gat1 = GATConv(
in_channels=in_channels,
out_channels=hidden_channels,
heads=n_heads,
dropout=dropout,
concat=True
)
# After concat: hidden_channels * n_heads features per node
self.gat2 = GATConv(
in_channels=hidden_channels * n_heads,
out_channels=hidden_channels,
heads=1,
dropout=dropout,
concat=False
)
# Residual projection (maps input dim to hidden dim)
self.residual_proj = nn.Linear(in_channels, hidden_channels)
self.norm = nn.LayerNorm(hidden_channels)
self.dropout = nn.Dropout(dropout)
# Prediction head: (hidden_channels -> 32 -> out_channels)
self.head = nn.Sequential(
nn.Linear(hidden_channels, 32),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(32, out_channels)
)
def forward(self, x, edge_index):
"""
Args:
x: (num_nodes, in_channels)
edge_index: (2, num_edges)
Returns:
pred: (num_nodes, out_channels)
"""
residual = self.residual_proj(x) # (N, hidden)
h = self.gat1(x, edge_index) # (N, hidden * heads)
h = F.elu(h)
h = self.dropout(h)
h = self.gat2(h, edge_index) # (N, hidden)
h = h + residual # residual connection
h = self.norm(h)
h = F.elu(h)
pred = self.head(h) # (N, 1)
return pred.squeeze(-1) # (N,)# ============================================================
# 3. Hybrid GCN + LSTM for Temporal-Structural Modeling
# ============================================================
class GCN_LSTM(nn.Module):
"""
Hybrid model: GCN extracts structural node embeddings,
LSTM models temporal dynamics across the sequence.
Based on the architecture used in GCN-LSTM papers
for ES-mini and VX Futures Forecasting (2024).
Input: Sequence of graph snapshots (T, N, F)
Output: Per-node predictions at the final time step (N, 1)
"""
def __init__(self,
in_channels: int,
gcn_hidden: int = 32,
lstm_hidden: int = 64,
lstm_layers: int = 2,
out_channels: int = 1,
dropout: float = 0.2):
super().__init__()
self.gcn = GCNConv(in_channels, gcn_hidden)
self.lstm = nn.LSTM(
input_size=gcn_hidden,
hidden_size=lstm_hidden,
num_layers=lstm_layers,
batch_first=True,
dropout=dropout if lstm_layers > 1 else 0.0
)
self.head = nn.Linear(lstm_hidden, out_channels)
self.dropout = nn.Dropout(dropout)
def forward(self, x_seq, edge_index):
"""
Args:
x_seq: (T, N, F) — T time steps, N nodes, F features
edge_index: (2, E) — static graph structure
Returns:
pred: (N, out_channels)
"""
T, N, F = x_seq.shape
gcn_outs = []
for t in range(T):
h_t = F_act(self.gcn(x_seq[t], edge_index)) # (N, gcn_hidden)
gcn_outs.append(h_t)
# Stack to (N, T, gcn_hidden) then pass through LSTM
gcn_stack = torch.stack(gcn_outs, dim=1) # (N, T, gcn_hidden)
gcn_stack = self.dropout(gcn_stack)
lstm_out, _ = self.lstm(gcn_stack) # (N, T, lstm_hidden)
# Take the last time step
last = lstm_out[:, -1, :] # (N, lstm_hidden)
pred = self.head(last) # (N, out_channels)
return pred
# Use F_act to avoid shadowing the F.elu import
F_act = F.elu
# ============================================================
# 4. Training and Evaluation Utilities
# ============================================================
def information_coefficient(pred: np.ndarray,
target: np.ndarray) -> float:
"""
Pearson correlation between predicted and realized returns
across all stocks in a given time period.
The standard evaluation metric for cross-sectional alpha models.
"""
if pred.std() < 1e-8 or target.std() < 1e-8:
return 0.0
return float(np.corrcoef(pred, target)[0, 1])
def rank_ic(pred: np.ndarray, target: np.ndarray) -> float:
"""
Spearman rank correlation (RankIC).
More robust than IC to outlier returns.
"""
from scipy.stats import spearmanr
rho, _ = spearmanr(pred, target)
return float(rho)
def directional_accuracy(pred: np.ndarray,
target: np.ndarray) -> float:
"""Fraction of stocks where sign(pred) == sign(target)."""
return float(np.mean(np.sign(pred) == np.sign(target)))
def train_epoch(model, loader, optimizer,
device, loss_fn=None):
"""Single training epoch over a DataLoader of graph snapshots."""
model.train()
total_loss = 0.0
if loss_fn is None:
loss_fn = nn.MSELoss()
for batch in loader:
batch = batch.to(device)
optimizer.zero_grad()
pred = model(batch.x, batch.edge_index)
loss = loss_fn(pred, batch.y)
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
optimizer.step()
total_loss += loss.item()
return total_loss / len(loader)
@torch.no_grad()
def evaluate(model, loader, device):
"""Evaluate model; return IC, RankIC, directional accuracy."""
model.eval()
all_pred, all_true = [], []
for batch in loader:
batch = batch.to(device)
pred = model(batch.x, batch.edge_index).cpu().numpy()
true = batch.y.cpu().numpy()
all_pred.append(pred)
all_true.append(true)
preds = np.concatenate(all_pred)
truths = np.concatenate(all_true)
return {
"IC": information_coefficient(preds, truths),
"RankIC": rank_ic(preds, truths),
"DirAcc": directional_accuracy(preds, truths)
}
# ============================================================
# 5. Full Training Pipeline
# ============================================================
def run_stock_prediction_pipeline():
"""
End-to-end pipeline for stock return prediction with GAT.
Uses S&P 500 tech sector as a demonstration universe.
"""
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Device: {device}")
# --- Universe and sector map ---
tickers = [
"AAPL", "MSFT", "NVDA", "GOOGL", "META",
"AMZN", "TSLA", "AVGO", "ORCL", "ADBE",
"AMD", "INTC", "QCOM", "TXN", "MU"
]
sector_map = {t: "Technology" for t in tickers}
# In production, use a proper GICS sector database
# and include cross-sector edges
# --- Fetch data ---
prices = fetch_price_data(tickers, start="2019-01-01", end="2024-12-31")
tickers = [t for t in tickers if t in prices.columns] # drop failures
n_stocks = len(tickers)
print(f"Stocks: {n_stocks}, Trading days: {len(prices)}")
# --- Compute features ---
features = compute_features(prices)
features = features.dropna()
n_feat_per_stock = 5 # ret20, ret5, vol20, rsi14, range_pos
# Align prices and features
common_idx = features.index.intersection(prices.index)
features = features.loc[common_idx]
prices = prices.loc[common_idx]
# --- Targets: next-period 5-day forward return ---
log_ret = np.log(prices / prices.shift(1))
fwd_ret = log_ret.shift(-5) # 5-day forward return
# Align features to have valid forward returns
valid_idx = fwd_ret.dropna().index
valid_idx = valid_idx.intersection(features.index)
features = features.loc[valid_idx]
fwd_ret = fwd_ret.loc[valid_idx]
# --- Build adjacency (on full sample for simplicity;
# in production compute rolling on training window only) ---
adj_corr = build_correlation_adjacency(log_ret.loc[valid_idx],
threshold=0.25,
window=120)
adj_sector = build_sector_adjacency(tickers, sector_map)
adj = combine_adjacency(adj_corr, adj_sector)
print(f"Graph: {n_stocks} nodes, {int(adj.sum())} edges")
# --- Normalize features ---
feat_array = features.values # (T, n_stocks * 5)
target_array = fwd_ret[tickers].values # (T, n_stocks)
scaler = StandardScaler()
feat_norm = scaler.fit_transform(feat_array)
# --- Build dataset ---
dataset = build_graph_dataset(
feat_norm, adj, target_array, n_stocks, n_feat_per_stock
)
# --- Walk-forward split: 70% train, 15% val, 15% test ---
T = len(dataset)
t1 = int(0.70 * T)
t2 = int(0.85 * T)
train_data = dataset[:t1]
val_data = dataset[t1:t2]
test_data = dataset[t2:]
print(f"Train: {t1}, Val: {t2-t1}, Test: {T-t2} snapshots")
train_loader = DataLoader(train_data, batch_size=32, shuffle=True)
val_loader = DataLoader(val_data, batch_size=32, shuffle=False)
test_loader = DataLoader(test_data, batch_size=32, shuffle=False)
# --- Initialise model ---
model = StockGAT(
in_channels=n_feat_per_stock,
hidden_channels=64,
out_channels=1,
n_heads=4,
dropout=0.2
).to(device)
n_params = sum(p.numel() for p in model.parameters()
if p.requires_grad)
print(f"Model parameters: {n_params:,}")
optimizer = torch.optim.AdamW(
model.parameters(), lr=1e-3, weight_decay=1e-4
)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(
optimizer, T_max=100
)
# --- Training loop ---
best_val_ic = -np.inf
print("\nTraining StockGAT...")
for epoch in range(1, 101):
train_loss = train_epoch(model, train_loader, optimizer, device)
val_metrics = evaluate(model, val_loader, device)
scheduler.step()
if val_metrics["IC"] > best_val_ic:
best_val_ic = val_metrics["IC"]
torch.save(model.state_dict(), "best_stock_gat.pt")
if epoch % 20 == 0:
print(
f"Epoch {epoch:3d} | Loss {train_loss:.4f} | "
f"Val IC {val_metrics['IC']:.4f} | "
f"Val RankIC {val_metrics['RankIC']:.4f} | "
f"Val DirAcc {val_metrics['DirAcc']:.3f}"
)
# --- Test evaluation ---
model.load_state_dict(torch.load("best_stock_gat.pt"))
test_metrics = evaluate(model, test_loader, device)
print(f"\nTest IC: {test_metrics['IC']:.4f}")
print(f"Test RankIC: {test_metrics['RankIC']:.4f}")
print(f"Test DirAcc: {test_metrics['DirAcc']:.3f}")
return model
if __name__ == "__main__":
model = run_stock_prediction_pipeline()Use Case 2: Volatility Forecasting with Dynamic Graph Structure
Volatility spillovers between markets are directional and regime-dependent. A shock in the US equity market propagates differently to European and Asian markets depending on the macro regime. Static correlation matrices miss this directionality. A Temporal GAT built on a Diebold-Yilmaz spillover network captures it.
The Temporal Graph Attention Network (Temporal GAT), combining GCN and GAT components, captures the temporal and structural dynamics of volatility spillovers. Adjacency matrices are constructed using either Pearson correlations among realized volatilities (giving a symmetric graph) or the Diebold-Yilmaz volatility spillover index computed via variance decomposition of VAR models (giving a directed graph that captures how volatility from one market affects another). Empirical results from a 15-year study of eight major global indices show that the Temporal GAT outperforms traditional GARCH models and other machine learning methods.
The key design requirement from the literature: adjacency matrices must be constructed separately for training, validation, and test sets, using only information available within each respective period. No future observations beyond the boundaries of a given partition are used when defining the graph topology for that period. This is the graph-structure version of look-ahead bias. Violating it invalidates the entire backtest.
"""
Volatility Forecasting with Temporal GAT
Dynamic adjacency via rolling Pearson correlation
Dependencies: torch, torch_geometric, numpy, pandas, scipy
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
import pandas as pd
from torch_geometric.nn import GATConv
from torch_geometric.utils import dense_to_sparse
# ============================================================
# 1. Dynamic Adjacency Builder
# ============================================================
def compute_rolling_adjacency(realized_vol: np.ndarray,
window: int = 60,
threshold: float = 0.3
) -> list[np.ndarray]:
"""
Build a rolling adjacency matrix for each time step.
realized_vol: (T, N) — realized volatility for N assets
Returns list of N×N adjacency matrices, one per time step.
CRITICAL: adjacency at step t uses ONLY data up to t
(no look-ahead bias in graph structure).
"""
T, N = realized_vol.shape
adjacencies = []
for t in range(T):
start = max(0, t - window + 1)
window_data = realized_vol[start : t + 1] # (<=window, N)
if window_data.shape[0] < 10:
# Not enough data: use identity (self-loops only)
adj = np.eye(N)
else:
corr = np.corrcoef(window_data.T) # (N, N)
np.fill_diagonal(corr, 1.0)
adj = (np.abs(corr) > threshold).astype(float)
adjacencies.append(adj)
return adjacencies
# ============================================================
# 2. Temporal GAT for Volatility Forecasting
# ============================================================
class TemporalGAT_Vol(nn.Module):
"""
Temporal Graph Attention Network for volatility forecasting.
Architecture (following Temporal GAT for volatility spillovers):
- GATConv encodes spatial (cross-market) dependencies
- GRU captures temporal dynamics of node embeddings
- MLP head outputs one-step-ahead realized volatility
Input: (seq_len, N, F) per call — sequence of graph snapshots
Output: (N,) — next-step volatility forecast for each asset
"""
def __init__(self,
in_channels: int,
gat_hidden: int = 32,
gru_hidden: int = 64,
n_heads: int = 4,
dropout: float = 0.1):
super().__init__()
self.gat = GATConv(
in_channels=in_channels,
out_channels=gat_hidden,
heads=n_heads,
dropout=dropout,
concat=True
)
gat_out_dim = gat_hidden * n_heads
self.gru = nn.GRU(
input_size=gat_out_dim,
hidden_size=gru_hidden,
num_layers=2,
batch_first=True,
dropout=dropout
)
self.head = nn.Sequential(
nn.Linear(gru_hidden, 32),
nn.ReLU(),
nn.Linear(32, 1)
)
self.dropout = nn.Dropout(dropout)
def forward(self, x_seq: list, edge_indices: list):
"""
Args:
x_seq: list of T tensors, each (N, F)
edge_indices: list of T tensors, each (2, E_t) —
dynamic graph at each time step
Returns:
pred: (N,) volatility forecast
"""
N = x_seq[0].shape[0]
T = len(x_seq)
gat_embeds = []
for t in range(T):
h_t = self.gat(x_seq[t], edge_indices[t]) # (N, gat*heads)
h_t = F.elu(h_t)
h_t = self.dropout(h_t)
gat_embeds.append(h_t)
# Stack to (N, T, gat_hidden*heads)
stacked = torch.stack(gat_embeds, dim=1) # (N, T, H)
gru_out, _ = self.gru(stacked) # (N, T, gru_hidden)
last = gru_out[:, -1, :] # (N, gru_hidden)
pred = self.head(last).squeeze(-1) # (N,)
return F.softplus(pred) # Ensure non-negative volatility
# ============================================================
# 3. Synthetic Data Demo (replace with real realized vol data)
# ============================================================
def generate_synthetic_vol_data(T: int = 500,
N: int = 8,
seed: int = 42
) -> tuple[np.ndarray, np.ndarray]:
"""
Generate synthetic realized volatility data with
cross-asset spillover structure (for demonstration).
Returns: (realized_vol, features) both (T, N)
"""
rng = np.random.RandomState(seed)
# Base volatility processes (GARCH-like)
vol = np.zeros((T, N))
vol[0] = rng.uniform(0.01, 0.03, N)
for t in range(1, T):
shock = rng.randn(N) * 0.005
# Spillover: asset 0 affects assets 1 and 2
spillover = np.zeros(N)
spillover[1] += 0.3 * vol[t - 1, 0]
spillover[2] += 0.2 * vol[t - 1, 0]
vol[t] = np.clip(
0.85 * vol[t - 1] + spillover + np.abs(shock),
0.001, 0.15
)
# Features per node: [current vol, lag-1 vol, lag-5 vol mean,
# 30-day rolling mean vol]
features = np.stack([
vol,
np.roll(vol, 1, axis=0),
np.array([vol[max(0, t - 5):t + 1].mean(axis=0) for t in range(T)]),
np.array([vol[max(0, t - 30):t + 1].mean(axis=0) for t in range(T)])
], axis=-1) # (T, N, 4)
return vol, features
def run_volatility_pipeline():
"""
Demonstrate Temporal GAT volatility forecasting.
In production: replace synthetic data with realized vol
from CBOE, Bloomberg, or computed from intraday returns.
"""
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
T, N = 500, 8
SEQ_LEN = 10 # look-back window per forecast step
IN_CHANNELS = 4 # number of features per node
# Generate data
realized_vol, features = generate_synthetic_vol_data(T, N)
targets = np.roll(realized_vol, -1, axis=0) # next-step target
# Build dynamic adjacency (no look-ahead)
print("Building dynamic adjacency matrices...")
adjacencies = compute_rolling_adjacency(
realized_vol, window=60, threshold=0.25
)
# Prepare sequences
def build_sequences(start, end):
seqs, adj_seqs, tgt_list = [], [], []
for t in range(start, end - 1):
if t < SEQ_LEN:
continue
x_seq = [
torch.FloatTensor(features[t - SEQ_LEN + s])
for s in range(SEQ_LEN)
]
ei_seq = [
dense_to_sparse(torch.FloatTensor(adjacencies[t - SEQ_LEN + s]))[0]
for s in range(SEQ_LEN)
]
seqs.append(x_seq)
adj_seqs.append(ei_seq)
tgt_list.append(torch.FloatTensor(targets[t]))
return seqs, adj_seqs, tgt_list
split1 = int(0.70 * T)
split2 = int(0.85 * T)
train_seqs, train_adjs, train_tgts = build_sequences(0, split1)
val_seqs, val_adjs, val_tgts = build_sequences(split1, split2)
test_seqs, test_adjs, test_tgts = build_sequences(split2, T)
# Model
model = TemporalGAT_Vol(
in_channels=IN_CHANNELS,
gat_hidden=32,
gru_hidden=64,
n_heads=4,
dropout=0.1
).to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=5e-4)
def step(seqs, adjs, tgts, train=True):
"""Process a list of sequences (no batching for simplicity)."""
model.train(train)
total_loss = 0.0
with torch.set_grad_enabled(train):
for x_seq, ei_seq, y in zip(seqs, adjs, tgts):
x_dev = [x.to(device) for x in x_seq]
ei_dev = [e.to(device) for e in ei_seq]
y_dev = y.to(device)
pred = model(x_dev, ei_dev)
loss = F.mse_loss(pred, y_dev)
if train:
optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(
model.parameters(), 1.0
)
optimizer.step()
total_loss += loss.item()
return total_loss / len(seqs)
print("\nTraining TemporalGAT-Vol...")
for epoch in range(1, 51):
tr_loss = step(train_seqs, train_adjs, train_tgts, train=True)
val_loss = step(val_seqs, val_adjs, val_tgts, train=False)
if epoch % 10 == 0:
print(f"Epoch {epoch:3d} | Train MSE {tr_loss:.6f} | "
f"Val MSE {val_loss:.6f}")
test_loss = step(test_seqs, test_adjs, test_tgts, train=False)
print(f"\nTest MSE: {test_loss:.6f}")
if __name__ == "__main__":
run_volatility_pipeline()Use Case 3: Portfolio Optimization with GNN-Estimated Covariance
Mean-variance optimization requires a covariance matrix. The traditional sample covariance matrix is noisy, especially in large universes, and suffers from the well-documented curse of dimensionality. GNNs offer a different path: learn the latent graph structure of asset correlations and use GNN-estimated node embeddings to produce a structured covariance estimate.
Large-scale time-varying portfolio optimization using Graph Attention Networks handles high-dimensional data and accommodates customized layers for specific purposes, making them appealing for large-scale problems. The framework explicitly models time-varying asset correlations by updating the graph structure at each rebalancing period.
Cost-aware regularization in GNN-based portfolio optimization reduces turnover by 20–40% without compromising performance - a critical property for production portfolios where transaction costs erode theoretical alpha.
"""
GNN-Based Portfolio Optimization
Learns asset embeddings from graph structure, then optimizes weights
Dependencies: torch, torch_geometric, numpy, scipy
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
from scipy.optimize import minimize
from torch_geometric.nn import GATConv
from torch_geometric.utils import dense_to_sparse
# ============================================================
# 1. GNN Embedding Model
# ============================================================
class AssetEmbeddingGNN(nn.Module):
"""
Encodes each asset into a latent embedding using GAT.
The embeddings are used to estimate a structured
covariance matrix for portfolio optimization.
"""
def __init__(self,
in_channels: int,
embed_dim: int = 32,
n_heads: int = 4,
dropout: float = 0.1):
super().__init__()
self.gat1 = GATConv(in_channels, embed_dim,
heads=n_heads, concat=True,
dropout=dropout)
self.gat2 = GATConv(embed_dim * n_heads, embed_dim,
heads=1, concat=False,
dropout=dropout)
self.norm = nn.LayerNorm(embed_dim)
self.dropout = nn.Dropout(dropout)
def forward(self, x, edge_index):
h = F.elu(self.gat1(x, edge_index))
h = self.dropout(h)
h = self.gat2(h, edge_index)
h = self.norm(h)
return h # (N, embed_dim)
class ReturnPredictionHead(nn.Module):
"""Predicts expected returns from asset embeddings."""
def __init__(self, embed_dim: int):
super().__init__()
self.fc = nn.Sequential(
nn.Linear(embed_dim, 16),
nn.ReLU(),
nn.Linear(16, 1)
)
def forward(self, embeddings):
return self.fc(embeddings).squeeze(-1) # (N,)
def gnn_covariance(embeddings: torch.Tensor,
returns: torch.Tensor,
alpha: float = 0.1) -> torch.Tensor:
"""
Construct a structured covariance matrix from GNN embeddings.
The covariance is decomposed as:
Sigma = L * L^T + diag(epsilon)
where L is a low-rank factor matrix derived from embeddings.
This produces a positive-definite covariance estimate
that respects the graph structure learned by the GNN.
args:
embeddings: (N, D) — GNN node embeddings
returns: (T, N) — historical returns for residual estimation
alpha: regularization strength for diagonal component
"""
N, D = embeddings.shape
# Low-rank component from embeddings
# Normalize so dot-product gives correlation-like structure
emb_norm = F.normalize(embeddings, dim=-1) # (N, D)
L = emb_norm # (N, D)
low_rank_cov = L @ L.T # (N, N)
# Residual variance from historical data
ret_t = torch.FloatTensor(returns)
residual_var = ret_t.var(dim=0) # (N,)
diag_cov = torch.diag(residual_var * alpha) # (N, N)
# Combine: structured + residual diagonal
cov = low_rank_cov + diag_cov
# Ensure positive definiteness with small ridge
cov = cov + 1e-4 * torch.eye(N)
return cov
# ============================================================
# 2. Portfolio Optimization Layer
# ============================================================
def mean_variance_optimize(mu: np.ndarray,
Sigma: np.ndarray,
gamma: float = 1.0,
lambda_tc: float = 0.001,
w_prev: np.ndarray = None,
long_only: bool = True
) -> np.ndarray:
"""
Solve mean-variance optimization:
max mu^T w - (gamma/2) w^T Sigma w - lambda_tc ||w - w_prev||_1
s.t. sum(w) = 1, w >= 0 (if long_only)
The L1 transaction cost term penalizes turnover,
equivalent to a proportional cost model.
"""
N = len(mu)
if w_prev is None:
w_prev = np.ones(N) / N
def neg_utility(w):
ret_term = mu @ w
risk_term = 0.5 * gamma * (w @ Sigma @ w)
tc_term = lambda_tc * np.abs(w - w_prev).sum()
return -(ret_term - risk_term - tc_term)
def neg_utility_grad(w):
ret_grad = mu
risk_grad = gamma * Sigma @ w
tc_grad = lambda_tc * np.sign(w - w_prev)
return -(ret_grad - risk_grad - tc_grad)
constraints = [{"type": "eq", "fun": lambda w: w.sum() - 1.0}]
bounds = [(0.0, 1.0)] * N if long_only else [(-0.3, 0.3)] * N
result = minimize(
neg_utility,
x0=w_prev,
jac=neg_utility_grad,
method="SLSQP",
bounds=bounds,
constraints=constraints,
options={"maxiter": 500, "ftol": 1e-9}
)
return result.x if result.success else w_prev# ============================================================
# 3. End-to-End GNN Portfolio Pipeline
# ============================================================
class GNNPortfolioModel(nn.Module):
"""
Joint model: GNN embedding + return prediction + covariance.
Trained with a portfolio-aware Sharpe ratio loss.
"""
def __init__(self, in_channels: int, embed_dim: int = 32,
n_heads: int = 4, dropout: float = 0.1):
super().__init__()
self.encoder = AssetEmbeddingGNN(
in_channels, embed_dim, n_heads, dropout
)
self.return_head = ReturnPredictionHead(embed_dim)
def forward(self, x, edge_index):
embeddings = self.encoder(x, edge_index) # (N, embed_dim)
mu_pred = self.return_head(embeddings) # (N,)
return mu_pred, embeddings
def portfolio_loss(mu_pred: torch.Tensor,
mu_true: torch.Tensor,
embeddings: torch.Tensor,
returns_hist: torch.Tensor,
gamma: float = 1.0) -> torch.Tensor:
"""
Portfolio-aware loss combining:
1. Return prediction MSE
2. Negative Sharpe ratio of the implied portfolio
Gradient flows through both terms into the GNN.
"""
N = mu_pred.shape[0]
# Term 1: Return prediction accuracy
mse_loss = F.mse_loss(mu_pred, mu_true)
# Term 2: Portfolio Sharpe (differentiable approximation)
# Compute soft-max weights (differentiable)
w = torch.softmax(mu_pred / (mu_pred.std() + 1e-8), dim=0) # (N,)
# Portfolio returns on historical window
port_ret = returns_hist @ w # (T,)
sharpe = port_ret.mean() / (port_ret.std() + 1e-8)
# Combined loss: minimize MSE, maximize Sharpe
return mse_loss - 0.1 * sharpe
def run_portfolio_pipeline():
"""
End-to-end GNN portfolio optimization demo.
Synthetic data: replace with real price data in production.
"""
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# Synthetic setup
N = 20 # number of assets
T = 252 # days of history
F = 8 # features per asset
rng = np.random.RandomState(0)
# Random features (in production: returns, vol, RSI, etc.)
features_np = rng.randn(T, N, F).astype(np.float32)
returns_np = (rng.randn(T, N) * 0.01).astype(np.float32)
# Correlation-based adjacency
corr = np.corrcoef(returns_np.T)
adj = (np.abs(corr) > 0.2).astype(np.float32)
np.fill_diagonal(adj, 1.0)
edge_index, _ = dense_to_sparse(torch.FloatTensor(adj))
edge_index = edge_index.to(device)
# Model
model = GNNPortfolioModel(
in_channels=F, embed_dim=32, n_heads=4, dropout=0.1
).to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
# Training: predict next-day returns using current graph
LOOKBACK = 60
weights_history = []
w_prev = np.ones(N) / N
print("Training GNN Portfolio Model...")
for epoch in range(1, 51):
total_loss = 0.0
model.train()
for t in range(LOOKBACK, T - 1):
x_t = torch.FloatTensor(features_np[t]).to(device) # (N, F)
ret_hist = torch.FloatTensor(
returns_np[t - LOOKBACK : t]
).to(device) # (LB, N)
ret_next = torch.FloatTensor(returns_np[t + 1]).to(device) # (N,)
mu_pred, embeddings = model(x_t, edge_index)
loss = portfolio_loss(mu_pred, ret_next, embeddings,
ret_hist, gamma=1.0)
optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
optimizer.step()
total_loss += loss.item()
if epoch % 10 == 0:
print(f"Epoch {epoch:3d} | Loss {total_loss / (T - LOOKBACK - 1):.4f}")
# --- Walk-forward backtest ---
print("\nRunning walk-forward portfolio backtest...")
model.eval()
portfolio_returns = []
with torch.no_grad():
for t in range(LOOKBACK, T - 1):
x_t = torch.FloatTensor(features_np[t]).to(device)
ret_hist = returns_np[t - LOOKBACK : t]
mu_pred, embeddings = model(x_t, edge_index)
mu_np = mu_pred.cpu().numpy()
emb_tens = embeddings
cov_t = gnn_covariance(emb_tens, ret_hist).numpy()
# Portfolio weights
w = mean_variance_optimize(
mu_np, cov_t, gamma=1.0,
lambda_tc=0.001, w_prev=w_prev
)
w_prev = w
# Realized portfolio return
realized_ret = returns_np[t + 1] @ w
portfolio_returns.append(realized_ret)
port_ret = np.array(portfolio_returns)
equal_ret = returns_np[LOOKBACK + 1 :].mean(axis=1)
print(f"\nGNN Portfolio Sharpe: "
f"{port_ret.mean() / port_ret.std() * np.sqrt(252):.3f}")
print(f"Equal-Weight Sharpe: "
f"{equal_ret.mean() / equal_ret.std() * np.sqrt(252):.3f}")
print(f"GNN Portfolio Annual Ret: "
f"{port_ret.mean() * 252:.2%}")
print(f"GNN Portfolio Annual Vol: "
f"{port_ret.std() * np.sqrt(252):.2%}")
if __name__ == "__main__":
run_portfolio_pipeline()Use Case 4: Transaction Fraud Detection with Temporal Graph Networks
Financial transactions naturally form dynamic graphs: edges (transactions) arrive continuously, and node connectivity evolves over time. Real-time dynamic graph learning with temporal attention captures fine-grained temporal dynamics and evolving topological patterns that static graph models - which assume a fixed, unchanging graph - fundamentally cannot model for fraud detection.
GNNs are exceptionally adept at capturing complex relational patterns and dynamics within financial networks, significantly outperforming traditional fraud detection methods. GCN-based, GAT-based, and temporal GNN approaches have all demonstrated superior fraud detection compared to non-graph baselines.
The fraud detection graph is built as follows: nodes are accounts or entities, directed edges are transactions with features (amount, timestamp, transaction type, merchant category), and the task is node classification - is this account fraudulent? - or edge classification - is this transaction fraudulent?
"""
Financial Fraud Detection with Temporal Graph Network
Simplified TGN implementation for transaction networks
Dependencies: torch, torch_geometric, numpy, pandas
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
from torch_geometric.nn import GATConv, SAGEConv
from torch_geometric.data import Data
# ============================================================
# 1. Graph Construction from Transaction Data
# ============================================================
def build_transaction_graph(transactions: list[dict],
node_feature_dim: int = 8
) -> tuple[Data, dict]:
"""
Build a transaction graph from a list of transaction records.
Each transaction: {
'src': source account id,
'dst': destination account id,
'amount': float,
'timestamp': int (unix seconds),
'tx_type': int (0=wire, 1=ACH, 2=card, ...),
'label': int (0=legit, 1=fraud) — for training only
}
Returns PyG Data object and node-to-index mapping.
"""
# Collect unique nodes
nodes = sorted(set(
[t["src"] for t in transactions] +
[t["dst"] for t in transactions]
))
node_to_idx = {n: i for i, n in enumerate(nodes)}
N = len(nodes)
# Edge list
edge_src, edge_dst, edge_feats, edge_labels = [], [], [], []
for tx in transactions:
i = node_to_idx[tx["src"]]
j = node_to_idx[tx["dst"]]
edge_src.append(i)
edge_dst.append(j)
# Edge features: [log(amount), tx_type_onehot...,
# normalized_timestamp]
log_amt = np.log1p(tx["amount"])
tx_type = [0] * 4
if tx["tx_type"] < 4:
tx_type[tx["tx_type"]] = 1
t_norm = (tx["timestamp"] % 86400) / 86400 # time-of-day
edge_feats.append([log_amt] + tx_type + [t_norm])
edge_labels.append(tx.get("label", 0))
edge_index = torch.LongTensor([edge_src, edge_dst]) # (2, E)
edge_attr = torch.FloatTensor(edge_feats) # (E, 6)
edge_y = torch.LongTensor(edge_labels) # (E,)
# Node features: aggregate over incident transactions
# (in-degree, out-degree, mean amount sent, mean amount received,
# std amount sent, std amount received, tx count, recency)
node_feats = np.zeros((N, node_feature_dim), dtype=np.float32)
for tx in transactions:
i = node_to_idx[tx["src"]]
j = node_to_idx[tx["dst"]]
log_amt = np.log1p(tx["amount"])
# out-degree accumulation for source
node_feats[i, 0] += 1
node_feats[i, 2] += log_amt
# in-degree accumulation for destination
node_feats[j, 1] += 1
node_feats[j, 3] += log_amt
# Compute means
out_deg = node_feats[:, 0:1] + 1e-8
in_deg = node_feats[:, 1:2] + 1e-8
node_feats[:, 2] /= out_deg[:, 0]
node_feats[:, 3] /= in_deg[:, 0]
x = torch.FloatTensor(node_feats)
data = Data(x=x, edge_index=edge_index,
edge_attr=edge_attr, edge_y=edge_y)
return data, node_to_idx
# ============================================================
# 2. Fraud Detection GNN (Edge Classification)
# ============================================================
class FraudGNN(nn.Module):
"""
Graph-based fraud detection model.
Architecture:
- GraphSAGE layers aggregate neighborhood features
(SAGE is preferred over GCN for inductive tasks where
new nodes/transactions arrive at inference time)
- Edge classifier combines source, destination, and
edge feature embeddings to predict fraud probability
Based on architecture patterns from published fraud
detection GNN literature (Cheng et al., 2025;
Frontiers of Computer Science).
"""
def __init__(self,
node_in_channels: int,
edge_in_channels: int,
hidden_channels: int = 64,
dropout: float = 0.3):
super().__init__()
# Node encoder: 2-layer GraphSAGE
self.sage1 = SAGEConv(node_in_channels, hidden_channels)
self.sage2 = SAGEConv(hidden_channels, hidden_channels)
self.node_norm1 = nn.LayerNorm(hidden_channels)
self.node_norm2 = nn.LayerNorm(hidden_channels)
# Edge classifier: concat(h_src, h_dst, edge_feat)
edge_clf_in = hidden_channels * 2 + edge_in_channels
self.edge_clf = nn.Sequential(
nn.Linear(edge_clf_in, 64),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(64, 32),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(32, 2) # binary: legit vs fraud
)
self.dropout = nn.Dropout(dropout)
def encode_nodes(self, x, edge_index):
"""Produce node embeddings via GraphSAGE."""
h = F.relu(self.sage1(x, edge_index))
h = self.node_norm1(h)
h = self.dropout(h)
h = F.relu(self.sage2(h, edge_index))
h = self.node_norm2(h)
return h # (N, hidden_channels)
def classify_edges(self, h_nodes, edge_index, edge_attr):
"""
Predict fraud probability for each edge.
Combines source node embedding, destination node embedding,
and edge features.
"""
src_nodes = edge_index[0] # (E,)
dst_nodes = edge_index[1] # (E,)
h_src = h_nodes[src_nodes] # (E, hidden)
h_dst = h_nodes[dst_nodes] # (E, hidden)
edge_emb = torch.cat([h_src, h_dst, edge_attr], dim=-1) # (E, 2H+F)
logits = self.edge_clf(edge_emb) # (E, 2)
return logits
def forward(self, x, edge_index, edge_attr):
h_nodes = self.encode_nodes(x, edge_index)
logits = self.classify_edges(h_nodes, edge_index, edge_attr)
return logits# ============================================================
# 3. Fraud-Aware Loss (Class Imbalance Handling)
# ============================================================
def focal_loss(logits: torch.Tensor,
labels: torch.Tensor,
gamma: float = 2.0,
alpha: float = 0.25) -> torch.Tensor:
"""
Focal loss for imbalanced fraud detection.
Down-weights easy (mostly legit) examples so the model
focuses on hard fraud cases.
Fraud prevalence in real transaction data is typically
0.1%–2%, making standard cross-entropy sub-optimal.
Lin et al. (2017): Focal Loss for Dense Object Detection.
"""
probs = torch.softmax(logits, dim=-1)
pt = probs[range(len(labels)), labels]
ce_loss = F.cross_entropy(logits, labels, reduction="none")
fl = alpha * (1 - pt) ** gamma * ce_loss
return fl.mean()
# ============================================================
# 4. Synthetic Transaction Data Generator
# ============================================================
def generate_synthetic_transactions(n_accounts: int = 100,
n_transactions: int = 2000,
fraud_rate: float = 0.05,
seed: int = 42) -> list[dict]:
"""
Generate synthetic transaction data with ring-shaped
fraud patterns (a common money-laundering structure
that is invisible to per-transaction models but
visible to graph models).
"""
rng = np.random.RandomState(seed)
txs = []
# Identify a fraud ring: 5–10 accounts that cycle funds
ring_size = rng.randint(5, 11)
fraud_accounts = list(rng.choice(n_accounts, ring_size,
replace=False))
for i in range(n_transactions):
is_fraud = rng.random() < fraud_rate
if is_fraud and len(fraud_accounts) >= 2:
# Fraud: transaction within the ring
src = int(rng.choice(fraud_accounts))
dst = int(rng.choice([a for a in fraud_accounts
if a != src]))
amount = float(rng.uniform(500, 9999)) # structuring amount
else:
src = int(rng.randint(0, n_accounts))
dst = int(rng.randint(0, n_accounts))
while dst == src:
dst = int(rng.randint(0, n_accounts))
amount = float(rng.lognormal(mean=5.0, sigma=1.5))
txs.append({
"src": src,
"dst": dst,
"amount": amount,
"timestamp": int(1_700_000_000 + i * 300), # 5-min intervals
"tx_type": int(rng.randint(0, 4)),
"label": int(is_fraud)
})
return txs
def run_fraud_detection_pipeline():
"""
End-to-end fraud detection demo with ring pattern.
"""
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
# Generate data
transactions = generate_synthetic_transactions(
n_accounts=200, n_transactions=3000, fraud_rate=0.05
)
print(f"Total transactions: {len(transactions)}")
print(f"Fraud transactions: {sum(t['label'] for t in transactions)}")
print(f"Fraud rate: "
f"{sum(t['label'] for t in transactions) / len(transactions):.2%}")
# Build graph
data, _ = build_transaction_graph(transactions, node_feature_dim=8)
# Train/test split on edges (transactions)
n_edges = data.edge_index.shape[1]
split = int(0.8 * n_edges)
train_mask = torch.zeros(n_edges, dtype=torch.bool)
test_mask = torch.zeros(n_edges, dtype=torch.bool)
train_mask[:split] = True
test_mask[split:] = True
# Model
model = FraudGNN(
node_in_channels=8,
edge_in_channels=6,
hidden_channels=64,
dropout=0.3
).to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3,
weight_decay=1e-4)
x = data.x.to(device)
edge_index = data.edge_index.to(device)
edge_attr = data.edge_attr.to(device)
edge_y = data.edge_y.to(device)
print("\nTraining FraudGNN...")
for epoch in range(1, 101):
model.train()
optimizer.zero_grad()
logits = model(x, edge_index, edge_attr) # (E, 2)
loss = focal_loss(
logits[train_mask], edge_y[train_mask],
gamma=2.0, alpha=0.75
)
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
optimizer.step()
if epoch % 20 == 0:
model.eval()
with torch.no_grad():
test_logits = logits[test_mask]
test_labels = edge_y[test_mask]
preds = test_logits.argmax(dim=-1)
tp = ((preds == 1) & (test_labels == 1)).sum().item()
fp = ((preds == 1) & (test_labels == 0)).sum().item()
fn = ((preds == 0) & (test_labels == 1)).sum().item()
precision = tp / (tp + fp + 1e-8)
recall = tp / (tp + fn + 1e-8)
f1 = (2 * precision * recall
/ (precision + recall + 1e-8))
print(f"Epoch {epoch:3d} | Loss {loss.item():.4f} | "
f"Precision {precision:.3f} | "
f"Recall {recall:.3f} | F1 {f1:.3f}")
if __name__ == "__main__":
run_fraud_detection_pipeline()Use Case 5: Systemic Risk and Interbank Contagion
GNN-based systemic risk and interbank contagion analysis extends temporal GNN lines to U.S. regulatory surveillance. Research applying ST-GAT to 8,103 institutions over 58 quarters incorporates macro-conditioned edge weights, validated explainability modules, and grounding in U.S. regulatory filings. Temporal graph learning for default prediction integrating macroeconomic trends has reported 88.3% AUC.
The graph for systemic risk is a financial network where nodes are banks or financial institutions, edges represent bilateral exposures (interbank loans, derivative counterparty exposure, common asset holdings), and the task is predicting which nodes will experience distress in future periods given the current state of the network and macroeconomic inputs.
"""
Systemic Risk GNN: Interbank Contagion Monitoring
Simplified ST-GAT for financial network surveillance
Dependencies: torch, torch_geometric, numpy
"""
import torch
import torch.nn as nn
import torch.nn.functional as F
import numpy as np
from torch_geometric.nn import GATConv
from torch_geometric.utils import dense_to_sparse
class SystemicRiskGAT(nn.Module):
"""
Spatial-Temporal GAT for interbank contagion detection.
Architecture inspired by Temporal Attentive Graph Networks
for financial surveillance (Zhang et al., 2026):
- GAT encodes current network topology and node features
- GRU captures temporal dynamics of institution stress
- Macro features modulate edge weights
- Node-level binary classification: distress in next period
Node features: capital ratio, leverage ratio, NPL ratio,
liquidity coverage ratio, interbank exposure,
return on assets, CDS spread proxy
Edge features: bilateral exposure as fraction of Tier 1 capital
Macro inputs: VIX, credit spread, yield curve slope,
policy rate
"""
def __init__(self,
node_in: int = 7,
edge_in: int = 1,
macro_in: int = 4,
gat_hidden: int = 32,
gru_hidden: int = 64,
n_heads: int = 4,
dropout: float = 0.2):
super().__init__()
# Macro conditioning: projects macro features into
# a scalar gate applied to edge weights
self.macro_gate = nn.Sequential(
nn.Linear(macro_in, 16),
nn.ReLU(),
nn.Linear(16, 1),
nn.Sigmoid()
)
# Spatial: GAT with macro-conditioned edge attention
self.gat1 = GATConv(
in_channels=node_in,
out_channels=gat_hidden,
heads=n_heads,
concat=True,
edge_dim=edge_in,
dropout=dropout
)
self.gat2 = GATConv(
in_channels=gat_hidden * n_heads,
out_channels=gat_hidden,
heads=1,
concat=False,
edge_dim=edge_in,
dropout=dropout
)
# Temporal: GRU over time-series of GAT embeddings
self.gru = nn.GRU(
input_size=gat_hidden,
hidden_size=gru_hidden,
num_layers=2,
batch_first=True,
dropout=dropout
)
# Classifier: node-level distress prediction
self.classifier = nn.Sequential(
nn.Linear(gru_hidden, 32),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(32, 2) # 0=healthy, 1=distressed
)
self.norm = nn.LayerNorm(gat_hidden)
self.dropout = nn.Dropout(dropout)
def forward(self,
x_seq: list, # list of T: (N, node_in)
edge_index_seq: list, # list of T: (2, E)
edge_attr_seq: list, # list of T: (E, edge_in)
macro_seq: list # list of T: (macro_in,)
) -> torch.Tensor:
"""
Returns: (N, 2) — distress probability logits at final step.
"""
T = len(x_seq)
gat_embeds = []
for t in range(T):
x_t = x_seq[t]
ei_t = edge_index_seq[t]
ea_t = edge_attr_seq[t]
macro_t = macro_seq[t].unsqueeze(0) # (1, macro_in)
# Macro gate: scale edge features by macro regime
gate = self.macro_gate(macro_t) # (1, 1) in [0, 1]
ea_scaled = ea_t * gate # (E, edge_in)
h = F.elu(self.gat1(x_t, ei_t, ea_scaled))
h = self.dropout(h)
h = F.elu(self.gat2(h, ei_t, ea_scaled))
h = self.norm(h) # (N, gat_hidden)
gat_embeds.append(h)
# (N, T, gat_hidden) → GRU
stacked = torch.stack(gat_embeds, dim=1) # (N, T, H)
gru_out, _ = self.gru(stacked) # (N, T, gru_h)
last = gru_out[:, -1, :] # (N, gru_h)
logits = self.classifier(last) # (N, 2)
return logits
def generate_interbank_network(N: int = 30,
T: int = 40,
distress_rate: float = 0.1,
seed: int = 0
) -> dict:
"""
Generate synthetic interbank network data.
In production: replace with
- FR Y-9C / Call Report data (US banks)
- EBA stress test disclosures (EU banks)
- BIS bilateral banking statistics
"""
rng = np.random.RandomState(seed)
# Static network backbone: random sparse directed graph
adj_base = (rng.random((N, N)) < 0.2).astype(float)
np.fill_diagonal(adj_base, 0)
data = {
"node_features": [], # list of T arrays (N, 7)
"edge_attrs": [], # list of T arrays (E, 1)
"macro_features": [], # list of T arrays (4,)
"adjacency": [], # list of T arrays (N, N)
"labels": [] # list of T arrays (N,) binary
}
distressed = set()
for t in range(T):
# Node features: financial ratios (synthetic)
node_feat = rng.uniform(0.0, 1.0, (N, 7)).astype(np.float32)
# Distressed nodes have lower capital ratio
for d in distressed:
node_feat[d, 0] *= 0.3 # weaker capital ratio
# Dynamic adjacency (exposure changes over time)
noise = (rng.random((N, N)) < 0.05).astype(float)
adj_t = np.clip(adj_base + noise - noise.T, 0, 1)
np.fill_diagonal(adj_t, 0)
# Edge weights: normalized bilateral exposures
edge_wts = (adj_t * rng.uniform(0.1, 1.0, (N, N))).astype(np.float32)
# Macro features: VIX, credit spread, yield slope, policy rate
macro = rng.uniform(0.0, 1.0, 4).astype(np.float32)
if t > T // 2:
macro[0] *= 2.0 # simulate vol spike in second half
# Labels: which nodes become distressed next period
if t < T - 1:
n_distress = max(1, int(distress_rate * N))
new_distress = set(
rng.choice(N, n_distress, replace=False).tolist()
)
# Contagion: neighbors of distressed also at risk
for d in distressed:
neighbors = np.where(adj_t[d] > 0)[0]
for nb in neighbors:
if rng.random() < 0.3:
new_distress.add(int(nb))
distressed = new_distress
else:
distressed = set()
labels = np.zeros(N, dtype=np.int64)
for d in distressed:
labels[d] = 1
data["node_features"].append(node_feat)
data["adjacency"].append(adj_t)
data["edge_attrs"].append(edge_wts)
data["macro_features"].append(macro)
data["labels"].append(labels)
return datadef run_systemic_risk_pipeline():
"""End-to-end systemic risk GNN demo."""
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
N, T = 30, 40
SEQ_LEN = 8 # look-back window
net_data = generate_interbank_network(N=N, T=T, distress_rate=0.1)
def to_tensors(t):
adj = net_data["adjacency"][t]
ew = net_data["edge_attrs"][t]
nf = net_data["node_features"][t]
macro = net_data["macro_features"][t]
ei, _ = dense_to_sparse(torch.FloatTensor(adj))
src, dst = ei[0], ei[1]
ea = torch.FloatTensor(ew[src.numpy(), dst.numpy()]).unsqueeze(1)
return (
torch.FloatTensor(nf).to(device),
ei.to(device),
ea.to(device),
torch.FloatTensor(macro).to(device)
)
model = SystemicRiskGAT(
node_in=7, edge_in=1, macro_in=4,
gat_hidden=32, gru_hidden=64, n_heads=4, dropout=0.2
).to(device)
optimizer = torch.optim.Adam(model.parameters(), lr=5e-4)
split = int(0.8 * (T - SEQ_LEN))
print(f"Train snapshots: {split}, Test: {T - SEQ_LEN - split}")
print("\nTraining SystemicRiskGAT...")
for epoch in range(1, 101):
model.train()
total_loss = 0.0
for t in range(SEQ_LEN, SEQ_LEN + split):
x_seq = [to_tensors(s)[0] for s in range(t - SEQ_LEN, t)]
ei_seq = [to_tensors(s)[1] for s in range(t - SEQ_LEN, t)]
ea_seq = [to_tensors(s)[2] for s in range(t - SEQ_LEN, t)]
mac_seq = [to_tensors(s)[3] for s in range(t - SEQ_LEN, t)]
labels = torch.LongTensor(
net_data["labels"][t]
).to(device)
logits = model(x_seq, ei_seq, ea_seq, mac_seq)
loss = F.cross_entropy(logits, labels,
weight=torch.tensor([1.0, 5.0],
device=device))
optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), 1.0)
optimizer.step()
total_loss += loss.item()
if epoch % 25 == 0:
# Test evaluation
model.eval()
test_preds, test_labels_all = [], []
with torch.no_grad():
for t in range(SEQ_LEN + split, T):
x_seq = [to_tensors(s)[0] for s in range(t - SEQ_LEN, t)]
ei_seq = [to_tensors(s)[1] for s in range(t - SEQ_LEN, t)]
ea_seq = [to_tensors(s)[2] for s in range(t - SEQ_LEN, t)]
mac_seq = [to_tensors(s)[3] for s in range(t - SEQ_LEN, t)]
labels = net_data["labels"][t]
logits = model(x_seq, ei_seq, ea_seq, mac_seq)
preds = logits.argmax(dim=-1).cpu().numpy()
test_preds.extend(preds.tolist())
test_labels_all.extend(labels.tolist())
tp = sum(p == 1 and l == 1
for p, l in zip(test_preds, test_labels_all))
fp = sum(p == 1 and l == 0
for p, l in zip(test_preds, test_labels_all))
fn = sum(p == 0 and l == 1
for p, l in zip(test_preds, test_labels_all))
prec = tp / (tp + fp + 1e-8)
rec = tp / (tp + fn + 1e-8)
f1 = 2 * prec * rec / (prec + rec + 1e-8)
print(f"Epoch {epoch:3d} | Train Loss "
f"{total_loss / split:.4f} | "
f"Test Precision {prec:.3f} | "
f"Recall {rec:.3f} | F1 {f1:.3f}")
if __name__ == "__main__":
run_systemic_risk_pipeline()The Hard Truth: Where Graph Engineering Breaks Down
Graph construction is a source of alpha - and a source of look-ahead bias. Every edge in your financial graph encodes information. If that information was not available at the time of trading, you have introduced look-ahead bias at the graph structure level, which is harder to detect than look-ahead in feature computation. Correlation-based adjacency computed over the full sample is invalid. Supply chain links must use the relationships that existed at the time of the trade, not current ones. Sector assignments that changed after restructurings must use point-in-time GICS classifications.
Oversmoothing in deep GNN architectures. GNN architectures should control oversmoothing through edge dropout, attention, or regime-gated message passing when correlation graphs become dense. When every stock is connected to every other stock through a dense correlation matrix, multiple rounds of message passing wash out the individual node features and produce uniform embeddings. In practice: keep graphs sparse (threshold correlations aggressively), use attention mechanisms that can learn to zero out uninformative edges, and limit GNN depth to 2–3 layers for financial applications.
Distribution shift across market regimes. A GNN trained on a low-volatility 2019–2021 period will have learned a correlation structure that breaks down during the 2022 rate shock or a 2008-style liquidity crisis. The graph structure itself changes across regimes. Practical mitigations: train on data spanning multiple market regimes, use dynamic adjacency that is recomputed from recent data at inference time, and monitor the statistics of learned edge attention weights as a regime change indicator.
Computational cost for real-time applications. Training a GNN on a 500-stock universe with 252 days of snapshots is computationally modest. Inferencing at tick frequency across a 5000-name universe with dynamic graph recomputation is not. The practical solution adopted in production is batching: rebuild the graph daily or weekly, run GNN inference on the updated graph, and cache embeddings for intraday use. Full real-time graph recomputation is reserved for fraud detection applications where latency is less constrained than HFT.
Data requirements for supply chain graphs. Building a production-quality supply chain graph requires point-in-time supply chain data (FactSet Revere, Bloomberg SPLC, or manual extraction from SEC filings), which is expensive and requires significant data engineering. The academic literature typically uses simplified proxies. Production deployments require data infrastructure investment before the model investment.
The Implementation Roadmap for Quant Teams
Week 1-2: Foundation. Install PyTorch Geometric. Build your first correlation-based adjacency matrix from your existing equity universe. Run the StockGAT code from Section 3 on your factor feature set. Establish baseline IC and RankIC from your existing linear factor models. Measure the gap.
Week 3-4: Graph Quality. Experiment with different adjacency construction methods: correlation threshold, sector membership, supply chain edges (if you have access). The graph structure is the primary determinant of GNN quality - more expressive architecture with a poor graph will underperform a simpler architecture with a well-constructed graph. Implement strict look-ahead controls: rolling adjacency computed only from data available before the prediction date.
Week 5-6: Temporal Extension. Add the GRU layer to your GCN/GAT to capture temporal dynamics. Implement the dynamic adjacency pipeline: the graph structure updates at each rebalancing date. Measure whether dynamic adjacency improves over static. For most equity applications, it does - particularly around earnings seasons when sector correlations shift dramatically.
Week 7: Portfolio Integration. Connect GNN return predictions to your portfolio optimizer. Implement the GNN covariance estimation approach from Section 5. Run a walk-forward backtest with realistic transaction costs. Measure Sharpe ratio, information ratio, and turnover. Compare to your existing factor model.
Week 8 and beyond: Production Hardening. Implement model monitoring: track the distribution of predicted returns, attention weight statistics (a proxy for regime change), and IC decay over time. Retrain on a rolling window. Build a regime detection module that flags when the current market correlation structure diverges significantly from the training distribution - this is when the GNN is most likely to fail.
Note : i wanted to reach larger audience, QT appreciated, if done i will personally dm you to get started your journey in quants.