Нейросеть DeePM для риск-весов кросс-активного портфеля
Код Machine Learning for Trading
Сводка
В документе описана архитектура нейросетевой политики, которая преобразует временные ряды и контекст по отдельным активам в ограниченные риск-веса портфеля. Основа для каждого актива объединяет встраивания контекста, модуляцию по признакам, отбор переменных, LSTM и причинное временное внимание. Затем учитываются взаимосвязи между активами с помощью лагированного поперечного внимания; дополнительно можно ограничить внимание графом смежности макроэкономических связей.
Статический контекст может кодировать идентификатор актива, принадлежность к группе и торговые издержки; маски учитывают недоступные активы. Выходной слой использует гиперболический тангенс и применяет маску активов, формируя для допустимых активов веса от минус одного до плюс одного. В документе объясняются компоненты архитектуры и меры защиты от строк внимания без допустимых ключей, но не приводятся процедура обучения, торговые результаты, сравнение с бенчмарками или свидетельства улучшения доходности портфеля благодаря этой архитектуре. Описание позволяет понять устройство модели, но не оценить её эмпирическую эффективность.
Ключевые идеи
- Контекст актива может влиять и на входные признаки, и на рекуррентные скрытые состояния.
- Отбор переменных объединяет признаки с помощью весов, зависящих от контекста.
- Модель использует рекуррентные слои и причинное внимание для представления временных закономерностей.
- Лагированное поперечное внимание и необязательная маска графа моделируют взаимосвязи между активами.
- Ограниченный выходной слой формирует знаковые риск-веса с учётом доступности активов.
Теги
Полный текст
# 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
```Полный текст с указанием источника опубликован на условиях его лицензии. Лицензия: MIT
Это краткое изложение подготовлено исследовательским агентом Stratmill по оригиналу и не является его копией.