如何为一维CNN时间序列嵌入实现正确的掩码处理？

你的核心需求是：让每个特征对应的CNN在处理第t个时间步时，只能用到该特征第t步及之前的历史数据，同时适配[sequences, features, timesteps]的输入结构，避免未来数据泄露。下面给两种可行方案：

方案一：手动生成时间步掩码（精准控制历史窗口） #

你当前的二维掩码仅标记了序列的有效长度，需要扩展为时间步依赖掩码，同时结合有效长度掩码，最终得到适配多特征的4维掩码：

代码实现 #

# 假设 inputs_padded 是形状为 [sequences, features, timesteps] 的输入张量
sequences, features, timesteps = inputs_padded.shape
device = inputs_padded.device

# 1. 生成未来时间步屏蔽掩码：仅保留当前时间步及之前的历史
timestep_idx = torch.arange(timesteps, device=device)
# 生成上三角矩阵（含对角线），shape [timesteps, timesteps]
# future_mask[t, s] = True 表示时间步 t 可以使用时间步 s 的数据（s <= t）
future_mask = timestep_idx.unsqueeze(1) <= timestep_idx.unsqueeze(0)
# 扩展维度适配序列和特征：shape [1, 1, timesteps, timesteps]，可广播到所有特征
future_mask = future_mask.unsqueeze(0).unsqueeze(0)

# 2. 生成序列有效长度掩码（基于你原来的代码优化）
seq_len = torch.LongTensor([len(x) for x in inputs]).to(device)
# shape [sequences, timesteps]，标记每个序列的有效时间步
seq_valid_mask = torch.arange(timesteps, device=device)[None, :] < seq_len[:, None]
# 扩展维度：shape [sequences, 1, timesteps, 1]，适配4维掩码结构
seq_valid_mask = seq_valid_mask.unsqueeze(1).unsqueeze(-1)

# 3. 合并掩码：同时屏蔽未来时间步和无效padding
final_mask = future_mask & seq_valid_mask
# final_mask 形状为 [sequences, 1, timesteps, timesteps]，可广播到特征维度

掩码使用方式 #

将输入张量扩展维度后与掩码相乘，得到每个时间步对应的历史输入：

# 扩展输入维度：[sequences, features, timesteps] → [sequences, features, timesteps, 1]
input_expanded = inputs_padded.unsqueeze(-1)
# 应用掩码，得到 shape [sequences, features, timesteps, timesteps] 的历史输入
masked_input = input_expanded * final_mask.float()

# 对每个特征单独处理：遍历特征维度，将 masked_input[:, i, :, :] 传入CNN
processed_features = []
for i in range(features):
    # 取第i个特征的历史输入，shape [sequences, timesteps, timesteps]
    feat_input = masked_input[:, i, :, :].transpose(1, 2)  # 调整维度适配CNN输入
    cnn_output = your_cnn(feat_input)
    processed_features.append(cnn_output)

# 拼接所有特征的处理结果，输入LSTM
concat_output = torch.cat(processed_features, dim=1)
lstm_input = concat_output.transpose(1, 2)  # 调整为LSTM需要的 [sequences, timesteps, hidden_size]

方案二：使用因果卷积（更简洁高效） #

不需要手动生成掩码，直接用因果卷积让CNN自动只使用当前及之前的时间步数据，这是时序任务中避免未来泄露的常用方法：

代码实现 #

import torch.nn as nn

# 定义因果CNN模块：仅对左边填充，确保不会用到未来数据
class CausalConv1d(nn.Module):
    def __init__(self, in_channels, out_channels, kernel_size):
        super().__init__()
        # 因果卷积需要左边填充 kernel_size-1 个0
        self.padding = kernel_size - 1
        self.conv = nn.Conv1d(
            in_channels=in_channels,
            out_channels=out_channels,
            kernel_size=kernel_size,
            padding=(self.padding, 0),  # 仅左边填充
            bias=False
        )
    
    def forward(self, x):
        # x shape: [sequences, 1, timesteps]（单个特征输入）
        x = self.conv(x)
        # 去掉因填充产生的多余长度，保证输出与输入时间步一致
        if self.padding > 0:
            x = x[:, :, :-self.padding]
        return x

# 为每个特征创建独立的因果CNN，或者用分组卷积优化
class FeatureWiseCausalCNN(nn.Module):
    def __init__(self, num_features, in_channels=1, out_channels=64, kernel_size=3):
        super().__init__()
        # 为每个特征分配一个独立的因果CNN
        self.cnns = nn.ModuleList([
            CausalConv1d(in_channels, out_channels, kernel_size)
            for _ in range(num_features)
        ])
    
    def forward(self, x):
        # x shape: [sequences, features, timesteps]
        processed = []
        for i in range(x.size(1)):
            # 提取单个特征，shape [sequences, 1, timesteps]
            feat = x[:, i:i+1, :]
            processed.append(self.cnns[i](feat))
        # 拼接所有特征的输出，shape [sequences, num_features*out_channels, timesteps]
        concat = torch.cat(processed, dim=1)
        # 调整为LSTM需要的输入格式：[sequences, timesteps, hidden_size]
        return concat.transpose(1, 2)

使用方式 #

# 初始化模型
feature_cnn = FeatureWiseCausalCNN(num_features=features, out_channels=64, kernel_size=3)
lstm = nn.LSTM(input_size=features*64, hidden_size=128, batch_first=True)

# 前向传播
cnn_output = feature_cnn(inputs_padded)
lstm_output, _ = lstm(cnn_output)

内容的提问来源于stack exchange，提问作者john aves