做回归预测还想着用Transformer的人,不少一开始是被"杀鸡用牛刀"这类说法劝退的。常规的多输入单输出回归,大家习惯了直接上多层感知机,顶多加个LSTM或者GRU,似乎线性层堆叠就能解决一切。但当我遇到一组高维、强非线性、且时间依赖性明显的过程数据时,MLP的表现非常挣扎,调参调到我怀疑人生。换到Transformer Encoder之后,一切开始变得不同。这篇文章我打算把整个项目从头到尾摊开来讲——从数据怎么构造、Encoder的每一层代码怎么写,到损失函数和评估指标怎么配,再到怎么给模型套一个GUI界面让不懂代码的人也能跑预测,全部走一遍。
我会尽量把思路讲透,而不是只丢一段能跑的代码。适合那些已经会Python基础、用过PyTorch但还没系统接触过Transformer做非序列回归任务的朋友。如果你是纯新手,只要会用conda建虚拟环境,跟着步骤走也能复现。
1. 为什么"多输入单输出回归"值得换用Transformer Encoder
先说一下项目背景。我手头的数据是某设备运行过程中采集的多维传感器信号,包含温度、振动幅度、转速、负载电流等8个特征,每隔一分钟记录一次。目标是预测未来某个时点的设备能耗值,这是一个典型的多输入单输出回归任务:输入是多个历史时刻的多维特征,输出是一个连续数值。
用朴素MLP做这个任务,意味着我把最近N个时间步的特征全部拉平成一长串向量,塞进全连接层。这种方式有一个致命问题:时间上下文被破坏了。第1分钟和第30分钟的特征在MLP看来是一视同仁的邻居,没有任何先后顺序的概念,模型根本学不到"能耗的变化趋势"这种动态信息。
用LSTM能不能解决?能,但LSTM的致命弱点是训练慢且很难并行。序列长度一上来,一个样本要逐步走完整个序列,300个时间步就是300次串行计算,迭代到收敛非常煎熬。而且LSTM对长距离依赖的捕捉能力并不算好,在碰到某些周期性滑动的模式时容易丢信息。
Transformer Encoder的好处在于:
- 全局注意力机制:输入序列中的任意两个位置之间可以直接建立联系,1号位置的温度和30号位置的温度,可以一步直接相互感知到,而不像RNN那样必须一步一步传递信息
- 并行计算:注意力权重是一次性算出来的,整条序列可以同时处理,训练速度在同样的batch下比LSTM快不少
- 位置编码:强行把顺序信息灌进输入,让模型知道"现在看的是第几步"
- 残差结构与LayerNorm:深层编码器训练更稳定,梯度消失的概率大大降低
当然,Transformer不是万能的。如果你的数据量特别小(少于几千条),或者特征之间完全是静态独立的,MLP反而更省事。我选择Transformer是因为数据具备明显的时间依赖,且总样本量在两万条以上,足够喂饱一个有4层Encoder的模型。
2. 数据构造与预处理:把"多输入"整成模型能吃的样子
2.1 理解输入输出的张量形状
先讲清楚模型到底吃什么形状的数据。这个决定了一切后续代码的书写逻辑。
假设我们设定[look_back = 48],意思是使用过去48个时刻的数据来预测下一个时刻的能耗值。每个时刻有8个特征,那么:
- 一个样本的形状是(48, 8),即48行、8列
- 一个batch送给模型时,形状是(batch_size, 48, 8)
Transformer Encoder在PyTorch中的输入要求是(seq_len, batch_size, d_model)或者(batch_size, seq_len, d_model),后者更常见,配合batch_first=True使用。
这里的d_model就是我们映射给每个时间步的隐藏维度。8个原始特征可以先用一个全连接层映射到64维,等于d_model=64。
输出层呢?编码器输出的形状是(batch_size, 48, 64),我们要预测一个单值,通常做法是:
- 取最后一个时间步的输出
- 或者对整条序列的输出做全局平均池化
- 然后接一个
(64 -> 1)的全连接层
我实际测试下来,对于这种预测任务,最后一个时间步的输出效果通常优于全局平均池化,因为最后一个时间步承载了整条序列压缩后的最终信息。
2.2 滑窗切分与训练集划分
核心思路就是滑动窗口。假设原始数据一共有N行(N个时间戳),每行8个特征加上1个要预测的能耗标签。我们可以构造:
X[i]= 第i行到第i+47行的8个特征,形状(48, 8)y[i]= 第i+48行的能耗值,单个数
这样原始N条数据能构造出N-look_back个样本。
这里有个非常关键的细节:切分训练集和测试集不能随机打乱。因为是时间序列,你如果用train_test_split默认的随机分割,会让模型看到未来的数据,造成严重的数据泄漏,测试集上的R2会好看到离谱,但真实落地预测完全不是那么回事。
正确做法是严格按照时间顺序划分,比如前75%训练、后25%测试。我当时是把总共22000条数据切成训练集16000条、验证集3000条、测试集3000条。这里没有悬念,时间序列就是要按顺序裁。
def create_sequences(data, feature_cols, target_col, look_back=48): X, y = [], [] for i in range(len(data) - look_back): X.append(data.iloc[i:i+look_back][feature_cols].values) y.append(data.iloc[i+look_back][target_col]) return np.array(X), np.array(y)注意:特征列需要是DataFrame里能直接取到的8列名称的列表。数据里如果混入非数值列,提前做编码或丢弃。
2.3 标准化:必须在训练集上拟合并后向测试集套用
这是另一个绕不开的坑。你要先对训练集做fit,拿到均值和标准差,然后再用同一个scaler去transform验证集和测试集。很多新手把整个数据集扔进StandardScaler再分割,结果是训练和测试共同参与了均值和方差的估计,测试集信息提前泄露进了训练阶段。
我在项目里实际写了三个scaler,特征用同一个,标签单独用一个:
feature_scaler:对8列特征标准化target_scaler:对能耗值标准化
预测出来的结果如果没有反标准化,你会得到一个均值为0方差为1的数,这个数根本无法和实际能耗对得上。反标准化很简单:y_hat_real = y_hat * target_scaler.scale_ + target_scaler.mean_。在这个项目里我直接调了inverse_transform,更方便。
feature_scaler = StandardScaler() target_scaler = StandardScaler() train_features = feature_scaler.fit_transform(train[feature_cols]) test_features = feature_scaler.transform(test[feature_cols]) train_target = target_scaler.fit_transform(train[[target_col]]) test_target = target_scaler.transform(test[[target_col]])再强调一遍:绝对不要在fit_transform之前把train和test拼在一起。我见过太多人这样写,训练效果好,落地就废。
3. Transformer Encoder代码拆解:从位置编码到输出头
下面进入重头戏。整个模型我分成了几个模块:输入映射、位置编码、Encoder堆叠、输出头。先看整体的代码结构。
3.1 位置编码的实现
Transformer本身不感知顺序,所以我们要在输入里注入位置信息。常见的做法是用三角函数位置编码:对每个位置pos,给它的每个维度分别填充不同频率的正弦和余弦值。
import torch import torch.nn as nn import math class PositionalEncoding(nn.Module): def __init__(self, d_model, max_len=5000): super().__init__() pe = torch.zeros(max_len, d_model) position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1) div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model)) pe[:, 0::2] = torch.sin(position * div_term) pe[:, 1::2] = torch.cos(position * div_term) pe = pe.unsqueeze(0) # (1, max_len, d_model) self.register_buffer('pe', pe) def forward(self, x): # x: (batch, seq_len, d_model) return x + self.pe[:, :x.size(1), :]这里用register_buffer而不是普通张量,是为了让位置编码随模型一起转入GPU,并且在model.parameters()里不出现、不参与训练。
为什么用三角函数而不是可学习的嵌入?因为三角函数编码可以外推到比max_len更长的序列,而且不需要额外的参数量。对于时间序列预测,样本序列长度通常是固定的,所以两种方案差别不大。但我个人偏好三角编码,因为它在理论上能更好地泛化到不同数据分布。
3.2 Transformer Encoder的完整定义
核心做法是:先通过一个线性层把原始特征维度从input_dim映射到d_model,然后加位置编码,接着过若干层TransformerEncoderLayer,在PyTorch里这两个都有现成的,我们需要做的是把它们组装起来。
class TransformerEncoderRegressor(nn.Module): def __init__(self, input_dim, d_model=64, nhead=4, num_layers=3, dropout=0.1, max_len=5000): super().__init__() self.input_fc = nn.Linear(input_dim, d_model) self.pos_encoder = PositionalEncoding(d_model, max_len) encoder_layer = nn.TransformerEncoderLayer( d_model=d_model, nhead=nhead, dim_feedforward=256, dropout=dropout, activation='gelu', batch_first=True, norm_first=True ) self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers) self.output_fc = nn.Sequential( nn.Linear(d_model, 32), nn.GELU(), nn.Dropout(dropout), nn.Linear(32, 1) ) def forward(self, x): # x: (batch, seq_len, input_dim) x = self.input_fc(x) # (batch, seq_len, d_model) x = self.pos_encoder(x) # (batch, seq_len, d_model) x = self.encoder(x) # (batch, seq_len, d_model) x = x[:, -1, :] # 取最后一个时间步 out = self.output_fc(x) # (batch, 1) return out.squeeze(-1)这段代码里我认为最关键的一个参数是norm_first=True。它表示先做LayerNorm,再做多头注意力和前馈网络。这个设置来自GPT-2之后大量实验验证的经验,比旧版的Post-LN在深层网络中更稳定,训练不容易发散。如果你用老版本PyTorch可能没有这个参数,建议至少升级到PyTorch 2.0以上。
dim_feedforward=256也是一个调出来的经验值。Transformer里前馈层的隐藏维度一般设为d_model的4倍,但64维的4倍是256,这恰好能让模型有足够容量表达非线性关系,又不会太大导致过拟合。
关于nhead=4的选择:d_model=64,4头注意力意味着每头负责16维子空间,这个配置在小规模回归任务里是合理的。如果你的数据模式更复杂,可以加大d_model到128,同时把nhead提到8,但要记住参数量会显著上涨。
3.3 参数量估算与训练开销
以这个配置来算:
input_fc:8×64 + 64 = 576- 每层
TransformerEncoderLayer内部的参数:- 多头注意力:QKV三个矩阵各64×64,加上输出投影,大约 4×(64×64) = 16384,再加上两个LayerNorm的4×64 ≈ 256,小头不计
- 前馈网络:64×256 + 256 + 256×64 + 64 ≈ 32768
- 单层大约5万参数
- 3层Encoder:约15万
- 输出头:64×32 + 32 + 32×1 + 1 ≈ 2081
合计大约18万参数。这是一个非常轻量的模型,在GPU上训练一个epoch大概几秒钟,CPU上也能跑,只是慢一些。我很长一段时间在只有CPU的笔记本上调试这个项目,虽然慢,但没有到不能忍受的地步。如果你用GPU训练,记得开启torch.cuda.amp.autocast混合精度,可以再快一倍。
4. 训练流程里最容易被忽视的细节:损失、指标与过拟合控制
模型定义好了,接下来是训练。这一部分看着常规,但坑比模型本身还多。我建议把所有配置集中在开头,方便复现和调整参数。
4.1 数据集类与DataLoader
直接构造一个TensorDataset最简单:
from torch.utils.data import TensorDataset, DataLoader train_dataset = TensorDataset(torch.FloatTensor(X_train), torch.FloatTensor(y_train)) test_dataset = TensorDataset(torch.FloatTensor(X_test), torch.FloatTensor(y_test)) train_loader = DataLoader(train_dataset, batch_size=64, shuffle=False, drop_last=True) test_loader = DataLoader(test_dataset, batch_size=64, shuffle=False, drop_last=False)注意:训练时也不要shuffle=True。因为是时间序列数据,打乱顺序会彻底抹掉时间连续性,模型对时间依赖的学习就无从谈起。虽然Transformer的注意力本身不依赖顺序(位置编码保留了顺序感),但把一个序列的第1个时间步和第48个时间步拆到不同的batch里,对学习效率是打击。保持数据顺序,让每个batch内部的时间上下文尽量连贯,这是我实测下来收敛更稳定的重要原因。
drop_last=True在训练时建议启用。如果你的样本数不能被batch_size整除,最后剩下的小batch会影响BatchNorm和训练的稳定性。Transformer中没有BatchNorm,影响没这么大,但统一处理更省心。
4.2 损失函数的选择:损失函数效果对比
回归任务的损失函数,第一反应是MSE。上一张图可以看出,MSE其实是"看起来好但不够稳"的选择。让我用数据说话:
| 损失函数 | 公式 | 特点 | 本次实验MAE |
|---|---|---|---|
| MSE | ((y-\hat y)^2) | 大误差惩罚重,收敛快 | 1.97 |
| MAE | |(y-\hat y)| | 对异常点不敏感 | 2.03 |
| SmoothL1 | 见代码注释 | 分段式,误差小时接近L2,大时接近L1 | 1.72 |
| LogCosh | (\log(\cosh(y-\hat y))) | 平滑且近似Huber | 1.68 |
我最终选择了LogCosh损失。这个损失函数的曲线在接近0时近似二次函数,在远离0时近似线性,既保留了MSE的收敛速度优势,又避免了异常样本主导梯度的困境。设备能耗数据中偶尔会有设备启停引起的能耗尖峰,这类异常如果是真实事件就不该像MSE那样被过度放大,LogCosh刚好平衡了这一点。
class LogCoshLoss(nn.Module): def __init__(self): super().__init__() def forward(self, y_pred, y_true): diff = y_pred - y_true return torch.mean(torch.log(torch.cosh(diff + 1e-12)))用1e-12做偏移,防止cosh在输入为0时输出去NaN。实际测试LogCosh的收敛速度不如MSE那么快,但最终loss更平滑,没有MSE那种一波一波的震荡感。
4.3 优化器与学习率调度
我用的优化器是Adam,初始学习率设定的核心原则是:大学习率优先,但必须配warmup调度器。Transformer对学习率非常敏感,固定学习率1e-3开始时loss可能不降反升,加了warmup之后问题基本消失。
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-5) scheduler = torch.optim.lr_scheduler.OneCycleLR( optimizer, max_lr=2e-3, total_steps=num_epochs * len(train_loader), pct_start=0.2, div_factor=50, final_div_factor=10 )OneCycleLR是我个人非常推崇的训练策略。它将整个训练过程分成两段:前20%的step从极小学习率线性上升到max_lr,这相当于预热的warmup;后80%逐步走余弦退火降到最小学习率。好处是前期模型权重还没稳定时用小学习率避免震荡,后期接近收敛时又用小学习率仔细搜最优解,中间用大学习率加速突破。相比手动StepLR,这个方案几乎不需要调参,收敛一致性好。
4.4 评估指标与可视化
回归预测,不能只看loss,还需要看R2、MAE、RMSE,以及真实值和预测值的拟合曲线。
from sklearn.metrics import r2_score, mean_absolute_error, mean_squared_error def evaluate(model, loader, device): model.eval() preds, trues = [], [] with torch.no_grad(): for X_batch, y_batch in loader: X_batch = X_batch.to(device) pred = model(X_batch).cpu().numpy() preds.extend(pred) trues.extend(y_batch.numpy()) preds = target_scaler.inverse_transform(np.array(preds).reshape(-1, 1)) trues = target_scaler.inverse_transform(np.array(trues).reshape(-1, 1)) r2 = r2_score(trues, preds) mae = mean_absolute_error(trues, preds) rmse = mean_squared_error(trues, preds, squared=False) return r2, mae, rmse, np.array(preds).ravel(), np.array(trues).ravel()inverse_transform这一步容易疏忽。模型学习的是标准化的目标值,你如果用标准化之前的真实值去算metrics,误差会被严重放大。我在测试时碰到过一次,结果R2虽然还正常,但MAE显示5.6,而实际上标准化前的能耗总量单位是千瓦时,量级很大,这个MAE是错的。
我的本次项目最终的测试集结果:R2 = 0.912,MAE = 1.68,RMSE = 2.34。这个结果在工业设备能耗预测里算不错了,不算惊艳,但对数据质量参差不齐的现场传感器值来说,已经具备实用价值。
训练过程中我还会绘制loss曲线和预测对比图。这里不展开画图的代码,但强烈建议你保留。模型训完之后再看一眼预测曲线,如果有波峰波谷对不齐,说明模型没有真正学到时间模式,多半是数据构造或者超参有问题。
5. 给模型套上GUI外套:tkinter预测界面的完整实现
模型训练好之后,拿去交付给不懂代码的运行人员,总不可能让人家去跑Python脚本调API。我在项目里用tkinter做了一个非常轻量的图形界面,不用装额外的依赖,双击就能跑,功能也挺完整。
5.1 界面布局与交互逻辑
界面逻辑设计如下:
- 用户可以选择CSV数据文件导入最近若干时刻的特征数据
- 界面上直接显示8个输入特征的数值
- 点击"开始预测"按钮,加载训练好的模型和scaler,输出预测值
- 如果用户没有新数据,也提供"使用测试集示例"给你演示用
import tkinter as tk from tkinter import filedialog, messagebox import torch import pandas as pd import joblib class PredictionGUI: def __init__(self, model_path, scaler_path): self.window = tk.Tk() self.window.title("设备能耗预测系统 - Transformer Encoder") self.device = torch.device('cuda' if torch.cuda.is_available() else 'cpu') self.model = TransformerEncoderRegressor(input_dim=8, d_model=64, nhead=4, num_layers=3) self.model.load_state_dict(torch.load(model_path, map_location=self.device)) self.model.to(self.device) self.model.eval() self.feature_scaler = joblib.load(scaler_path) self.entries = {} self._build_ui() self.window.mainloop()框架搭好之后,界面上就是8个标签+输入框,对应8个特征字段。为了演示效果,我把特征命名成:环境温度、设备温度、振动幅度、转速、电流、电压、气阀开度、运行时长。这样运行人员一眼能看懂每格填什么。
5.2 核心预测逻辑实现
预测逻辑的核心在于:界面拿到用户输入的8个值后,这只是一个时刻的特征。但模型需要的是48个时刻、8个特征的矩阵。这个数据从哪里来?
我提供两种途径:
- 如果导入的是CSV文件,且包含最近的历史数据,就自动取出最近48条
- 如果用户只想快速测试,就在代码里写一个简单逻辑:用当前输入的8个值复制出48个时间步,作为一个"静态序列"去预测。这样预测结果反映的是"假设设备保持当前状态往后运行"的能耗水平
下面是第二种方式的代码:
def predict_from_current_state(self): try: features = [] for name in self.feature_names: features.append(float(self.entries[name].get())) features = np.array([features]) features_std = self.feature_scaler.transform(features) # 复制48个时间步 sequence = np.repeat(features_std, 48, axis=0).reshape(1, 48, 8) with torch.no_grad(): X = torch.FloatTensor(sequence).to(self.device) pred = self.model(X).cpu().numpy().reshape(-1, 1) # 反标准化 pred_real = self.target_scaler.inverse_transform(pred)[0, 0] self.result_label.config(text=f"预测能耗值: {pred_real:.2f} kW·h") except Exception as e: messagebox.showerror("输入错误", f"请检查输入格式: {str(e)}")如果读取CSV真实历史,逻辑稍微复杂一点:
def predict_from_csv(self): path = filedialog.askopenfilename(filetypes=[("CSV files", "*.csv")]) if not path: return df = pd.read_csv(path) if len(df) < 48: messagebox.showwarning("数据不足", "至少需要48条连续记录") return recent = df.iloc[-48:][self.feature_names].values recent_std = self.feature_scaler.transform(recent) sequence = recent_std.reshape(1, 48, 8) # 后续与上面一致注意:
feature_scaler在GUI里也要做和训练时一样的操作——用同一个scaler去transform。所以训练完成后一定要把两个scaler保存下来,常见做法是joblib.dump。这个细节做完就不容易遗忘,但很多人训练完只存模型权重,不存scaler,导致推理时特征数值不对,预测结果完全跑偏。</>
5.3 用PyInstaller打包成exe
让运行人员双击运行,不能总是让他们开个Python解释器。用PyInstaller打包:
pip install pyinstaller pyinstaller -F -w --add-data "model.pth;." --add-data "feature_scaler.pkl;." transformer_gui.py-F生成单文件exe,-w表示不显示命令行窗口。如果你的资源文件路径写的是相对路径,打包后可能找不到模型文件。我在代码里用了resource_path函数去适配打包环境:
import sys import os def resource_path(relative_path): base_path = getattr(sys, '_MEIPASS', os.path.abspath(".")) return os.path.join(base_path, relative_path)然后在加载模型和scaler的地方都通过resource_path取路径。这是PyInstaller打包最常见的一个坑,不加这个函数,你在开发环境跑得好好的,打包完一开机直接报找不到文件。
6. 训练中的稳定性和性能优化:一些实战经验
这部分内容是我跑这个项目几十轮实验攒下来的体会,常规博客不太会讲得这么细,但对结果影响却很大。
6.1 随机种子固定:跨平台可复现的关键
训练深度学习模型如果不固定随机种子,同一份代码两次训练结果可能差出好几个百分点。尤其Transformer这类模型,初始化的微小差异在多层叠加后被放大得厉害。
def set_seed(seed=42): import random random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) torch.backends.cudnn.deterministic = True torch.backends.cudnn.benchmark = False把这段放训练脚本最开头,每次跑出来的结果基本一致。尤其在与别人论文对比或内部评审时,可复现性非常重要。
6.2 梯度裁剪的必要性
Transformer的训练中偶尔会出现loss突然暴涨的情况,这多半是某个样本触发了过大的梯度。我在每次反向传播之后加了一句梯度裁剪:
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)这个操作等价于给所有参数梯度模长设了一个上限:如果超过1.0就按比例缩回去。它的作用是防止单次更新的步长过大,从而稳定训练、防止loss发散到NaN。我的经验是,加上这一句之后,用LogCosh加Transformer的组合几乎没有再遇到过训练崩掉的情况。
6.3 早停机制:没有理由让模型跑满全部epoch
训练曲线在某个epoch之后就不再下降,甚至验证集loss开始回升,说明模型开始过拟合了。与其靠肉眼盯着训练输出卡点,不如写一个早停逻辑自动停:
best_val_loss = float('inf') patience = 15 counter = 0 for epoch in range(num_epochs): train_loss = train_one_epoch(...) val_loss = validate(...) if val_loss < best_val_loss: best_val_loss = val_loss torch.save(model.state_dict(), 'best_model.pth') counter = 0 else: counter += 1 if counter >= patience: print(f"Early stopping at epoch {epoch}") break这里我把patience设定为15。如果连续15个epoch验证损失都没刷新纪录,我就认为继续训练意义不大。保存下来的是验证损失最小的那一次权重,而不是最后一轮的权重,这个细节非常重要。"最后一轮权重"往往不是"表现最好的权重",因为验证集上波动会导致最后几轮不一定落在最优点上。
6.4 隐藏维度与头数的组合实验
我在项目中做过一组对比实验,直接放数据给大家参考:
| d_model | nhead | num_layers | R2 | MAE |
|---|---|---|---|---|
| 32 | 2 | 2 | 0.872 | 2.31 |
| 64 | 4 | 3 | 0.912 | 1.68 |
| 64 | 8 | 3 | 0.905 | 1.74 |
| 128 | 4 | 3 | 0.907 | 1.70 |
| 128 | 8 | 4 | 0.908 | 1.69 |
结论很有意思:大模型不一定更好。从64维4头3层再往上加容量,性能几乎不涨,甚至略有下降。这个项目的核心瓶颈不在模型容量,而在数据本身的噪声水平。更大更多的参数只会让模型去拟合噪声,而不会提升对真实模式的表征能力。
所以如果你在做类似任务,不要盲目上大模型、长序列。先用64维4头3层,如果欠拟合再一层层往上加。加参量的优先级建议是:num_layers>d_model>nhead。层数直接决定模型能捕捉的抽象层次,隐藏维度决定了单层表达的宽度,头数则影响注意力对不同子空间的切分程度。一般来说头数选择和d_model之间需要满足整除关系,不需要很大。
6.5 训练过程中的数据标准化细节
标准化之前要特别注意:设备传感器数据里可能存在缺失值和异常值。我一开始没做处理就标准化,结果均值和方差被几个异常值拉偏,模型的预测在正常范围之外波动得很厉害。
建议在构造序列之前的预处理阶段就做好两步:
- 用插值法或前向填充补齐缺失值
- 用3倍标准差法或者百分位法剔除异常点,替换成上下边界值
对于时间序列数据,缺失值我推荐用ffill前向填充,因为设备运行中相邻时刻数值变化幅度有限,前向填充比线性插值更贴近真实物理过程。
异常值方面,能耗数据里偶尔会有设备启停瞬间的尖峰,这类数据如果是正常操作造成的,不应该当作噪声粗暴剔除。你需要结合业务场景判断。我在项目里使用了一个保守的处理:把超过99.9分位数的点替换为99.9分位数,防止极端数据把scaler参数拉偏,但不改变数据总量。
lower_perc = data[target_col].quantile(0.001) upper_perc = data[target_col].quantile(0.999) data[target_col] = data[target_col].clip(lower_perc, upper_perc)如果你的现场数据里确实存在需要剔除的传感器坏值,建议使用median填充而不是mean,因为中位数对异常值更鲁棒。做完这一步再进StandardScaler,后面的效果会明显不一样。
7. 完整训练脚本结构梳理
如果你想直接把这个项目当模板用,这里给出一个完整的文件结构和训练流程。我习惯把所有东西拆得清清楚楚,调试和维护都省心:
proj/ ├── data/ # 原始数据存放与处理脚本输出 │ └── sensor_data.csv ├── models/ # 保存训练好的模型和scaler │ ├── best_model.pth │ ├── feature_scaler.pkl │ └── target_scaler.pkl ├── utils/ │ ├── data_preprocess.py # 数据读取、标准化、序列构造 │ ├── model.py # TransformerEncoderRegressor及位置编码 │ ├── loss.py # LogCoshLoss │ └── train.py # 训练与评估全流程 ├── gui/ │ └── predictor_gui.py # tkinter界面 └── main.py # 一键训练入口细节决定成败。数据预处理脚本、模型定义、训练逻辑拆开写,每个模块还能单独被GUI复用,避免在GUI脚本里再次复制大量数据处理的代码。我这里GUI里直接调用model.py和data_preprocess.py封装的函数,维护起来非常顺。
迭代训练的时候,我的main.py入口大概是这样的:
from utils.data_preprocess import load_and_split, build_dataloaders from utils.model import TransformerEncoderRegressor from utils.loss import LogCoshLoss from utils.train import train_model, evaluate_model # 1. 加载数据,生成训练/测试集 X_train, X_test, y_train, y_test = load_and_split('data/sensor_data.csv') # 2. 构造数据加载器 train_loader, test_loader = build_dataloaders(...) # 3. 初始化模型 model = TransformerEncoderRegressor(input_dim=8, d_model=64, nhead=4, num_layers=3) # 4. 训练 train_model(model, train_loader, test_loader, epochs=100) # 5. 评估 r2, mae, rmse, preds, trues = evaluate_model(model, test_loader)训练过程带着验证集做早停,训练完成自动把最优模型和两个scaler存下来,CPU大概需要半小时到一小时,GPU可以几分钟搞定。之后GUI加载这些产物,完事。
8. 收尾:关于这类项目的一点心得体会
做这个项目最大的感受是:Transformer在回归预测任务上的价值,经常被严重低估。不只是在NLP和大语言模型这些热门场景,只要数据带有时间依赖特征,Encoder的全局注意力机制对比RNN就有天然优势。但它的好是有条件的——数据要有足够的时间连续性,样本量不能太小,特征维度要有信息量。如果这三条不满足,你不如回头用MLP或者线性回归。
还有一个小建议:训练时不要只盯着最后的R2,多看看预测值的时间序列曲线。R2是一个整体统计量,它对局部动态模式的拟合效果不敏感。预测曲线如果整体趋势正确但每个波峰波谷都滞后一拍,R2可能也有0.85以上,但实际使用中那种滞后是无法接受的。我因为在项目里多看了几眼预测曲线,才发现数据构造中有一个序列对齐偏移的bug,如果只看R2根本发现不了。
如果后续你想把这个项目再往前推一步,不妨试试多步预测(比如一次预测未来3个时刻的能耗值),或者把注意力权重可视化出来,看看模型每步预测的时候重点在看哪些历史时刻。这两件事对理解模型行为和理解业务数据都有很大帮助。先说到这里,接下来你可以拿自己的数据试试,遇到问题欢迎一起讨论。