深圳融克迪特科技有限公司

1. 从单点张量操作到端到端特征流

上一讲《PyTorch量化建模入门：张量、自动求导与训练循环》已确立三大基石：其一，张量（Tensor）作为统一数据载体，支持CPU/GPU设备协同、dtype精度可控（如torch.float32与torch.bfloat16在因子矩阵计算中的误差累积差异）、内存连续性保障（contiguous()调用时机直接影响后续reshape效率）；其二，Autograd图实现了动态梯度传播，明确requires_grad=True仅需设于可学习参数（如线性层权重）或需反向传播的中间变量（如归一化后的因子值），而原始行情输入（OHLCV张量）通常设为False以节省显存；其三，训练循环完成原子化拆解——前向传播、损失计算、反向传播、参数更新四步闭环，并通过torch.no_grad()上下文管理器保障评估阶段无梯度缓存。本讲在此基础上，将输入数据从静态张量升级为可复用、可复现、可审计的特征管线（Feature Pipeline），解决量化建模中三个不可回避的工程瓶颈：（1）金融时序天然非平稳，样本切片需兼顾因果性（label不能泄露未来信息）、滚动性（避免训练-测试集时间重叠）、长度一致性（LSTM等RNN要求固定序列长度）；（2）因子量纲差异巨大（如换手率∈[0,10]、市值对数∈[10,25]、波动率∈[0.01,0.5]），单一全局标准化（如StandardScaler）会淹没低幅值因子的判别力，而逐资产标准化又破坏跨资产可比性；（3）批处理（batching）在金融场景下存在结构性约束——同一batch内不同股票序列长度可能差异显著（退市股vs新上市股）、标签滞后步长需严格对齐（如预测t+5收益率则所有样本的label索引必须统一偏移）、填充（padding）策略若未与mask机制联动，将导致无效位置参与loss计算。

本讲不引入任何外部框架（如sklearn.preprocessing或tsfresh），全程基于PyTorch原生API构建，确保管线可无缝嵌入训练循环、支持梯度回传、兼容分布式训练（DistributedDataParallel）。所有设计均以可部署性为第一准则：即特征生成代码必须能直接用于实盘信号生成，而非仅限回测环境。

2. 依赖版本与硬件适配性声明

本讲验证环境为：Python 3.9.18、PyTorch 2.1.2（CUDA 11.8）、NumPy 1.24.4、Pandas 2.0.3。关键版本约束如下：

• PyTorch ≥ 2.0 是硬性要求：因本讲大量使用torch.compile()加速特征变换（见4.3节），该API自2.0起稳定支持；低于此版本需手动替换为torch.jit.script，但后者对动态控制流（如if len(seq) > 100）支持较弱，易触发编译失败。
• NumPy ≤ 1.24：高版本NumPy（≥1.25）在np.array()构造含nan的金融数据时默认启用strict模式，与PyTorch torch.tensor()的nan容忍策略不一致，可能导致RuntimeError: Found dtype Double but expected Float类报错（详见5.2节）。
• GPU显存建议 ≥ 16GB：因动态批处理需预分配最大序列长度的显存块，若单个batch含100只股票、每只1000日行情、每日20维因子，则原始张量尺寸为(100, 1000, 20)，float32下占用约76MB；但经滑动窗口切片（窗口=60）后生成(100, 941, 60, 20)张量，再经pad_sequence填充至统一长度，峰值显存消耗可达原始数据的3.2倍。实测在RTX 4090上，batch_size=64时显存占用稳定在12.4GB，符合预期。

环境初始化代码需显式声明设备与dtype：

import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
dtype = torch.float32  # 严禁使用torch.float64：金融因子无需双精度，且会翻倍显存

此处device与dtype必须全局统一，禁止在Dataset中创建CPU张量而在Dataloader中强制to(device)——这将引发隐式拷贝，成为性能瓶颈。所有张量应在Dataset的__getitem__中直接于目标设备上构建（如torch.empty(..., device=device, dtype=dtype)）。

3. 因果滑动窗口与多粒度切片协议

