ETT-small 数据集实战基于 PyTorch 的 5 步时间序列预测数据加载与预处理电力变压器温度预测是能源管理领域的关键任务ETT-small 数据集作为业界公认的基准数据为开发者提供了真实场景下的时间序列建模机会。本文将带您快速掌握 PyTorch 环境下数据处理的完整流程从原始 CSV 到可直接输入模型的张量数据。1. 环境准备与数据获取首先确保已安装最新版 PyTorch 和常用数据处理库。推荐使用 conda 创建虚拟环境conda create -n ts-forecast python3.8 conda activate ts-forecast pip install torch pandas scikit-learn matplotlibETT-small 数据集包含四个子集分别对应不同时间粒度数据集名称时间粒度记录频率数据点数量ETTh1小时级每小时17,544ETTh2小时级每小时17,544ETTm1分钟级15分钟70,080ETTm2分钟级15分钟70,080通过以下代码快速下载数据集import pandas as pd def download_ett_data(dataset_name): base_url fhttps://raw.githubusercontent.com/zhouhaoyi/ETDataset/main/ETT-small/{dataset_name}.csv return pd.read_csv(base_url) # 示例加载ETTh1数据集 ett_data download_ett_data(ETTh1)2. 数据结构解析与特征工程原始数据包含8个关键字段其物理意义如下# 字段说明字典 feature_description { date: 记录时间戳YYYY-MM-DD HH:MM:SS, HUFL: 高有用负载High UseFul Load, HULL: 高无用负载High UseLess Load, MUFL: 中有用负载Middle UseFul Load, MULL: 中无用负载Middle UseLess Load, LUFL: 低有用负载Low UseFul Load, LULL: 低无用负载Low UseLess Load, OT: 油温Oil Temperature预测目标 }时间序列预测通常需要构建滑动窗口样本。以下函数将原始数据转换为监督学习格式def create_sliding_windows(data, window_size, horizon): 创建滑动窗口样本 :param data: 输入特征矩阵 (seq_len, num_features) :param window_size: 历史窗口长度 :param horizon: 预测步长 :return: (X, y) 特征-标签对 X, y [], [] for i in range(len(data)-window_size-horizon1): X.append(data[i:iwindow_size]) y.append(data[iwindow_size:iwindow_sizehorizon, -1]) # 只预测OT列 return np.array(X), np.array(y)3. 数据标准化与分割策略电力负载数据存在量纲差异需要进行标准化处理。为避免数据泄露应采用先分割后标准化策略from sklearn.preprocessing import StandardScaler def split_and_scale(data, train_ratio0.7, val_ratio0.15): 数据集分割与标准化 :return: (train_data, val_data, test_data), (feature_scaler, target_scaler) n len(data) train_end int(n * train_ratio) val_end train_end int(n * val_ratio) # 分割数据集 train data[:train_end] val data[train_end:val_end] test data[val_end:] # 初始化标准化器 feature_scaler StandardScaler().fit(train[:, :-1]) target_scaler StandardScaler().fit(train[:, -1:]) # 应用标准化 def scale_dataset(dataset): scaled_features feature_scaler.transform(dataset[:, :-1]) scaled_target target_scaler.transform(dataset[:, -1:]) return np.column_stack([scaled_features, scaled_target]) return (scale_dataset(train), scale_dataset(val), scale_dataset(test)), (feature_scaler, target_scaler)注意油温(OT)需要单独标准化因为预测任务需要还原真实温度值4. PyTorch Dataset 实现创建自定义Dataset类实现高效数据加载import torch from torch.utils.data import Dataset class ETTSeriesDataset(Dataset): def __init__(self, data, window_size96, horizon24): :param data: 标准化后的完整数据集 :param window_size: 历史窗口长度建议小时数据取1687天分钟数据取2882天 :param horizon: 预测步长建议小时数据取241天分钟数据取964小时 self.X, self.y create_sliding_windows(data, window_size, horizon) def __len__(self): return len(self.X) def __getitem__(self, idx): return torch.FloatTensor(self.X[idx]), torch.FloatTensor(self.y[idx])高级技巧实现样本权重分配解决数据不平衡问题def compute_sample_weights(dataset, temperature_threshold35): 计算样本权重对高温时段样本增加权重 :param temperature_threshold: 油温警戒阈值摄氏度 _, y dataset[:] # 获取所有标签 original_temps target_scaler.inverse_transform(y.cpu().numpy()) weights np.where(original_temps temperature_threshold, 2.0, 1.0) return torch.FloatTensor(weights)5. DataLoader 配置与验证最终构建完整数据管道from torch.utils.data import DataLoader, WeightedRandomSampler # 参数配置 config { batch_size: 64, window_size: 168, # 7天历史数据 horizon: 24, # 预测未来24小时 num_workers: 4 } # 创建数据集 (train_data, val_data, test_data), scalers split_and_scale(ett_data.values) train_dataset ETTSeriesDataset(train_data, config[window_size], config[horizon]) # 创建带权重的采样器 weights compute_sample_weights(train_dataset) sampler WeightedRandomSampler(weights, len(weights)) # 构建DataLoader train_loader DataLoader(train_dataset, batch_sizeconfig[batch_size], samplersampler, num_workersconfig[num_workers]) val_loader DataLoader(ETTSeriesDataset(val_data, config[window_size], config[horizon]), batch_sizeconfig[batch_size], shuffleFalse, num_workersconfig[num_workers])数据验证可视化代码import matplotlib.pyplot as plt def plot_sample(batch_idx0): X, y next(iter(train_loader)) plt.figure(figsize(12, 6)) # 反标准化数据 X_ X[batch_idx].cpu().numpy() X_features scalers[0].inverse_transform(X_[:, :-1]) X_temp scalers[1].inverse_transform(X_[:, -1:]) y_ y[batch_idx].cpu().numpy() y_temp scalers[1].inverse_transform(y_.reshape(-1, 1)) # 绘制历史油温 plt.plot(range(config[window_size]), X_temp, labelHistory Temperature) # 绘制预测油温 plt.plot(range(config[window_size], config[window_size]config[horizon]), y_temp, r--, labelFuture to Predict) plt.legend() plt.title(Sample from Training Set) plt.xlabel(Time Steps) plt.ylabel(Oil Temperature (°C)) plt.grid(True) plt.show() plot_sample()6. 高级预处理技巧6.1 时间特征增强原始时间戳可分解为更有意义的特征def enhance_time_features(df): 从date列提取时间特征 df[hour] df[date].dt.hour df[day_of_week] df[date].dt.dayofweek df[day_of_month] df[date].dt.day df[month] df[date].dt.month df[is_weekend] df[day_of_week].isin([5,6]).astype(int) # 周期编码 df[hour_sin] np.sin(2*np.pi*df[hour]/24) df[hour_cos] np.cos(2*np.pi*df[hour]/24) return df.drop(columns[date])6.2 异常值处理基于3σ原则自动检测异常数据def detect_outliers(data, feature_idx, sigma3): 使用Z-score检测异常值 :return: 异常值索引列表 feature_data data[:, feature_idx] z_scores (feature_data - np.mean(feature_data)) / np.std(feature_data) return np.where(np.abs(z_scores) sigma)[0]6.3 多变量相关性分析计算特征间Pearson相关系数def plot_correlation_matrix(data, feature_names): corr_matrix np.corrcoef(data.T) plt.figure(figsize(10,8)) sns.heatmap(corr_matrix, annotTrue, xticklabelsfeature_names, yticklabelsfeature_names) plt.title(Feature Correlation Matrix) plt.show() # 使用示例 plot_correlation_matrix(train_data[:, :-1], [HUFL,HULL,MUFL,MULL,LUFL,LULL])7. 生产环境优化建议内存映射技术处理超大规模数据时使用np.memmaplarge_data np.memmap(ett_memmap.dat, dtypefloat32, moder, shape(70080, 8))分布式数据加载多节点训练时采用DistributedSamplerfrom torch.utils.data.distributed import DistributedSampler sampler DistributedSampler(dataset, shuffleTrue) dataloader DataLoader(dataset, batch_size64, samplersampler)在线数据增强训练时随机添加噪声class AugmentDataset(Dataset): def __init__(self, base_dataset, noise_std0.01): self.base_dataset base_dataset self.noise_std noise_std def __getitem__(self, idx): x, y self.base_dataset[idx] if self.training: x x torch.randn_like(x) * self.noise_std return x, y缓存机制使用lmdb数据库加速重复读取import lmdb def write_cache(env, data): with env.begin(writeTrue) as txn: for idx, (x, y) in enumerate(data): txn.put(str(idx).encode(), pickle.dumps((x.numpy(), y.numpy())))自动化管道监控记录数据统计信息class DataMonitor: def __init__(self): self.stats defaultdict(list) def update(self, batch): self.stats[mean].append(batch.mean().item()) self.stats[std].append(batch.std().item()) self.stats[max].append(batch.max().item())