用PyTorch复现FactorVAE：一个能预测股票收益的变分自编码器实战教程

用PyTorch实战FactorVAE：构建可解释的量化投资因子模型

在量化投资领域，因子模型一直是资产定价和收益预测的核心工具。传统线性因子模型如Fama-French三因子模型虽然解释性强，但难以捕捉市场中的非线性关系。本文将手把手教你用PyTorch实现FactorVAE——一种融合变分自编码器(VAE)与动态因子模型(DFM)的创新方法，不仅能预测股票收益，还能通过潜在空间分布估计风险。

1. 环境准备与数据预处理

1.1 安装依赖库

首先确保已安装最新版PyTorch。推荐使用conda创建独立环境：

conda create -n factorvae python=3.8 conda activate factorvae pip install torch==1.12.0+cu113 -f https://download.pytorch.org/whl/torch_stable.html pip install qlib pyportfolioopt

1.2 加载Alpha158特征集

我们使用Qlib提供的Alpha158特征集，包含158个经过验证的量化特征：

from qlib.data.dataset import DatasetH from qlib.data.dataset.handler import Alpha158 handler_config = { "start_time": "2010-01-01", "end_time": "2020-12-31", "fit_start_time": "2010-01-01", "fit_end_time": "2017-12-31", "instruments": "csi300" } # 初始化特征处理器 handler = Alpha158(**handler_config) dataset = DatasetH(handler, segments={ "train": ("2010-01-01", "2017-12-31"), "valid": ("2018-01-01", "2018-12-31"), "test": ("2019-01-01", "2020-12-31") })

注意：Alpha158特征已进行标准化处理，包含动量、波动率、流动性等多维指标

2. 模型架构设计

2.1 特征提取器（GRU网络）

使用GRU处理时间序列特征，提取股票潜在表征：

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() ) self.gru = nn.GRU( input_size=hidden_dim, hidden_size=hidden_dim, batch_first=True ) def forward(self, x): # x形状: [batch_size, seq_len, n_stocks, n_features] batch_size, seq_len, n_stocks, _ = x.shape x = x.permute(0, 2, 1, 3) # [batch, stocks, seq, features] x = x.reshape(-1, seq_len, x.size(-1)) projected = self.projection(x) # [batch*stocks, seq, hidden] _, hidden = self.gru(projected) return hidden.squeeze(0).view(batch_size, n_stocks, -1)

2.2 因子编码器与解码器

2.2.1 因子编码器

class FactorEncoder(nn.Module): def __init__(self, latent_dim=32, n_portfolios=10): super().__init__() self.portfolio_weights = nn.Linear(latent_dim, n_portfolios) self.mu_net = nn.Sequential( nn.Linear(n_portfolios, latent_dim), nn.LeakyReLU() ) self.sigma_net = nn.Sequential( nn.Linear(n_portfolios, latent_dim), nn.Softplus() ) def forward(self, latent_features, returns): weights = torch.softmax(self.portfolio_weights(latent_features), dim=-1) portfolio_returns = torch.einsum('bsf,bs->bf', weights, returns) return self.mu_net(portfolio_returns), self.sigma_net(portfolio_returns)

2.2.2 因子解码器

class FactorDecoder(nn.Module): def __init__(self, latent_dim=32, factor_dim=8): super().__init__() self.alpha_net = nn.Sequential( nn.Linear(latent_dim, latent_dim), nn.LeakyReLU(), nn.Linear(latent_dim, 1) ) self.beta_net = nn.Sequential( nn.Linear(latent_dim, factor_dim), nn.LeakyReLU() ) def forward(self, factors, latent_features): alpha = self.alpha_net(latent_features) beta = self.beta_net(latent_features) return alpha + torch.einsum('bsf,bf->bs', beta, factors)

2.3 多头注意力预测器

class FactorPredictor(nn.Module): def __init__(self, latent_dim=32, factor_dim=8, n_heads=4): super().__init__() self.attention = nn.MultiheadAttention( embed_dim=latent_dim, num_heads=n_heads, batch_first=True ) self.mu_net = nn.Linear(latent_dim, factor_dim) self.sigma_net = nn.Sequential( nn.Linear(latent_dim, factor_dim), nn.Softplus() ) def forward(self, x): attn_out, _ = self.attention(x, x, x) global_rep = attn_out.mean(dim=1) return self.mu_net(global_rep), self.sigma_net(global_rep)

