用PyTorch复现FactorVAE：一个能同时预测收益和风险的量化模型（附完整代码）

csdn产品小助手

853人浏览 · 2026-05-20 05:08:14

csdn产品小助手 · 2026-05-20 05:08:14 发布

从零实现FactorVAE：构建兼具收益预测与风险管理的量化模型

在量化投资领域，传统线性因子模型正逐渐被非线性机器学习方法所取代。FactorVAE作为融合变分自编码器与动态因子模型的创新架构，不仅能预测股票收益，还能通过潜在空间分布估计投资风险。本文将完整呈现PyTorch实现过程，包含特征提取、因子编码、先验-后验学习等核心模块的代码级解析。

1. 环境配置与数据准备

1.1 基础环境搭建

推荐使用Python 3.8+和PyTorch 1.12+环境，主要依赖库包括：

# requirements.txt
torch==1.12.1+cu113
numpy==1.23.5
pandas==1.5.3
scikit-learn==1.2.2
tqdm==4.65.0

关键配置参数示例：

config = {
    "time_span": 60,          # 历史数据时间窗口
    "characteristic_size": 158, # Alpha158特征维度
    "latent_size": 64,        # 潜在特征维度
    "factor_size": 10,        # 因子数量
    "batch_size": 256,
    "learning_rate": 3e-4,
    "gamma": 0.5             # KL散度权重
}

1.2 数据预处理流程

典型金融数据处理流程包含以下关键步骤：

特征标准化：对Alpha158特征进行横截面Z-score标准化
收益率计算：未来20日累计收益作为预测目标
缺失值处理：向前填充结合行业均值填充
数据集划分：按时间序列划分训练/验证/测试集

class StockDataset(Dataset):
    def __init__(self, characteristics, returns, time_span=60):
        self.x = torch.FloatTensor(characteristics)  # (N, T, C)
        self.y = torch.FloatTensor(returns)          # (N,)
        
    def __getitem__(self, idx):
        return self.x[idx], self.y[idx]

2. 模型核心架构实现

2.1 特征提取器设计

采用GRU网络捕捉时序特征，配合MLP进行维度变换：

class FeatureExtractor(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers=2):
        super().__init__()
        self.gru = nn.GRU(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True
        )
        self.mlp = nn.Sequential(
            nn.Linear(hidden_size, hidden_size),
            nn.LeakyReLU(),
            nn.LayerNorm(hidden_size)
        )
        
    def forward(self, x):
        # x: (batch, time_span, characteristic_size)
        _, h_n = self.gru(x)  # h_n: (num_layers, batch, hidden_size)
        return self.mlp(h_n[-1])  # (batch, hidden_size)

2.2 因子编码器实现

创新性采用投资组合构建方法降维：

class FactorEncoder(nn.Module):
    def __init__(self, input_size, factor_size):
        super().__init__()
        self.portfolio = nn.Linear(input_size, factor_size)
        self.mu_net = nn.Sequential(
            nn.Linear(factor_size, factor_size),
            nn.LeakyReLU()
        )
        self.sigma_net = nn.Sequential(
            nn.Linear(factor_size, factor_size),
            nn.Softplus()
        )
        
    def forward(self, h, returns):
        weights = F.softmax(self.portfolio(h), dim=1)  # (batch, factor)
        portfolio_ret = torch.matmul(weights.T, returns.unsqueeze(-1))
        return self.mu_net(portfolio_ret), self.sigma_net(portfolio_ret)

3. 先验-后验学习机制

3.1 多头注意力因子预测器

class FactorPredictor(nn.Module):
    def __init__(self, hidden_size, factor_size, num_heads=4):
        super().__init__()
        self.attention = nn.MultiheadAttention(
            embed_dim=hidden_size,
            num_heads=num_heads,
            batch_first=True
        )
        self.mu_layer = nn.Linear(hidden_size, factor_size)
        self.sigma_layer = nn.Sequential(
            nn.Linear(hidden_size, factor_size),
            nn.Softplus()
        )
        
    def forward(self, h):
        attn_out, _ = self.attention(h, h, h)
        global_feat = torch.mean(attn_out, dim=1)
        return self.mu_layer(global_feat), self.sigma_layer(global_feat)

