用PyTorch复现FactorVAE：一个能同时预测收益与风险的量化模型实战教程

用PyTorch实战FactorVAE：构建收益与风险双预测的智能量化模型

在量化投资领域，传统线性因子模型正逐渐被深度学习取代，但金融数据的高噪声特性让模型训练充满挑战。FactorVAE通过将变分自编码器(VAE)与动态因子模型结合，不仅能预测股票收益，还能从潜在空间分布中估计风险，为量化策略提供了全新工具。本文将手把手带您用PyTorch实现这一前沿模型，从数据预处理到投资组合构建，完整覆盖实战全流程。

1. 环境准备与数据工程

1.1 基础环境配置

推荐使用Python 3.8+和PyTorch 1.10+环境，关键依赖包括：

pip install torch==1.12.1+cu113 -f https://download.pytorch.org/whl/torch_stable.html pip install qlib pandas==1.4.3 scikit-learn==1.1.2

对于GPU加速，建议配置NVIDIA驱动470+和CUDA 11.3。可通过以下代码验证环境：

import torch print(f"PyTorch版本: {torch.__version__}") print(f"CUDA可用: {torch.cuda.is_available()}") print(f"GPU数量: {torch.cuda.device_count()}")

1.2 Qlib数据预处理

使用Qlib的Alpha158数据集作为特征源，该数据集包含158个技术指标和基本面因子。我们需要进行以下预处理：

1. 数据标准化：对特征进行横截面Z-score标准化
2. 缺失值处理：用行业均值填充缺失值
3. 收益计算：未来20日收益率作为预测目标

from qlib.data.dataset import DatasetH from qlib.data.dataset.handler import DataHandlerLP def create_qlib_dataset(start_time, end_time): return DatasetH( handler=DataHandlerLP( instruments='csi300', start_time=start_time, end_time=end_time, infer_processors=[ {'class': 'RobustZScoreNorm', 'kwargs': {'fields_group': 'feature'}}, {'class': 'Fillna', 'kwargs': {'fields_group': 'feature'}} ], learn_processors=[{'class': 'DropnaLabel'}], label=['Ref($close, -20)/$close - 1', 'LABEL'] ) )

提示：建议将数据缓存在SSD硬盘上，QLib原始数据约需50GB存储空间

2. 模型架构实现

2.1 特征提取器设计

采用GRU网络处理时序特征，其优势在于：

• 能捕捉金融数据的长短期依赖
• 相比LSTM参数更少，训练更快
• 对噪声数据具有鲁棒性

import torch.nn as nn class FeatureExtractor(nn.Module): def __init__(self, input_dim=158, hidden_dim=64): super().__init__() self.projection = nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.LeakyReLU(negative_slope=0.1) ) self.gru = nn.GRU( input_size=hidden_dim, hidden_size=hidden_dim, batch_first=True ) def forward(self, x): # x形状: [batch_size, seq_len, stock_num, feature_dim] batch_size, seq_len, stock_num, _ = x.shape x = x.permute(0, 2, 1, 3) # [batch, stock, seq, feature] x = x.reshape(-1, seq_len, x.size(-1)) # [batch*stock, seq, feature] h_proj = self.projection(x) # 特征投影 _, h_last = self.gru(h_proj) # 提取时序特征 h_last = h_last.view(batch_size, stock_num, -1) # 恢复形状 return h_last

2.2 因子编码器实现

因子编码器将未来收益映射到潜在空间，关键创新点包括：

• 动态投资组合构建降低维度
• Softplus保证标准差非负
• 高斯分布建模因子不确定性

class FactorEncoder(nn.Module): def __init__(self, latent_dim=32, portfolio_num=10): super().__init__() self.portfolio_layer = nn.Sequential( nn.Linear(latent_dim, portfolio_num), nn.Softmax(dim=-1) ) self.mu_net = nn.Linear(portfolio_num, latent_dim) self.sigma_net = nn.Sequential( nn.Linear(portfolio_num, latent_dim), nn.Softplus() ) def forward(self, latent_feature, future_return): # latent_feature: [batch, stock, latent] # future_return: [batch, stock] weights = self.portfolio_layer(latent_feature) # [batch, stock, port_num] port_return = torch.bmm(weights.transpose(1,2), future_return.unsqueeze(-1)).squeeze(-1) mu = self.mu_net(port_return) sigma = self.sigma_net(port_return) return mu, sigma # 均值和标准差