3. 训练策略与目标函数

3.1 先验-后验学习机制

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

1. 后验阶段：使用未来收益信息学习最优因子分布
2. 先验阶段：仅基于历史数据预测因子分布

def compute_loss(model, batch, gamma=0.5): features, future_returns = batch # 后验因子 latent = model.feature_extractor(features) mu_post, sigma_post = model.factor_encoder(latent, future_returns) z_post = mu_post + sigma_post * torch.randn_like(sigma_post) # 重建收益 reconstructed = model.factor_decoder(z_post, latent) nll_loss = -Normal(reconstructed, 0.1).log_prob(future_returns).mean() # 先验因子 mu_prior, sigma_prior = model.factor_predictor(latent) kl_loss = kl_divergence( Normal(mu_post, sigma_post), Normal(mu_prior, sigma_prior) ).mean() return nll_loss + gamma * kl_loss

3.2 训练循环实现

from torch.optim import AdamW from torch.utils.data import DataLoader def train_model(model, dataset, epochs=100, lr=1e-3): optimizer = AdamW(model.parameters(), lr=lr) train_loader = DataLoader(dataset, batch_size=32, shuffle=True) for epoch in range(epochs): total_loss = 0 for batch in train_loader: optimizer.zero_grad() loss = compute_loss(model, batch) loss.backward() optimizer.step() total_loss += loss.item() print(f"Epoch {epoch+1}/{epochs} | Loss: {total_loss/len(train_loader):.4f}")

4. 回测验证与结果分析

4.1 预测收益生成

def predict(model, features): with torch.no_grad(): latent = model.feature_extractor(features) mu_prior, sigma_prior = model.factor_predictor(latent) z_prior = mu_prior + sigma_prior * torch.randn_like(sigma_prior) return model.factor_decoder(z_prior, latent)

4.2 TopK-Drop策略实现

import pandas as pd from qlib.contrib.evaluate import backtest_daily def evaluate_strategy(predictions, k=50, n=5): # 生成每日持仓信号 signals = predictions.groupby(level='datetime').apply( lambda x: x.nlargest(k).index.get_level_values('instrument') ) # 执行回测 report = backtest_daily( signals, strategy="topk", topk=k, turnover_n=n, benchmark="SH000300" ) return pd.DataFrame(report).T

4.3 风险调整策略

def risk_adjusted_strategy(predictions, risk_aversion=1.0): mu = predictions.groupby(level='datetime').mean() sigma = predictions.groupby(level='datetime').std() adjusted = mu - risk_aversion * sigma return adjusted.sort_values(ascending=False).head(50)

5. 模型优化技巧

5.1 特征重要性分析

使用Integrated Gradients方法识别关键特征：

from captum.attr import IntegratedGradients def feature_importance(model, sample): ig = IntegratedGradients(model.feature_extractor) attributions = ig.attribute( sample, target=0, n_steps=50 ) return attributions.mean(dim=(0,1,2)).abs().sort(descending=True)

5.2 超参数调优

建议使用Optuna进行自动化调参：

import optuna def objective(trial): params = { 'latent_dim': trial.suggest_categorical('latent_dim', [32, 64, 128]), 'factor_dim': trial.suggest_int('factor_dim', 4, 16), 'gamma': trial.suggest_float('gamma', 0.1, 1.0), 'lr': trial.suggest_float('lr', 1e-4, 1e-3, log=True) } model = FactorVAE(**params) return train_and_validate(model)

5.3 模型可解释性增强

通过因子旋转提升经济意义解释：

from sklearn.decomposition import FactorAnalysis def rotate_factors(model, data_loader): factors = [] for batch in data_loader: latent = model.feature_extractor(batch[0]) mu, _ = model.factor_predictor(latent) factors.append(mu) all_factors = torch.cat(factors).numpy() rotated = FactorAnalysis(n_components=model.factor_dim).fit_transform(all_factors) return rotated

在实际项目中，我们发现将GRU隐藏层维度设置为64、因子维度为8时，模型在保持预测精度的同时具有最佳的可解释性。风险厌恶系数η设为0.5时，能平衡收益与波动率。