金融时序样本构建的核心矛盾在于：模型需要足够长的历史窗口捕捉趋势，但过长窗口会引入非平稳性并加剧计算负担。本讲采用分层滑动窗口协议（Hierarchical Sliding Window Protocol, HSWP），定义三个正交维度：

1. 主窗口（Primary Window）：模型输入序列长度，记为window_len。例如LSTM预测t+1收益率时，常用window_len=60（覆盖3个月交易日）。该窗口必须严格满足因果性——即索引t处的样本由[t-window_len+1, t]区间数据构成，t时刻的label（如close[t+1]/close[t]-1）不得参与输入构建。
2. 标签滞后（Label Lag）：预测目标的时间偏移量，记为label_lag。典型值：label_lag=1（次日收益）、label_lag=5（周度收益）。关键约束：label_lag必须为正整数，且label_lag < window_len，否则窗口内无足够历史支撑预测（如window_len=10, label_lag=15会导致索引越界）。
3. 步长（Stride）：窗口滑动间隔，记为stride。stride=1实现逐日滚动（最高数据利用率），但会加剧样本间相关性；stride=5降低冗余度，提升泛化性。实践中建议stride ∈ [1, window_len//3]，避免步长过大导致训练样本不足。

HSWP的实现必须规避两大反例：

• 反例1：错误的索引偏移。常见错误写法：X = data[i:i+window_len]; y = data[i+window_len+label_lag]。此写法使y对应i+window_len+label_lag时刻，而X仅覆盖至i+window_len-1，造成label_lag+1步滞后。正确应为：y = data[i+window_len-1+label_lag]，确保y基于X最后一个观测点推演。
• 反例2：忽略停牌/缺失值截断。若某股票在窗口内存在连续5日停牌（OHLC全NaN），直接切片将导致有效序列断裂。本讲采用窗口内有效长度校验：在__getitem__中计算torch.isnan(X).any(dim=(1,2))，若任一日存在全NaN，则跳过该样本（返回None并在collate_fn中过滤），而非简单前向填充——后者会污染特征分布。

参数示例：window_len=60, label_lag=3, stride=1，输入张量形状为(60, n_features)，输出标量y。对于包含1000只股票、各2000个交易日的数据集，理论样本数为1000 × (2000 - 60 + 1 - 3) = 1,940,000，实际因停牌过滤后约保留92%。

4. 跨维度解耦与在线更新机制

金融因子标准化绝非简单的StandardScaler.fit_transform()。本讲提出三维解耦标准化（Three-Dimensional Decoupled Standardization, 3DDS），分别处理资产内、时间轴、因子间三个维度的分布特性：

• 资产内标准化（Intra-Asset）：对单只股票的全部历史因子序列独立标准化。适用于个股特异性指标（如换手率、市盈率TTM），公式为：z = (x - μ_asset) / σ_asset，其中μ_asset, σ_asset为该股票全周期均值与标准差。优势：保留个股风格差异；劣势：无法跨资产比较。实现时需预计算每只股票的统计量并存为dict，避免每次__getitem__重复计算。
• 时间轴标准化（Temporal）：对每个交易日，计算当日全市场所有股票该因子的横截面均值与标准差，再对个股值标准化。公式：z_t,i = (x_t,i - μ_t,factor) / σ_t,factor。适用于市场共性因子（如行业动量、全市场波动率），能突出相对强弱。但需注意：若某日某因子全市场缺失率>30%，则μ_t,factor置为NaN，该日所有样本跳过标准化（保持原始值），防止异常传播。
• 因子间标准化（Inter-Factor）：对每个因子，计算其在全数据集（所有股票×所有日期）的全局统计量。公式：z = (x - μ_factor) / σ_factor。适用于需绝对量纲一致的场景（如多任务学习中不同label的loss权重平衡）。但必须配合clip操作：z = torch.clamp(z, -6, 6)，因金融因子常含极端值（如ST股涨跌幅限制导致的波动率尖峰），未裁剪的z-score可能达±20，破坏网络激活函数（如ReLU）的数值稳定性。

3DDS的落地难点在于在线更新（Online Update）。实盘中每日新增行情，若重新计算全局统计量成本过高。本讲采用Welford在线算法实现单次遍历更新：

# 初始化
self.n = 0
self.mean = torch.zeros(n_factors, device=device, dtype=dtype)
self.M2 = torch.zeros(n_factors, device=device, dtype=dtype)

# 每日增量更新
def update(self, x_batch):  # x_batch: (batch_size, n_factors)
    self.n += x_batch.size(0)
    delta = x_batch - self.mean
    self.mean += delta.sum(dim=0) / self.n
    delta2 = x_batch - self.mean
    self.M2 += (delta * delta2).sum(dim=0)

# 获取当前标准差
@property
def std(self):
    return torch.sqrt(self.M2 / (self.n - 1)) if self.n > 1 else torch.ones_like(self.mean)

该算法内存复杂度O(n_factors)，时间复杂度O(batch_size×n_factors)，远优于存储全量数据再计算。

5. 动态填充与掩码协同设计

PyTorch的DataLoader默认要求同batch内所有样本张量形状一致，但金融数据天然异构：新股上市初期数据稀疏、退市股末期序列短、不同指数成分股纳入时间不同。强行截断至最小长度会丢失信息，填充至最大长度则浪费显存并引入噪声。本讲采用动态填充-掩码协同（Dynamic Padding-Masking Co-design）：

1. 填充（Padding）：使用torch.nn.utils.rnn.pad_sequence，按batch内最大序列长度填充。关键参数：batch_first=True（输出形状(batch_size, max_len, n_features)），padding_value=torch.nan（显式标记填充位，便于后续mask识别）。
2. 掩码（Masking）：生成布尔型mask张量，形状同填充后X，True表示有效位置，False表示填充位。实现：mask = ~torch.isnan(X)。注意：torch.isnan()对float32安全，但对torch.bfloat16需先转float32，否则返回全False。
3. 协同应用：在模型前向传播中，mask必须与RNN/LSTM的packed_padded_sequence或Transformer的attn_mask联动。例如LSTM：

packed = torch.nn.utils.rnn.pack_padded_sequence(
    X, lengths, batch_first=True, enforce_sorted=False
)
output, _ = self.lstm(packed)
output, _ = torch.nn.utils.rnn.pad_packed_sequence(
    output, batch_first=True, padding_value=0.0
)
# 此时output中填充位对应值为0，但需在loss计算前屏蔽
valid_output = output[mask]  # 展平后取有效位置

常见错误是仅填充未掩码，导致填充位参与loss计算。例如MSE Loss：loss = ((pred - y) ** 2).mean()，若pred含填充位预测值，loss被严重低估。正确做法：loss = ((pred[mask] - y[mask]) ** 2).mean()。

边界条件：当batch内所有样本长度均为1时，pad_sequence仍正常工作，但pack_padded_sequence会警告enforce_sorted=False，此时应设置enforce_sorted=True并预先排序样本（按长度降序），避免警告影响CI/CD流水线。

6. 可复现性与状态管理

torch.utils.data.Dataset子类必须满足确定性（Deterministic）与无状态（Stateless）。反例：在__init__中调用np.random.shuffle()打乱索引——这会使同一数据集多次实例化结果不同，破坏可复现性。正确方案：

• 使用torch.Generator管理随机种子：

class FinancialDataset(Dataset):
    def __init__(self, data_dict, indices, generator=None):
        self.data_dict = data_dict  # 预加载的{stock_id: tensor}字典
        self.indices = indices      # 全局索引列表，如[(stock_id, date_idx)]
        self.generator = generator or torch.Generator()
        
    def __getitem__(self, idx):
        stock_id, date_idx = self.indices[idx]
        # 基于date_idx和window_len切片，无随机性
        return self._build_sample(stock_id, date_idx)

• Dataloader的shuffle=True时，必须传入generator：

dataloader = DataLoader(
    dataset, 
    batch_size=32,
    shuffle=True,
    generator=torch.Generator().manual_seed(42),
    collate_fn=custom_collate
)

否则多进程（num_workers>0）下各worker使用不同默认种子，导致epoch间顺序不一致。

collate_fn是批处理核心：

def custom_collate(batch):
    # 过滤None样本（停牌过滤）
    batch = [b for b in batch if b is not None]
    if not batch:
        return None
    X_list, y_list = zip(*batch)
    # 动态填充
    X_padded = pad_sequence(X_list, batch_first=True, padding_value=float('nan'))
    y_tensor = torch.stack(y_list)
    # 生成长度张量
    lengths = torch.tensor([x.size(0) for x in X_list])
    return {'X': X_padded, 'y': y_tensor, 'lengths': lengths}

注意：collate_fn必须处理空batch（如全样本被过滤），否则Dataloader迭代中断。

7. 从显存溢出到梯度错位

8. RuntimeError: CUDA out of memory

根因：动态填充后峰值显存超限，尤其当max_len远大于mean_len（如一批含新股与老股）。排查路径：

• 运行torch.cuda.memory_summary()定位显存峰值位置；
• 检查collate_fn中pad_sequence的padding_value是否为float('nan')而非0.0（后者虽省显存但破坏标准化逻辑）；
• 临时设置batch_size=1验证是否单样本即溢出，若是则检查单样本张量尺寸是否异常（如误将全市场数据载入单样本）。

9. RuntimeError: one of the variables needed for gradient computation has been modified by an inplace operation

根因：在__getitem__中使用X.copy_(...)或X[:] = ...等in-place操作修改了requires_grad=True的张量。排查路径：

• 在__getitem__末尾添加assert not X.requires_grad；
• 确保所有特征变换（如标准化）使用X_new = (X - mean) / std而非X.sub_(mean).div_(std)。

10. ValueError: Expected input to be 3D, got 2D instead

根因：pad_sequence输入list中张量维度不一致（如有的(60,20)，有的(59,20)）。排查路径：

• 在collate_fn前插入调试：print([x.shape for x in batch])；
• 检查样本构建逻辑中window_len是否被动态修改（如根据股票上市日期调整），应统一为配置参数。

11. Loss becomes NaN after epoch 0

根因：标准化后出现inf或nan，常见于σ=0（某因子全周期恒定）或log(0)（对价格取对数未加epsilon）。排查路径：

• 在__getitem__返回前添加assert not torch.isnan(X).any() and not torch.isinf(X).any()；
• 对σ添加epsilon：std = torch.where(std == 0, torch.tensor(1e-8, device=std.device), std)。

12. 从开发到实盘的五级验证清单

为确保特征管线生产就绪，执行以下验证：

1. 单元验证（Unit Validation）：对单只股票、单日数据，手工计算window_len=3, label_lag=1下的X与y，与代码输出逐元素比对。
2. 分布验证（Distribution Validation）：抽取1000个batch，统计各因子经3DDS后的均值/标准差，确认|mean| < 1e-3且|std - 1.0| < 1e-2。
3. 时序验证（Temporal Validation）：检查相邻两个batch的date_idx是否严格递增（stride=1时应为连续日期），防止时间穿越。
4. 显存验证（Memory Validation）：运行torch.cuda.memory_allocated()监控单epoch内显存增长曲线，确认无持续上升（排除内存泄漏）。
5. 梯度验证（Gradient Validation）：在训练循环中，对第一个batch执行loss.backward()后，检查model.lstm.weight.grad是否非None且无nan/inf。

落地建议：将特征管线封装为FeaturePipeline类，提供fit()（离线计算统计量）、transform()（在线生成张量）、save_state()（保存统计量至.pt文件）三接口。实盘部署时，每日调用transform()传入新行情，无需重新拟合——这正是在线更新机制的价值。

13. 系列衔接

本讲是《PyTorch量化建模完整学习计划》的第 2/12 讲，当前主题是《PyTorch金融特征管线：样本构建、标准化与批处理策略》。

上一讲：第 1 讲《PyTorch量化建模入门：张量、自动求导与训练循环》。

下一讲：第 3 讲《PyTorch时序网络基础：LSTM/GRU在收益预测中的实践》。

后续安排：第 4 讲《PyTorch Transformer在多资产序列建模中的应用》；第 5 讲《PyTorch损失函数设计：预测误差与交易目标的统一》。

14. 风险揭示与免责声明