3.2 目标函数实现

结合负对数似然与KL散度的复合损失：

def compute_loss(pred_ret, true_ret, mu_post, sigma_post, mu_prior, sigma_prior, gamma):
    # 重构损失
    recon_loss = F.mse_loss(pred_ret, true_ret)
    
    # KL散度计算
    post_dist = Normal(mu_post, sigma_post)
    prior_dist = Normal(mu_prior, sigma_prior)
    kl_loss = torch.distributions.kl_divergence(post_dist, prior_dist).mean()
    
    return recon_loss + gamma * kl_loss

4. 训练优化技巧

4.1 学习率调度策略

采用余弦退火配合热启动：

scheduler = torch.optim.lr_scheduler.CosineAnnealingWarmRestarts(
    optimizer,
    T_0=50,
    T_mult=2,
    eta_min=1e-5
)

4.2 梯度裁剪与早停

torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)

# 早停监控
best_loss = float('inf')
patience = 10

4.3 批归一化技巧

在特征提取器后添加行业感知归一化：

class IndustryNorm(nn.Module):
    def __init__(self, num_industries, hidden_size):
        super().__init__()
        self.gamma = nn.Parameter(torch.ones(num_industries, hidden_size))
        self.beta = nn.Parameter(torch.zeros(num_industries, hidden_size))
        
    def forward(self, x, industry_idx):
        return x * self.gamma[industry_idx] + self.beta[industry_idx]

5. 回测与结果分析

5.1 预测性能评估指标

指标	公式	说明
Rank IC	corr(rank(y_pred), rank(y_true))	预测排序能力
ICIR	mean(IC) / std(IC)	IC稳定性
Annual Return	∏(1+ret)^252 - 1	年化收益
Sharpe Ratio	mean(ret) / std(ret) * √252	风险调整后收益

5.2 组合构建策略

TopK-Drop策略实现要点：

def topk_drop(returns_pred, risk_pred, k=50, n=5, eta=0.5):
    # 风险调整后收益
    adj_returns = returns_pred - eta * risk_pred
    
    # 获取TopK股票
    topk = torch.topk(adj_returns, k=k).indices
    
    # 换仓约束处理
    if hasattr(self, 'prev_holdings'):
        overlap = len(set(topk) & set(self.prev_holdings))
        if overlap < k - n:
            keep = list(set(topk) & set(self.prev_holdings))
            add = torch.topk(adj_returns, k=k-len(keep)).indices
            topk = torch.cat([keep, add])
    
    self.prev_holdings = topk
    return topk

5.3 实际部署注意事项

数据延迟处理：实际交易需考虑T+1结算制度
交易成本模型：包含佣金和滑点模拟
风险控制模块：单行业/单股票暴露限制
实时监控系统：异常预测值检测机制

class RiskController:
    def __init__(self, max_sector_exposure=0.3):
        self.sector_limits = max_sector_exposure
        
    def check(self, weights, sector_info):
        sector_exposure = torch.zeros(len(sector_info))
        for i, sector in enumerate(sector_info):
            sector_exposure[i] = weights[sector].sum()
        
        if (sector_exposure > self.sector_limits).any():
            return False
        return True

6. 模型优化方向

6.1 特征工程改进

引入高频量价特征
结合另类数据（新闻情绪、供应链数据）
动态特征选择机制

6.2 架构增强方案

class EnhancedFactorVAE(nn.Module):
    def __init__(self, base_model):
        super().__init__()
        self.base = base_model
        self.temporal_attn = TemporalAttention(hidden_size)
        self.cross_asset_gnn = GATLayer(hidden_size)