3. 训练策略与损失函数

3.1 先验-后验学习机制

FactorVAE的核心创新在于先验-后验学习框架：

class FactorVAE(nn.Module): def __init__(self, feature_dim=158, latent_dim=32): super().__init__() self.feature_extractor = FeatureExtractor(feature_dim, latent_dim) self.encoder = FactorEncoder(latent_dim) self.predictor = FactorPredictor(latent_dim) self.decoder = FactorDecoder(latent_dim) def forward(self, x, y_future=None, gamma=0.5): # 特征提取 h = self.feature_extractor(x) # 训练阶段使用后验因子 if y_future is not None: mu_post, sigma_post = self.encoder(h, y_future) z_post = mu_post + sigma_post * torch.randn_like(sigma_post) y_recon, mu_alpha, sigma_alpha, beta = self.decoder(z_post, h) # 计算重建损失 recon_loss = self._nll_loss(y_recon, y_future, mu_alpha, sigma_alpha, beta, mu_post, sigma_post) # 计算KL散度 mu_prior, sigma_prior = self.predictor(h) kl_loss = self._kl_div(mu_post, sigma_post, mu_prior, sigma_prior) return recon_loss + gamma * kl_loss # 预测阶段使用先验因子 mu_prior, sigma_prior = self.predictor(h) z_prior = mu_prior + sigma_prior * torch.randn_like(sigma_prior) y_pred, _, _, _ = self.decoder(z_prior, h) return y_pred, mu_prior, sigma_prior

3.2 多目标损失函数

损失函数由两部分组成：

1. 负对数似然损失：

   \mathcal{L}_{NLL} = -\sum \log p(y|\mu_y, \sigma_y)

2. KL散度损失：

   \mathcal{L}_{KL} = D_{KL}(q(z|y) \parallel p(z|x))

实现代码如下：

def _nll_loss(self, y_pred, y_true, mu_alpha, sigma_alpha, beta, mu_z, sigma_z): # 计算预测收益的分布参数 mu_y = mu_alpha + torch.bmm(beta, mu_z.unsqueeze(-1)).squeeze(-1) sigma_y = torch.sqrt(sigma_alpha**2 + torch.bmm(beta**2, sigma_z**2).squeeze(-1)) # 构建高斯分布 dist = torch.distributions.Normal(mu_y, sigma_y) return -dist.log_prob(y_true.unsqueeze(-1)).mean() def _kl_div(self, mu1, sigma1, mu2, sigma2): var1 = sigma1**2 var2 = sigma2**2 kl = torch.log(sigma2/sigma1) + (var1 + (mu1-mu2)**2)/(2*var2) - 0.5 return kl.sum(dim=-1).mean()

4. 投资组合构建与回测

4.1 风险调整策略

利用模型输出的均值和标准差，可构建风险调整后的投资组合：

def risk_adjusted_selection(returns_pred, risk_pred, top_k=50, eta=1.0): """ returns_pred: 预测收益 [stock_num] risk_pred: 预测风险 [stock_num] eta: 风险厌恶系数 """ adj_scores = returns_pred - eta * risk_pred _, selected = torch.topk(adj_scores, k=top_k) return selected

不同风险厌恶系数η的影响：

4.2 回测实现要点

使用QLib的回测模块时需注意：

1. 交易成本：A股通常按0.1%计算
2. 停牌处理：跳过无法交易的股票
3. 涨跌停限制：避免无法成交的订单

from qlib.contrib.evaluate import backtest_daily from qlib.contrib.strategy import TopkDropoutStrategy def run_backtest(predictions): strategy = TopkDropoutStrategy( topk=50, n_drop=5, risk_degree=0.95 ) report = backtest_daily( start_time="2019-01-01", end_time="2020-12-31", strategy=strategy, **predictions ) analysis = report["analysis"] print(f"年化收益: {analysis.annualized_return:.2%}") print(f"夏普比率: {analysis.sharpe_ratio:.2f}")

在实际项目中，我发现模型对参数初始化非常敏感。使用Kaiming初始化GRU参数，并将学习率设置为3e-4时，模型收敛最稳定。另一个实用技巧是在训练初期(前5个epoch)将KL损失的权重γ设为0，待重建损失下降后再逐渐增加γ值，这样能避免模型过早陷入局部最优。