Red neuronal DeePM para ponderar el riesgo de carteras entre activos
Código Machine Learning for Trading
Resumen
Este documento describe la arquitectura de una red neuronal de políticas que transforma series temporales y contexto a nivel de activo en ponderaciones acotadas del riesgo de cartera. Su estructura principal por activo combina incrustaciones de contexto, modulación por característica, selección de variables, una LSTM y atención temporal causal. Después procesa las relaciones entre activos con atención transversal rezagada y puede aplicar, de forma opcional, atención restringida por un grafo de adyacencia macroeconómica.
El contexto estático puede codificar la identidad del activo, la pertenencia a un grupo y los costes de trading, mientras que las máscaras gestionan los activos no disponibles. La capa de salida utiliza una tangente hiperbólica y aplica la máscara de activos, generando ponderaciones entre menos uno y más uno para los activos válidos. El documento explica los componentes de la arquitectura y las protecciones frente a filas de atención sin claves válidas, pero no ofrece un procedimiento de entrenamiento, resultados de trading, comparación con un índice de referencia ni pruebas de que la arquitectura mejore el rendimiento de la cartera. La descripción permite entender el diseño del modelo, pero no juzgar su eficacia empírica.
Ideas clave
- El contexto de los activos puede condicionar tanto las características de entrada como los estados ocultos recurrentes.
- La selección de variables combina características con ponderaciones que dependen del contexto.
- El modelo utiliza capas recurrentes y de atención causal para representar patrones temporales.
- La atención transversal rezagada y una máscara gráfica opcional modelan las relaciones entre activos.
- Una capa de salida acotada genera ponderaciones de riesgo con signo y respeta la disponibilidad de los activos.
Etiquetas
Texto completo
# model.py
```py
"""PyTorch implementation of the DeePM deep portfolio manager.
Architecture components:
1. Per-asset temporal backbone (shared weights): FiLM conditioning, variable
selection, LSTM, temporal self-attention.
2. Cross-sectional attention with Directed Delay for causality.
3. Macroeconomic graph prior as adjacency-masked attention.
4. Output: bounded risk weight p_{i,t} in (-1, 1) via tanh.
"""
from __future__ import annotations
import torch
from torch import nn
from .configs import ModelConfig
from .utils import causal_attention_mask
class StaticContextEncoder(nn.Module):
"""Encode per-asset static context (asset id, group id, costs)."""
def __init__(
self,
*,
n_assets: int,
n_groups: int | None,
cfg: ModelConfig,
) -> None:
super().__init__()
self.cfg = cfg
self.asset_emb = nn.Embedding(n_assets, cfg.asset_embedding_dim)
self.group_emb: nn.Embedding | None = None
if cfg.use_group_embedding:
if n_groups is None:
raise ValueError("n_groups required when use_group_embedding is True")
self.group_emb = nn.Embedding(n_groups, cfg.group_embedding_dim)
self.include_cost = cfg.use_cost_in_context
@property
def context_dim(self) -> int:
dim = self.cfg.asset_embedding_dim
if self.group_emb is not None:
dim += self.cfg.group_embedding_dim
if self.include_cost:
dim += 1
return dim
def forward(
self,
*,
asset_ids: torch.Tensor,
group_ids: torch.Tensor | None,
costs: torch.Tensor | None,
) -> torch.Tensor:
"""Return context embedding of shape (N, C)."""
emb_list = [self.asset_emb(asset_ids)]
if self.group_emb is not None:
if group_ids is None:
raise ValueError("group_ids required when group_emb is enabled")
emb_list.append(self.group_emb(group_ids))
if self.include_cost:
if costs is None:
raise ValueError("costs required when use_cost_in_context is True")
emb_list.append(costs)
return torch.cat(emb_list, dim=-1)
class FiLM(nn.Module):
"""Feature-wise linear modulation: x -> x * (1 + gamma) + beta."""
def __init__(self, *, context_dim: int, n_features: int) -> None:
super().__init__()
self.proj = nn.Linear(context_dim, 2 * n_features)
self.n_features = n_features
def forward(self, x: torch.Tensor, context: torch.Tensor) -> torch.Tensor:
gb = self.proj(context) # (N, 2F)
gamma, beta = gb[:, : self.n_features], gb[:, self.n_features :]
gamma = gamma.unsqueeze(0).unsqueeze(0) # (1, 1, N, F)
beta = beta.unsqueeze(0).unsqueeze(0)
return x * (1.0 + gamma) + beta
class VectorizedVariableSelection(nn.Module):
"""Lightweight variable selection network (V-VSN)."""
def __init__(
self,
*,
n_features: int,
d_model: int,
context_dim: int,
hidden_dim: int,
dropout: float,
) -> None:
super().__init__()
self.n_features = n_features
self.d_model = d_model
self.feature_weight = nn.Parameter(torch.empty(n_features, d_model))
self.feature_bias = nn.Parameter(torch.zeros(n_features, d_model))
nn.init.xavier_uniform_(self.feature_weight)
self.selector = nn.Sequential(
nn.Linear(n_features + context_dim, hidden_dim),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(hidden_dim, n_features),
)
self.out_norm = nn.LayerNorm(d_model)
def forward(self, x: torch.Tensor, context: torch.Tensor) -> torch.Tensor:
b, t, n, f = x.shape
context_bt = context.unsqueeze(0).unsqueeze(0).expand(b, t, n, -1)
logits = self.selector(torch.cat([x, context_bt], dim=-1))
weights = torch.softmax(logits, dim=-1)
z = torch.einsum("btnf,fd->btnfd", x, self.feature_weight) + self.feature_bias
h = (weights.unsqueeze(-1) * z).sum(dim=-2)
return self.out_norm(h)
class AdapterBlock(nn.Module):
"""FFN adapter with residual connection and LayerNorm."""
def __init__(self, *, d_model: int, hidden_mult: int, dropout: float) -> None:
super().__init__()
d_ff = int(hidden_mult * d_model)
self.ln = nn.LayerNorm(d_model)
self.ff = nn.Sequential(
nn.Linear(d_model, d_ff), nn.GELU(), nn.Dropout(dropout), nn.Linear(d_ff, d_model)
)
self.dropout = nn.Dropout(dropout)
def forward(self, x: torch.Tensor) -> torch.Tensor:
return x + self.dropout(self.ff(self.ln(x)))
class TemporalSelfAttentionBlock(nn.Module):
"""Causal temporal self-attention per asset."""
def __init__(self, *, d_model: int, n_heads: int, dropout: float, adapter_mult: int) -> None:
super().__init__()
self.mha = nn.MultiheadAttention(
embed_dim=d_model, num_heads=n_heads, dropout=dropout, batch_first=True
)
self.ln = nn.LayerNorm(d_model)
self.dropout = nn.Dropout(dropout)
self.adapter = AdapterBlock(d_model=d_model, hidden_mult=adapter_mult, dropout=dropout)
def forward(self, x: torch.Tensor) -> torch.Tensor:
t = x.shape[1]
attn_mask = causal_attention_mask(t, device=x.device)
y, _ = self.mha(x, x, x, attn_mask=attn_mask)
x = self.ln(x + self.dropout(y))
return self.adapter(x)
class CrossSectionalAttention(nn.Module):
"""Cross-asset attention with Directed Delay (time lag)."""
def __init__(self, *, d_model: int, n_heads: int, dropout: float, lag: int) -> None:
super().__init__()
self.lag = int(lag)
self.mha = nn.MultiheadAttention(
embed_dim=d_model, num_heads=n_heads, dropout=dropout, batch_first=True
)
self.ln = nn.LayerNorm(d_model)
self.dropout = nn.Dropout(dropout)
def forward(self, h: torch.Tensor, mask: torch.Tensor) -> torch.Tensor:
b, t, n, d = h.shape
if self.lag > 0:
pad = torch.zeros((b, self.lag, n, d), device=h.device, dtype=h.dtype)
h_kv = torch.cat([pad, h[:, : t - self.lag, :, :]], dim=1)
pad_m = torch.zeros((b, self.lag, n), device=mask.device, dtype=mask.dtype)
m_kv = torch.cat([pad_m, mask[:, : t - self.lag, :]], dim=1)
else:
h_kv = h
m_kv = mask
q = h.reshape(b * t, n, d)
kv = h_kv.reshape(b * t, n, d)
key_padding_mask = m_kv.reshape(b * t, n) < 0.5
# When lag > 0, the first `lag` timesteps have all-zero keys and masks.
# All keys masked → softmax(all -inf) → NaN. Unmask all positions for
# those rows; attention over zero-valued keys yields zero, so the
# residual connection passes through h unchanged.
all_masked = key_padding_mask.all(dim=-1, keepdim=True)
if all_masked.any():
key_padding_mask = key_padding_mask & ~all_masked
out, _ = self.mha(q, kv, kv, key_padding_mask=key_padding_mask)
out = out.reshape(b, t, n, d)
return self.ln(h + self.dropout(out))
class MacroGraphAttention(nn.Module):
"""Adjacency-masked cross-asset attention (GAT-like)."""
def __init__(
self,
*,
d_model: int,
n_heads: int,
dropout: float,
adjacency_mask: torch.Tensor,
) -> None:
super().__init__()
self.register_buffer("adjacency_mask", adjacency_mask.to(dtype=torch.bool))
self.mha = nn.MultiheadAttention(
embed_dim=d_model, num_heads=n_heads, dropout=dropout, batch_first=True
)
self.ln = nn.LayerNorm(d_model)
self.dropout = nn.Dropout(dropout)
def forward(self, h: torch.Tensor, mask: torch.Tensor) -> torch.Tensor:
b, t, n, d = h.shape
x = h.reshape(b * t, n, d)
key_padding_mask = mask.reshape(b * t, n) < 0.5
# Guard: if adjacency_mask + key_padding_mask blocks ALL keys for any
# query, softmax produces NaN. Unmask everything for those rows.
combined = self.adjacency_mask.unsqueeze(0) | key_padding_mask.unsqueeze(1)
all_blocked = combined.all(dim=-1) # (B*T, N) — True if query i has no valid key
if all_blocked.any():
key_padding_mask = key_padding_mask & ~all_blocked
out, _ = self.mha(x, x, x, attn_mask=self.adjacency_mask, key_padding_mask=key_padding_mask)
out = out.reshape(b, t, n, d)
return self.ln(h + self.dropout(out))
class DeepmPolicy(nn.Module):
"""DeePM policy network that outputs risk weights p_{i,t} in (-1, 1)."""
def __init__(
self,
*,
n_assets: int,
n_features: int,
n_groups: int | None,
adjacency_mask: torch.Tensor | None,
cfg: ModelConfig,
) -> None:
super().__init__()
self.n_assets = int(n_assets)
self.n_features = int(n_features)
self.cfg = cfg
self.context_encoder = StaticContextEncoder(n_assets=n_assets, n_groups=n_groups, cfg=cfg)
context_dim = self.context_encoder.context_dim
self.film = FiLM(context_dim=context_dim, n_features=n_features)
self.vvsn = VectorizedVariableSelection(
n_features=n_features,
d_model=cfg.d_model,
context_dim=context_dim,
hidden_dim=cfg.vvsn_hidden_dim,
dropout=cfg.dropout,
)
self.lstm = nn.LSTM(
input_size=cfg.d_model,
hidden_size=cfg.d_model,
num_layers=cfg.lstm_layers,
batch_first=True,
dropout=cfg.dropout if cfg.lstm_layers > 1 else 0.0,
)
self.h0_proj = nn.Linear(context_dim, cfg.lstm_layers * cfg.d_model)
self.c0_proj = nn.Linear(context_dim, cfg.lstm_layers * cfg.d_model)
self.temporal_blocks = nn.ModuleList(
[
TemporalSelfAttentionBlock(
d_model=cfg.d_model,
n_heads=cfg.n_heads,
dropout=cfg.dropout,
adapter_mult=cfg.adapter_hidden_mult,
)
for _ in range(cfg.temporal_mha_layers)
]
)
self.cross_attn = CrossSectionalAttention(
d_model=cfg.d_model,
n_heads=cfg.cross_attention_heads,
dropout=cfg.dropout,
lag=cfg.cross_attention_lag,
)
self.macro_graph: MacroGraphAttention | None = None
if adjacency_mask is not None:
self.macro_graph = MacroGraphAttention(
d_model=cfg.d_model,
n_heads=cfg.macro_gnn_heads,
dropout=cfg.dropout,
adjacency_mask=adjacency_mask,
)
self.head = nn.Linear(cfg.d_model, 1)
def forward(
self,
x: torch.Tensor,
*,
mask: torch.Tensor,
asset_ids: torch.Tensor,
group_ids: torch.Tensor | None,
costs: torch.Tensor | None,
) -> torch.Tensor:
"""Forward pass: features (B,T,N,F) -> risk weights (B,T,N) in (-1,1)."""
b, t, n, f = x.shape
context = self.context_encoder(asset_ids=asset_ids, group_ids=group_ids, costs=costs)
x_mod = self.film(x, context)
h = self.vvsn(x_mod, context) # (B,T,N,D)
# Per-asset temporal backbone
h_bn = h.permute(0, 2, 1, 3).reshape(b * n, t, self.cfg.d_model)
ctx_bn = context.unsqueeze(0).expand(b, n, -1).reshape(b * n, -1)
h0 = (
self.h0_proj(ctx_bn)
.reshape(b * n, self.cfg.lstm_layers, self.cfg.d_model)
.permute(1, 0, 2)
.contiguous()
)
c0 = (
self.c0_proj(ctx_bn)
.reshape(b * n, self.cfg.lstm_layers, self.cfg.d_model)
.permute(1, 0, 2)
.contiguous()
)
h_bn, _ = self.lstm(h_bn, (h0, c0))
for block in self.temporal_blocks:
h_bn = block(h_bn)
h = h_bn.reshape(b, n, t, self.cfg.d_model).permute(0, 2, 1, 3).contiguous()
# Cross-sectional blocks
h = self.cross_attn(h, mask)
if self.macro_graph is not None:
h = self.macro_graph(h, mask)
# Output head -> tanh
p = torch.tanh(self.head(h).squeeze(-1))
return p * mask
```Se muestra íntegramente con atribución según la licencia de la fuente. Licencia: MIT
Este resumen lo redactó el agente de investigación de Stratmill a partir del original; no es una copia de la fuente.