用于资产定价的对抗式随机贴现因子网络
代码 《交易机器学习》
总结
本实现介绍了一种神经随机贴现因子模型和一个用于检验资产定价关系的对抗矩网络。SDF网络将资产特征（可选择与来自LSTM的宏观经济状态表征相结合）映射为资产权重。这些权重与收益共同构成时点层面的贴现因子。启用宏观输入时，LSTM状态会在每个时间步由所有资产共享；否则，模型仅使用资产特征。
对抗网络学习用于揭示条件定价矩失效的工具变量。代码定义了无条件和条件平方矩损失，用掩码处理缺失的资产观测，并提供了符合隐含SDF投资组合收益约定的夏普比率计算方法。该摘录说明的是模型结构和损失构造，而非拟合后的研究、数据或实证结果。因此，它不能证明学习得到的SDF能够成功为资产定价，也没有说明模型对抗式框架所暗示的完整训练流程。
核心观点
- SDF网络将资产特征和可选的宏观状态转换为资产级权重。
- 贴现因子通过每个时间步将加权资产收益加到1上构造。
- 可选的LSTM编码宏观经济输入，并为各资产提供共同状态。
- 矩网络学习用于寻找条件定价矩违背情况的工具变量。
- 列出的损失和夏普比率约定描述的是训练目标与测量方法，并非实证验证。
标签
全文
# sdf.py
```py
"""Stochastic Discount Factor Network (Chen, Pelger, Zhu 2024).
Adversarial architecture:
- 3-phase training: unconditional warmup -> adversarial rounds
When n_macro_features=0, the LSTM branch is omitted (case study usage).
When n_macro_features>0, full LSTM processes macro indicators (teaching notebook).
Reference: Chen, Pelger, Zhu (2024) "Deep Learning in Asset Pricing"
"""
from __future__ import annotations
import torch
import torch.nn as nn
class SDFNetwork(nn.Module):
"""SDF Network: learns portfolio weights from asset characteristics + optional macro state.
Architecture:
- Optional MacroLSTM processes macro features to extract economic regime state
- FFN combines asset features (+ macro state) to produce per-stock weights
- SDF = 1 + sum(weights * returns)
Args:
n_asset_features: Number of asset characteristics
n_macro_features: Number of macro features (0 to disable LSTM)
state_dim: LSTM hidden state dimension
hidden_dim: FFN hidden layer size
dropout: Dropout rate
"""
def __init__(
self,
n_asset_features: int,
n_macro_features: int = 0,
state_dim: int = 4,
hidden_dim: int = 64,
dropout: float = 0.05,
):
super().__init__()
self.state_dim = state_dim
self.use_macro = n_macro_features > 0
if self.use_macro:
# Paper-faithful CPZ uses dropout only inside the SDF FFN — the
# macro LSTM input is fed raw. The previous implementation added
# `nn.Dropout` on the macro path which is not in the published
# spec; removed to match the reference.
self.lstm = nn.LSTM(
input_size=n_macro_features,
hidden_size=state_dim,
batch_first=True,
)
ffn_input_dim = n_asset_features + (state_dim if self.use_macro else 0)
self.ffn = nn.Sequential(
nn.Linear(ffn_input_dim, hidden_dim),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(hidden_dim, hidden_dim),
nn.ReLU(),
nn.Dropout(dropout),
nn.Linear(hidden_dim, 1),
)
def forward(
self,
asset_features: torch.Tensor, # (T, N, F_asset)
macro_features: torch.Tensor | None = None, # (T, F_macro)
mask: torch.Tensor | None = None, # (T, N)
h0: torch.Tensor | None = None,
c0: torch.Tensor | None = None,
) -> tuple[torch.Tensor, tuple[torch.Tensor | None, torch.Tensor | None]]:
"""Forward pass producing SDF weights.
Returns:
weights: (n_valid,) weight per valid observation
(h_n, c_n): LSTM states (None if no macro)
"""
if self.use_macro and macro_features is None:
raise ValueError(
"macro_features must be provided when n_macro_features > 0; "
"the SDF LSTM branch has no fallback."
)
if (h0 is None) != (c0 is None):
raise ValueError("h0 and c0 must be provided together (or both None).")
T, N, F = asset_features.shape
if mask is None:
mask = torch.ones(T, N, dtype=torch.bool, device=asset_features.device)
if self.use_macro and macro_features is not None:
if h0 is None:
h0 = torch.zeros(1, 1, self.state_dim, device=asset_features.device)
c0 = torch.zeros(1, 1, self.state_dim, device=asset_features.device)
macro_seq = macro_features.unsqueeze(0) # (1, T, F_macro)
macro_states, (h_n, c_n) = self.lstm(macro_seq, (h0, c0))
macro_states = macro_states.squeeze(0) # (T, state_dim)
macro_tiled = macro_states.unsqueeze(1).expand(-1, N, -1)
asset_flat = asset_features[mask]
macro_flat = macro_tiled[mask]
ffn_input = torch.cat([asset_flat, macro_flat], dim=1)
else:
asset_flat = asset_features[mask]
ffn_input = asset_flat
h_n = c_n = None
weights = self.ffn(ffn_input).squeeze(-1)
return weights, (h_n, c_n)
class MomentNetwork(nn.Module):
"""Moment Network (adversary): learns instruments for adversarial moment conditions.
Finds test asset portfolios where E[M * R * Z] != 0, exposing SDF pricing failures.
Args:
n_asset_features: Number of asset characteristics
n_macro_features: Number of macro features (0 to disable LSTM)
n_instruments: Number of learned instruments
state_dim: LSTM hidden state dimension
dropout: Dropout rate
"""
def __init__(
self,
n_asset_features: int,
n_macro_features: int = 0,
n_instruments: int = 8,
state_dim: int = 32,
dropout: float = 0.05,
):
super().__init__()
self.state_dim = state_dim
self.n_instruments = n_instruments
self.use_macro = n_macro_features > 0
if self.use_macro:
# Paper-faithful CPZ moment net has no FFN hidden layers and no
# macro-input dropout; LSTM input is fed raw.
self.lstm = nn.LSTM(
input_size=n_macro_features,
hidden_size=state_dim,
batch_first=True,
)
ffn_input_dim = n_asset_features + (state_dim if self.use_macro else 0)
self.ffn = nn.Sequential(
nn.Linear(ffn_input_dim, n_instruments),
nn.Tanh(),
)
def forward(
self,
asset_features: torch.Tensor, # (T, N, F_asset)
macro_features: torch.Tensor | None = None, # (T, F_macro)
h0: torch.Tensor | None = None,
c0: torch.Tensor | None = None,
) -> tuple[torch.Tensor, tuple[torch.Tensor | None, torch.Tensor | None]]:
"""Forward pass producing instruments.
Returns:
instruments: (n_instruments, T, N)
(h_n, c_n): LSTM states (None if no macro)
"""
if self.use_macro and macro_features is None:
raise ValueError(
"macro_features must be provided when n_macro_features > 0; "
"the moment-network LSTM branch has no fallback."
)
if (h0 is None) != (c0 is None):
raise ValueError("h0 and c0 must be provided together (or both None).")
T, N, F = asset_features.shape
if self.use_macro and macro_features is not None:
if h0 is None:
h0 = torch.zeros(1, 1, self.state_dim, device=asset_features.device)
c0 = torch.zeros(1, 1, self.state_dim, device=asset_features.device)
macro_seq = macro_features.unsqueeze(0)
macro_states, (h_n, c_n) = self.lstm(macro_seq, (h0, c0))
macro_states = macro_states.squeeze(0)
macro_tiled = macro_states.unsqueeze(1).expand(-1, N, -1)
ffn_input = torch.cat([asset_features, macro_tiled], dim=2)
else:
ffn_input = asset_features
h_n = c_n = None
instruments = self.ffn(ffn_input) # (T, N, n_instruments)
instruments = instruments.permute(2, 0, 1) # (n_instruments, T, N)
return instruments, (h_n, c_n)
# ---------------------------------------------------------------------------
# SDF construction and loss functions
# ---------------------------------------------------------------------------
def get_segment_ids(mask: torch.Tensor) -> torch.Tensor:
"""Create segment IDs mapping valid observations to time steps.
Args:
mask: (T, N) boolean mask
Returns:
segment_ids: (n_valid,) time step index per valid observation
"""
T, N = mask.shape
time_ids = torch.arange(T, device=mask.device).unsqueeze(1).expand(-1, N)
return time_ids[mask]
def construct_sdf(weights: torch.Tensor, returns: torch.Tensor, mask: torch.Tensor) -> torch.Tensor:
"""Construct SDF from weights and returns.
SDF_t = 1 + sum_i(w_i * r_i) for each time step.
Args:
weights: (n_valid,) SDF weights
returns: (T, N) asset returns
mask: (T, N) valid observations
Returns:
sdf: (T,) SDF value per time step
"""
T, N = returns.shape
returns_flat = returns[mask]
segment_ids = get_segment_ids(mask)
weighted_returns = weights * returns_flat
sdf_values = torch.zeros(T, device=weights.device)
sdf_values.scatter_add_(0, segment_ids, weighted_returns)
return 1 + sdf_values
def unconditional_loss(
weights: torch.Tensor,
returns: torch.Tensor,
mask: torch.Tensor,
n_obs_per_asset: torch.Tensor,
) -> tuple[torch.Tensor, torch.Tensor]:
"""Unconditional pricing loss: E[M * R * 1]^2 with constant instrument Z=1.
Returns:
loss: scalar MSE
sdf: (T,)
"""
T, N = returns.shape
mask_float = mask.float()
sdf = construct_sdf(weights, returns, mask)
sdf_expanded = sdf.unsqueeze(1)
instruments = torch.ones(1, T, N, device=weights.device)
sample_moments = returns * mask_float * sdf_expanded * instruments
weighted_moments = sample_moments.sum(dim=1) / n_obs_per_asset.clamp(min=1)
n_obs_norm = n_obs_per_asset / n_obs_per_asset.max()
loss = (weighted_moments.pow(2) * n_obs_norm).mean()
return loss, sdf
def conditional_loss(
weights: torch.Tensor,
instruments: torch.Tensor,
returns: torch.Tensor,
mask: torch.Tensor,
n_obs_per_asset: torch.Tensor,
) -> tuple[torch.Tensor, torch.Tensor]:
"""Conditional pricing loss: E[M * R * Z]^2 with learned instruments.
Args:
weights: (n_valid,)
instruments: (n_instruments, T, N)
returns: (T, N)
mask: (T, N)
n_obs_per_asset: (N,)
Returns:
loss: scalar MSE
sdf: (T,)
"""
T, N = returns.shape
n_instruments = instruments.shape[0]
mask_float = mask.float()
sdf = construct_sdf(weights, returns, mask)
sdf_expanded = sdf.unsqueeze(1)
sample_moments = returns * mask_float * sdf_expanded * instruments
weighted_moments = sample_moments.sum(dim=1) / n_obs_per_asset.clamp(min=1)
n_obs_norm = n_obs_per_asset / n_obs_per_asset.max()
n_obs_tiled = n_obs_norm.unsqueeze(0).expand(n_instruments, -1)
loss = (weighted_moments.pow(2) * n_obs_tiled).mean()
return loss, sdf
def compute_sharpe(sdf: torch.Tensor) -> torch.Tensor:
"""Sharpe ratio of SDF portfolio (1 - M).
Uses population standard deviation (``unbiased=False``) for a fixed-
convention validation metric across folds. Empty / constant SDF series
return 0 rather than NaN so the trainer's checkpoint comparison stays
deterministic.
"""
portfolio_return = 1 - sdf
if portfolio_return.numel() == 0:
return torch.zeros((), device=sdf.device, dtype=sdf.dtype)
mean = portfolio_return.mean()
std = portfolio_return.std(unbiased=False).clamp(min=1e-8)
out = mean / std
return torch.where(torch.isfinite(out), out, torch.zeros_like(out))
```在遵守原作品许可的前提下，附作者信息全文展示。 许可协议: MIT
此摘要由 Stratmill 研究智能体根据原文撰写，并非原文副本。