【Bug已解决】PyTorch: Dataloader for time series task 解决方案问题描述在时间序列预测任务中构建合适的DataLoader是一个核心挑战。时间序列数据与普通表格数据不同它具有时间依赖性需要将连续的时间窗口转换为模型可用的输入-目标对。开发者在使用 PyTorchDataLoader处理时间序列时经常遇到以下问题如何将一维时间序列切分为滑动窗口的(sequence, label)对如何处理变长时间序列Variable Length Sequences如何正确实现多变量多步预测的数据加载如何在DataLoader中保持时间顺序不能随机打乱如何高效处理大规模时间序列数据避免内存溢出如何实现时间序列的归一化/标准化而不造成数据泄露Data Leakage本文将系统讲解时间序列DataLoader的构建方法从基础的滑动窗口到高级的多变量多步预测提供完整的可运行代码。错误复现错误 1直接使用表格数据的 DataLoader 方式import torch from torch.utils.data import Dataset, DataLoader import numpy as np # 错误将时间序列当作普通表格数据处理 time_series np.sin(np.linspace(0, 100, 1000)) # 1000个时间步的正弦波 class WrongTimeSeriesDataset(Dataset): def __init__(self, data): self.data torch.FloatTensor(data) def __len__(self): return len(self.data) def __getitem__(self, idx): # 错误每个样本只有一个点没有上下文窗口 return self.data[idx] dataset WrongTimeSeriesDataset(time_series) dataloader DataLoader(dataset, batch_size32, shuffleTrue) # shuffle 破坏了时间顺序 for batch in dataloader: print(batch.shape) # torch.Size([32])没有序列维度 break错误 2滑动窗口实现中的索引越界# 错误窗口切分时索引越界 class BuggyDataset(Dataset): def __init__(self, data, window_size10): self.data torch.FloatTensor(data) self.window_size window_size def __len__(self): return len(self.data) # 错误应该减去 window_size def __getitem__(self, idx): # 当 idx 接近末尾时会越界 x self.data[idx:idx self.window_size] y self.data[idx self.window_size] # IndexError! return x, y错误 3数据泄露Data Leakage# 错误在整个数据集上做归一化导致未来信息泄露到训练集 from sklearn.preprocessing import MinMaxScaler full_data np.random.randn(1000, 5) # 5个特征的时间序列 scaler MinMaxScaler() scaled_data scaler.fit_transform(full_data) # 用全部数据 fit包含测试集信息 # 正确做法只用训练集 fit train_size int(0.8 * len(scaled_data)) # scaler 应该只 fit(train_data)根因分析1. 时间序列数据的特殊性时间序列数据的核心特征是时间依赖性当前时刻的值与历史值相关。这意味着数据不能随机打乱shuffleTrue会破坏时间结构输入需要是一个序列窗口如过去 30 天的数据而不是单个时间点目标通常是未来一个或多个时间步的值2. 滑动窗口机制滑动窗口Sliding Window是时间序列数据加载的核心概念时间序列: [t0, t1, t2, t3, t4, t5, t6, t7, t8, t9] 窗口大小3, 预测步长1: 样本1: 输入[t0, t1, t2] → 目标[t3] 样本2: 输入[t1, t2, t3] → 目标[t4] 样本3: 输入[t2, t3, t4] → 目标[t5] ...3. 训练/验证/测试集划分时间序列的划分不能随机抽样必须按时间顺序切分|------- 训练集 -------|-- 验证 --|-- 测试 --| 80% 10% 10%解决方案方案一基础滑动窗口 Datasetimport torch from torch.utils.data import Dataset, DataLoader import numpy as np class TimeSeriesDataset(Dataset): 基础时间序列滑动窗口 Dataset 支持单变量和多变量时间序列 def __init__(self, data, sequence_length, prediction_length1, stride1): Args: data: numpy array 或 torch tensor, shape (total_timesteps, num_features) sequence_length: 输入序列长度窗口大小 prediction_length: 预测长度多步预测时 1 stride: 窗口滑动步长 super().__init__() if isinstance(data, np.ndarray): data torch.FloatTensor(data) self.data data self.sequence_length sequence_length self.prediction_length prediction_length self.stride stride # 计算有效样本数 self.num_samples (len(data) - sequence_length - prediction_length) // stride 1 def __len__(self): return self.num_samples def __getitem__(self, idx): start idx * self.stride end start self.sequence_length x self.data[start:end] y self.data[end:end self.prediction_length] return x, y方案二多变量多步预测 Datasetclass MultiVariateTimeSeriesDataset(Dataset): 多变量时间序列 Dataset 支持选择特定特征作为输入特定特征作为目标 def __init__(self, data, input_features, target_features, sequence_length, prediction_length1, stride1): Args: data: shape (total_timesteps, num_features) input_features: list of int, 输入特征的列索引 target_features: list of int, 目标特征的列索引 sequence_length: 输入序列长度 prediction_length: 预测步长 super().__init__() if isinstance(data, np.ndarray): data torch.FloatTensor(data) self.data data self.input_features input_features self.target_features target_features self.sequence_length sequence_length self.prediction_length prediction_length self.stride stride self.num_samples (len(data) - sequence_length - prediction_length) // stride 1 def __len__(self): return self.num_samples def __getitem__(self, idx): start idx * self.stride end start self.sequence_length # 输入选择输入特征 x self.data[start:end, self.input_features] # 目标选择目标特征的预测窗口 y self.data[end:end self.prediction_length, self.target_features] return x, y方案三支持归一化和反归一化的 Datasetclass NormalizedTimeSeriesDataset(Dataset): 带归一化的时间序列 Dataset 使用训练集统计量进行归一化避免数据泄露 def __init__(self, data, sequence_length, prediction_length1, train_meanNone, train_stdNone): if isinstance(data, np.ndarray): data torch.FloatTensor(data) # 归一化 if train_mean is not None and train_std is not None: self.mean train_mean self.std train_std else: self.mean data.mean(dim0) self.std data.std(dim0) self.normalized_data (data - self.mean) / (self.std 1e-8) self.sequence_length sequence_length self.prediction_length prediction_length self.num_samples len(data) - sequence_length - prediction_length 1 def __len__(self): return self.num_samples def __getitem__(self, idx): end idx self.sequence_length x self.normalized_data[idx:end] y self.normalized_data[end:end self.prediction_length] return x, y def inverse_transform(self, normalized_data, feature_idx0): 反归一化 return normalized_data * self.std[feature_idx] self.mean[feature_idx]完整修复代码以下是一个完整的时间序列预测项目包含数据生成、Dataset 定义、DataLoader 构建、LSTM 模型训练和评估import torch import torch.nn as nn import torch.optim as optim from torch.utils.data import Dataset, DataLoader import numpy as np import matplotlib matplotlib.use(Agg) import matplotlib.pyplot as plt # 数据生成 def generate_synthetic_time_series(num_points2000, num_features3): 生成多变量合成时间序列数据 t np.linspace(0, 50, num_points) # 特征1正弦波 趋势 feature1 np.sin(t * 0.5) 0.1 * t np.random.randn(num_points) * 0.1 # 特征2余弦波 季节性 feature2 np.cos(t * 0.3) np.sin(t * 2) * 0.5 np.random.randn(num_points) * 0.05 # 特征3与特征1相关的信号 feature3 feature1 * 0.5 np.sin(t * 1.2) * 0.3 np.random.randn(num_points) * 0.08 data np.stack([feature1, feature2, feature3], axis1) return data # Dataset 定义 class TimeSeriesDataset(Dataset): 通用时间序列滑动窗口 Dataset def __init__(self, data, sequence_length, prediction_length1, stride1): if isinstance(data, np.ndarray): data torch.FloatTensor(data) self.data data self.sequence_length sequence_length self.prediction_length prediction_length self.stride stride self.num_samples (len(data) - sequence_length - prediction_length) // stride 1 def __len__(self): return self.num_samples def __getitem__(self, idx): start idx * self.stride end start self.sequence_length x self.data[start:end] y self.data[end:end self.prediction_length, 0] # 预测第一个特征 return x, y class TimeSeriesScaler: 时间序列归一化器避免数据泄露 def __init__(self): self.mean None self.std None def fit(self, data): 只用训练数据 fit if isinstance(data, torch.Tensor): data data.numpy() self.mean data.mean(axis0) self.std data.std(axis0) 1e-8 return self def transform(self, data): if isinstance(data, torch.Tensor): data data.numpy() return (data - self.mean) / self.std def fit_transform(self, data): return self.fit(data).transform(data) def inverse_transform(self, data): 反归一化 if isinstance(data, torch.Tensor): return data * torch.FloatTensor(self.std) torch.FloatTensor(self.mean) return data * self.std self.mean # 数据准备 def prepare_data_loaders(data, sequence_length30, prediction_length1, batch_size64, train_ratio0.7, val_ratio0.15): 准备训练/验证/测试 DataLoader 关键按时间顺序划分不随机打乱 n len(data) train_end int(n * train_ratio) val_end int(n * (train_ratio val_ratio)) # 按时间顺序划分 train_data data[:train_end] val_data data[train_end:val_end] test_data data[val_end:] # 只用训练集计算归一化参数 scaler TimeSeriesScaler() train_scaled scaler.fit_transform(train_data) val_scaled scaler.transform(val_data) test_scaled scaler.transform(test_data) # 创建 Dataset train_dataset TimeSeriesDataset(train_scaled, sequence_length, prediction_length) val_dataset TimeSeriesDataset(val_scaled, sequence_length, prediction_length) test_dataset TimeSeriesDataset(test_scaled, sequence_length, prediction_length) # 创建 DataLoader # 关键时间序列不能 shuffle特别是验证和测试集 train_loader DataLoader(train_dataset, batch_sizebatch_size, shuffleTrue, drop_lastTrue, num_workers0) val_loader DataLoader(val_dataset, batch_sizebatch_size, shuffleFalse, drop_lastFalse, num_workers0) test_loader DataLoader(test_dataset, batch_sizebatch_size, shuffleFalse, drop_lastFalse, num_workers0) print(f数据划分: 训练 {len(train_data)} | 验证 {len(val_data)} | 测试 {len(test_data)}) print(f样本数: 训练 {len(train_dataset)} | 验证 {len(val_dataset)} | 测试 {len(test_dataset)}) return train_loader, val_loader, test_loader, scaler # 模型定义 class LSTMPredictor(nn.Module): LSTM 时间序列预测模型 def __init__(self, input_size, hidden_size, num_layers, output_size, dropout0.2): super().__init__() self.hidden_size hidden_size self.num_layers num_layers self.lstm nn.LSTM( input_sizeinput_size, hidden_sizehidden_size, num_layersnum_layers, batch_firstTrue, dropoutdropout if num_layers 1 else 0 ) self.fc nn.Linear(hidden_size, output_size) def forward(self, x): # x shape: (batch, seq_len, input_size) lstm_out, _ self.lstm(x) # 取最后一个时间步的输出 last_output lstm_out[:, -1, :] output self.fc(last_output) return output # 训练与评估 def train_model(train_loader, val_loader, model, num_epochs50, lr0.001, devicecpu): criterion nn.MSELoss() optimizer optim.Adam(model.parameters(), lrlr) scheduler optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience5, factor0.5) best_val_loss float(inf) train_losses, val_losses [], [] for epoch in range(num_epochs): # 训练 model.train() epoch_train_loss 0 for batch_x, batch_y in train_loader: batch_x, batch_y batch_x.to(device), batch_y.to(device) optimizer.zero_grad() outputs model(batch_x) # 处理目标形状 if batch_y.dim() 2 and batch_y.size(1) 1: loss criterion(outputs, batch_y) else: loss criterion(outputs.squeeze(), batch_y.squeeze()) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() epoch_train_loss loss.item() avg_train_loss epoch_train_loss / len(train_loader) train_losses.append(avg_train_loss) # 验证 model.eval() epoch_val_loss 0 with torch.no_grad(): for batch_x, batch_y in val_loader: batch_x, batch_y batch_x.to(device), batch_y.to(device) outputs model(batch_x) if batch_y.dim() 2 and batch_y.size(1) 1: loss criterion(outputs, batch_y) else: loss criterion(outputs.squeeze(), batch_y.squeeze()) epoch_val_loss loss.item() avg_val_loss epoch_val_loss / len(val_loader) val_losses.append(avg_val_loss) scheduler.step(avg_val_loss) if avg_val_loss best_val_loss: best_val_loss avg_val_loss torch.save(model.state_dict(), best_ts_model.pth) if (epoch 1) % 10 0: print(fEpoch [{epoch1}/{num_epochs}], fTrain Loss: {avg_train_loss:.6f}, fVal Loss: {avg_val_loss:.6f}) return train_losses, val_losses def evaluate_model(test_loader, model, scaler, devicecpu): 评估模型并绘制预测结果 model.eval() predictions, actuals [], [] with torch.no_grad(): for batch_x, batch_y in test_loader: batch_x batch_x.to(device) outputs model(batch_x).cpu() predictions.append(outputs) actuals.append(batch_y) predictions torch.cat(predictions, dim0).numpy() actuals torch.cat(actuals, dim0).numpy() # 反归一化只对第一个特征 predictions_real predictions * scaler.std[0] scaler.mean[0] actuals_real actuals * scaler.std[0] scaler.mean[0] # 计算 RMSE rmse np.sqrt(np.mean((predictions_real - actuals_real) ** 2)) mae np.mean(np.abs(predictions_real - actuals_real)) print(f\n测试集 RMSE: {rmse:.4f}, MAE: {mae:.4f}) # 绘制预测结果 fig, axes plt.subplots(2, 1, figsize(14, 8)) # 预测 vs 实际 axes[0].plot(actuals_real[:200], labelActual, alpha0.8) axes[0].plot(predictions_real[:200], labelPredicted, alpha0.8) axes[0].set_title(Time Series Prediction: Actual vs Predicted) axes[0].set_xlabel(Time Step) axes[0].set_ylabel(Value) axes[0].legend() axes[0].grid(True, alpha0.3) # 预测误差 errors predictions_real.flatten() - actuals_real.flatten() axes[1].hist(errors, bins50, alpha0.7, colorsteelblue) axes[1].axvline(x0, colorred, linestyle--) axes[1].set_title(Prediction Error Distribution) axes[1].set_xlabel(Error) axes[1].set_ylabel(Frequency) plt.tight_layout() plt.savefig(ts_prediction_results.png, dpi150) print(预测结果图已保存为 ts_prediction_results.png) return rmse, mae # 主程序 def main(): # 设置随机种子 torch.manual_seed(42) np.random.seed(42) # 生成数据 print( * 60) print(生成合成时间序列数据...) data generate_synthetic_time_series(num_points2000, num_features3) print(f数据形状: {data.shape} (timesteps, features)) # 准备 DataLoader print(\n * 60) print(准备 DataLoader...) train_loader, val_loader, test_loader, scaler prepare_data_loaders( data, sequence_length30, prediction_length1, batch_size64 ) # 检查数据形状 for batch_x, batch_y in train_loader: print(f\nBatch 输入形状: {batch_x.shape}) # (64, 30, 3) print(fBatch 目标形状: {batch_y.shape}) # (64, 1) break # 创建模型 print(\n * 60) print(创建 LSTM 模型...) device torch.device(cuda if torch.cuda.is_available() else cpu) model LSTMPredictor( input_size3, hidden_size64, num_layers2, output_size1, dropout0.2 ).to(device) print(f设备: {device}) print(f模型参数量: {sum(p.numel() for p in model.parameters())}) # 训练 print(\n * 60) print(开始训练...) train_losses, val_losses train_model( train_loader, val_loader, model, num_epochs30, lr0.001, devicedevice ) # 绘制训练曲线 fig, ax plt.subplots(figsize(10, 5)) ax.plot(train_losses, labelTrain Loss) ax.plot(val_losses, labelVal Loss) ax.set_xlabel(Epoch) ax.set_ylabel(Loss (MSE)) ax.set_title(Training and Validation Loss) ax.legend() ax.grid(True, alpha0.3) plt.tight_layout() plt.savefig(ts_training_curve.png, dpi150) print(训练曲线已保存为 ts_training_curve.png) # 加载最佳模型并评估 print(\n * 60) print(评估模型...) model.load_state_dict(torch.load(best_ts_model.pth, weights_onlyTrue)) rmse, mae evaluate_model(test_loader, model, scaler, device) print(\n时间序列预测任务完成) if __name__ __main__: main()常见陷阱与注意事项1. 训练集 shuffle 的选择# 时间序列训练集可以 shuffle打乱样本顺序不破坏窗口内时间结构 train_loader DataLoader(train_dataset, batch_size64, shuffleTrue) # 但验证和测试集不应 shuffle以便按时间顺序评估 val_loader DataLoader(val_dataset, batch_size64, shuffleFalse)2. 归一化的数据泄露# 错误用全部数据计算归一化参数 scaler.fit(all_data) # 测试集信息泄露到训练集 # 正确只用训练集计算 scaler.fit(train_data) train_scaled scaler.transform(train_data) val_scaled scaler.transform(val_data) # 用训练集的参数 test_scaled scaler.transform(test_data) # 用训练集的参数3. 多步预测的处理# 单步预测prediction_length1 # 输入: (batch, seq_len, features) → 输出: (batch, 1) # 多步预测prediction_length5 # 输入: (batch, seq_len, features) → 输出: (batch, 5) # 需要修改模型输出层和损失计算 # 递归多步预测用预测结果作为下一步输入 def recursive_predict(model, initial_input, steps10): predictions [] current_input initial_input.clone() for _ in range(steps): pred model(current_input.unsqueeze(0)) predictions.append(pred) # 将预测值加入输入窗口 current_input torch.cat([current_input[1:], pred], dim0) return torch.cat(predictions, dim0)4. 变长时间序列的 collate_fndef collate_fn(batch): 处理变长序列按长度降序排列 batch.sort(keylambda x: len(x[0]), reverseTrue) sequences, targets zip(*batch) # 填充到最大长度 lengths [len(seq) for seq in sequences] max_len max(lengths) padded_seqs torch.zeros(len(sequences), max_len, sequences[0].size(-1)) for i, seq in enumerate(sequences): padded_seqs[i, :len(seq)] seq return padded_seqs, torch.tensor(lengths), torch.stack(targets)5. 大规模数据的内存优化# 使用 memmap 处理超大文件 data np.memmap(large_timeseries.npy, dtypefloat32, moder, shape(10000000, 5)) # Dataset 中按需读取不一次性加载 class LargeTimeSeriesDataset(Dataset): def __init__(self, memmap_data, seq_len, pred_len): self.data memmap_data # memmap 不会全部加载到内存 self.seq_len seq_len self.pred_len pred_len def __getitem__(self, idx): # 每次只读取需要的窗口 x torch.FloatTensor(self.data[idx:idx self.seq_len]) y torch.FloatTensor(self.data[idx self.seq_len:idx self.seq_len self.pred_len, 0]) return x, y总结构建时间序列DataLoader的核心要点滑动窗口将连续时间序列切分为(input_sequence, target)对这是时间序列数据加载的基础。时间顺序划分训练/验证/测试集必须按时间顺序切分不能随机抽样。避免数据泄露归一化参数只能从训练集计算然后应用到验证和测试集。shuffle 策略训练集可以 shuffle不破坏窗口内时间结构验证/测试集不应 shuffle。多变量支持通过特征列索引灵活选择输入特征和目标特征。内存优化对于大规模数据使用np.memmap或按需读取避免内存溢出。多步预测通过调整prediction_length或使用递归预测实现多步预测。通过掌握这些方法你可以高效地构建适用于各种时间序列任务的数据加载管道从简单的单变量预测到复杂的多变量多步预测。