ETT-small 数据集实战：基于 PyTorch 的 5 步时间序列预测数据加载与预处理

电力变压器温度预测是能源管理领域的关键任务，ETT-small 数据集作为业界公认的基准数据，为开发者提供了真实场景下的时间序列建模机会。本文将带您快速掌握 PyTorch 环境下数据处理的完整流程，从原始 CSV 到可直接输入模型的张量数据。

1. 环境准备与数据获取

首先确保已安装最新版 PyTorch 和常用数据处理库。推荐使用 conda 创建虚拟环境：

conda create -n ts-forecast python=3.8 conda activate ts-forecast pip install torch pandas scikit-learn matplotlib

ETT-small 数据集包含四个子集，分别对应不同时间粒度：

通过以下代码快速下载数据集：

import pandas as pd def download_ett_data(dataset_name): base_url = f"https://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-horizon+1): X.append(data[i:i+window_size]) y.append(data[i+window_size:i+window_size+horizon, -1]) # 只预测OT列 return np.array(X), np.array(y)

3. 数据标准化与分割策略

电力负载数据存在量纲差异，需要进行标准化处理。为避免数据泄露，应采用先分割后标准化策略：

from sklearn.preprocessing import StandardScaler def split_and_scale(data, train_ratio=0.7, val_ratio=0.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_size=96, horizon=24): """ :param data: 标准化后的完整数据集 :param window_size: 历史窗口长度（建议：小时数据取168=7天，分钟数据取288=2天） :param horizon: 预测步长（建议：小时数据取24=1天，分钟数据取96=4小时） """ 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_threshold=35): """ 计算样本权重，对高温时段样本增加权重 :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_size=config['batch_size'], sampler=sampler, num_workers=config['num_workers']) val_loader = DataLoader(ETTSeriesDataset(val_data, config['window_size'], config['horizon']), batch_size=config['batch_size'], shuffle=False, num_workers=config['num_workers'])

数据验证可视化代码：

import matplotlib.pyplot as plt def plot_sample(batch_idx=0): 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, label='History Temperature') # 绘制预测油温 plt.plot(range(config['window_size'], config['window_size']+config['horizon']), y_temp, 'r--', label='Future 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, sigma=3): """ 使用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, annot=True, xticklabels=feature_names, yticklabels=feature_names) plt.title('Feature Correlation Matrix') plt.show() # 使用示例 plot_correlation_matrix(train_data[:, :-1], ['HUFL','HULL','MUFL','MULL','LUFL','LULL'])

7. 生产环境优化建议

1. 内存映射技术：处理超大规模数据时使用np.memmap

large_data = np.memmap('ett_memmap.dat', dtype='float32', mode='r', shape=(70080, 8))

2. 分布式数据加载：多节点训练时采用DistributedSampler

from torch.utils.data.distributed import DistributedSampler sampler = DistributedSampler(dataset, shuffle=True) dataloader = DataLoader(dataset, batch_size=64, sampler=sampler)

3. 在线数据增强：训练时随机添加噪声

class AugmentDataset(Dataset): def __init__(self, base_dataset, noise_std=0.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

4. 缓存机制：使用lmdb数据库加速重复读取

import lmdb def write_cache(env, data): with env.begin(write=True) as txn: for idx, (x, y) in enumerate(data): txn.put(str(idx).encode(), pickle.dumps((x.numpy(), y.numpy())))

5. 自动化管道监控：记录数据统计信息

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())