PyTorch+Transformer多维时间序列分类工程实战与源码详解

简介面向多维时间序列分类入门与应用人群这份基于PyTorch与Transformer的源码包提供了从数据处理到模型训练评估的完整实现。资源共65个文件、压缩包13.48MB主要包含Python源码、文档说明与结果图像其中.py与.pyc文件支撑代码运行.md/.txt提供文档.jpg/.png用于展示模型结构与可视化结果另有字体和备份文件作为辅助。目前已有30人学习内容经助教审定难度适中可直接运行验证。实际代码涵盖数据预处理、Gated Transformer搭建、heatmap与注意力分析、聚类和特征提取可视化等环节配合README说明便于读者按图索骥在较短时间内掌握多维时间序列分类的工程化实现。1. 多维时间序列分类为什么这个任务值得单独建一套源码时间序列分类不像图像分类那样有天然的“像素”结构也不像文本那样有明确的词边界。多维时间序列更难因为通道之间既存在空间相关又存在时间依赖单靠增大 LSTM 的隐层维度往往只能抓到短程模式长程依赖丢失得很厉害。Transformer 的 self-attention 机制正好擅长建模任意两个时间步之间的依赖而且在多通道场景下还可以让注意力同时跨通道与跨时间步计算这正是多维时间序列分类任务近年转向 Transformer 架构的核心原因。这份源码要解决的不是“怎么把手写数字分类对”而是更工程化的东西拿到一个多变量传感器数据集怎么切窗、归一化、把 Patch Embedding 与位置编码拼起来训练出一个能在测试集上稳定复现的分类模型并且把参数、评估口径和可复现性都锁住。适读人群是已经跑通过一次 PyTorch 基础框架、想跳到序列任务上做工程落地的开发者。以下方案就是围绕这个目标搭出来的一条完整落地路径。2. 多维时序类任务的输入结构与预处理管线2.1 先明确输入张量的形状与语义多维时间序列分类的原始数据通常不是 Tensor而是一张表格每一行是一个采样时刻每一列是一个传感器通道。以 UCI 的 Human Activity Recognition 数据集为例一个窗口通常是 128 个时间步每个时间步有 9 个加速度计通道。PyTorch 模型接受的输入形状需要约定成(batch_size, seq_len, num_channels)这是 Transformer 编码器的标准输入布局。不要在这个问题上凭直觉乱改。很多人直接把数据排列成(batch_size, num_channels, seq_len)然后丢给 Transformer结果注意力在通道维上计算语义就全偏了。Transformer 编码器内部是按seq_len维度做 attention 的所以输入必须把时间步放在中间维。源码里建议在Dataset的__getitem__中直接统一返回(seq_len, channels)的张量DataLoader 自动堆叠出(batch, seq_len, channels)。关键参数表输入构造阶段需要明确的三个量参数含义建议seq_len单个样本包含的时间步数由下游任务决定常用 128 或 256num_channels传感器/特征通道数等于原始特征列数stride滑动窗口步长重叠采样时常用seq_len // 2这三者直接决定数据集的大小和模型学到的模式粒度。窗口太大样本数骤减模型容易过拟合窗口太小高频噪声会盖过长程模式。步长用seq_len // 2做重叠采样是信号处理里最常见的做法既不丢信息又不至于让相邻样本完全重复。2.2 用 PyTorch Dataset 封装切窗逻辑下面这段代码直接复制就能跑通数据侧的核心逻辑import torch from torch.utils.data import Dataset class MultivariateTSDataset(Dataset): def __init__(self, data: torch.Tensor, labels: torch.Tensor, seq_len: int 128, stride: int 64): data: (total_len, num_channels) 的浮点张量 labels: (total_len, ) 对应每个时刻的标签窗口内取最后一个时刻的标签 seq_len: 窗口长度 stride: 滑动步长 self.seq_len seq_len self.stride stride self.data data self.labels labels def __len__(self): return (self.data.shape[0] - self.seq_len) // self.stride 1 def __getitem__(self, idx): start idx * self.stride end start self.seq_len x self.data[start:end] # (seq_len, channels) y self.labels[end - 1] # 窗口末时刻的标签 return x, y这段代码的要点有两个第一个是标签对齐方式线割类任务里labels[end - 1]这种“取窗口末端标签”的做法比“取多数投票”更稳因为状态切换通常发生在窗口后半段第二个是索引计算__len__里的减法和1必须配套否则最后一个窗口会越界。很多免费 python 源码大全里的切窗实现都忽略了尾部窗口的丢弃问题这里显式暴露出来。如果原始 CSV 有缺失值切窗之前做一次线性插值是成本最低的处理不要直接把 NaN 送进模型Transformer 对 NaN 的敏感程度比 CNN 高一条 NaN 就足以让 attention 权重整体偏移。2.3 归一化必须在切窗之后按窗口统计量做归一化是个容易让人翻车的地方。按整个训练集做全局均值和方差归一化会让早期窗口的分布被后期窗口拉偏按每个窗口单独做归一化又会破坏窗口之间的相对幅值关系。多维时间序列分类的标准做法是按通道分别计算训练集的均值和方差然后对每个窗口使用同一组统计量做标准化。def fit_standardize(data: torch.Tensor, eps: float 1e-6): # data: (total_len, channels) mean data.mean(dim0, keepdimTrue) std data.std(dim0, keepdimTrue) return (data - mean) / (std eps), mean, std这段代码里keepdimTrue很重要它保证了广播运算时不会丢通道维度。eps是给方差为 0 的通道兜底用的比如某个压力传感器在整段采集中没有任何波动没有eps会出现除零。验证集和测试集必须使用训练集保存下来的mean和std不允许重新 fit这一点建议在源码里直接写进validate函数避免后期误用。3. 用 PyTorch 实现 Transformer 分类模型3.1 Patch Embedding 还是逐点映射图像 Transformer 的做法是把一个 patch 展平后过线性层。时间序列分类可以照搬这个思路把连续若干个时间步组成的 patch 映射成一个 token。为什么比逐时刻映射好因为单步点的信息量太少self-attention 在逐点模式下很容易把注意力分配给噪声而 patch 相当于做了局部平滑等价于一种可学习的下采样。常见做法是设置patch_len把长度为seq_len的序列切成num_patches seq_len // patch_len个 patch每个 patch 内所有通道拼接后过一个 Linear 层。下面这段代码对应源码中的 PatchEmbedding 部分import torch.nn as nn class PatchEmbedding(nn.Module): 输入 x: (batch, seq_len, channels) 输出: (batch, num_patches, d_model) def __init__(self, seq_len: int, in_channels: int, d_model: int, patch_len: int): super().__init__() assert seq_len % patch_len 0, seq_len 必须能被 patch_len 整除 self.patch_len patch_len self.proj nn.Linear(in_channels * patch_len, d_model) def forward(self, x): batch, seq_len, channels x.shape num_patches seq_len // self.patch_len # (batch, num_patches, patch_len, channels) x x.reshape(batch, num_patches, self.patch_len, channels) # (batch, num_patches, patch_len * channels) x x.reshape(batch, num_patches, -1) return self.proj(x) # (batch, num_patches, d_model)patch_len的取值影响巨大。patch 太小退化成逐点映射patch 太大局部细节丢失。我个人习惯在 8 到 16 之间调HAR 这类传感器数据用 8 效果不错。注意reshape的顺序(batch, num_patches, self.patch_len, channels)这一步如果写错维度会把不同通道的数据混进同一个 patch模型直接学废。3.2 位置编码与 CLS Token 的设计取舍标准 Transformer 分类任务通常加一个[CLS]token最后一层取它的输出过分类头。多维时间序列分类里这个设计不是必须的。原因在于图像 patch 之间没有天然的顺序关系依赖CLS token 可以在最后一层汇总全局信息而时间序列有明确的时序先验在 patch 序列上做全局平均池化或者取最后一个 patch 的表示往往比 CLS token 更稳定。这里给两种方案并在源码中通过参数切换class TimeSeriesTransformer(nn.Module): def __init__(self, d_model64, nhead4, num_layers3, num_classes6, seq_len128, in_channels9, patch_len8, poolingmean): super().__init__() self.patch_embed PatchEmbedding(seq_len, in_channels, d_model, patch_len) num_patches seq_len // patch_len self.pos_embed nn.Parameter(torch.randn(1, num_patches, d_model) * 0.02) encoder_layer nn.TransformerEncoderLayer( d_modeld_model, nheadnhead, batch_firstTrue) self.encoder nn.TransformerEncoder(encoder_layer, num_layersnum_layers) self.pooling pooling self.classifier nn.Linear(d_model, num_classes) def forward(self, x): x self.patch_embed(x) x x self.pos_embed x self.encoder(x) if self.pooling mean: x x.mean(dim1) elif self.pooling last: x x[:, -1, :] return self.classifier(x)位置编码用Parameter是可训练式效果比固定正弦编码好但随机初始化为0.02这个细节是关键初始值太大会在训练早期破坏 patch 特征。batch_firstTrue是 PyTorch 1.13 之后更友好的写法避免 permute 换来换去。如果显存紧张可以等训练收敛后把训练好的位置编码拿去测试不同长度实践中这种行为泛化性很差所以推理输入长度必须保持和训练一致。3.3 TransformerEncoder 的三个必调参数与越界信号d_model、nhead、num_layers这三个参数构成一个相互牵制的系统。nhead必须整除d_model这是多头注意力的硬约束。d_model不宜过大多维时间序列的 patch 数量往往只有 8 到 32 个d_model128在这么短的序列上很容易过拟合d_model64配合nhead4、num_layers3是一个不错的起点。模型训练不收敛时最先检查的往往不是 Transformer 结构而是学习率。Transformer 对学习率的敏感性远高于 LSTM1e-4是安全的起点5e-4就可能在前几百步出现 loss 爆炸。如果你的自定义优化器里用了clip_grad_norm_(model.parameters(), 1.0)loss 仍然乱跳检查位置编码的标准差这是最容易忽略的越界信号。4. 训练循环、类别权重与多分类评估全套代码4.1 训练循环模板日志、早停与模型保存训练部分不算难难在把训练、验证、测试三段逻辑写干净。下面是一套可以直接嵌入任意 PyTorch 基础框架的模板注意它对验证集按 epoch 做早停并保留最优模型import torch from torch.nn import CrossEntropyLoss from torch.optim import AdamW from copy import deepcopy def train_one_epoch(model, loader, optimizer, criterion, device): model.train() total_loss 0.0 for x, y in loader: x, y x.float().to(device), y.long().to(device) optimizer.zero_grad() logits model(x) loss criterion(logits, y) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0) optimizer.step() total_loss loss.item() * x.size(0) return total_loss / len(loader.dataset) def evaluate(model, loader, criterion, device): model.eval() total_loss, correct 0.0, 0 with torch.no_grad(): for x, y in loader: x, y x.float().to(device), y.long().to(device) logits model(x) loss criterion(logits, y) total_loss loss.item() * x.size(0) preds logits.argmax(dim1) correct (preds y).sum().item() return total_loss / len(loader.dataset), correct / len(loader.dataset) def fit(model, train_loader, val_loader, epochs, lr, device, patience10): optimizer AdamW(model.parameters(), lrlr, weight_decay1e-4) scheduler torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_maxepochs) criterion CrossEntropyLoss() best_acc, best_weights, bad_epochs 0.0, None, 0 for epoch in range(1, epochs 1): train_loss train_one_epoch(model, train_loader, optimizer, criterion, device) val_loss, val_acc evaluate(model, val_loader, criterion, device) scheduler.step() print(f[Epoch {epoch:02d}] train_loss{train_loss:.4f} val_loss{val_loss:.4f} val_acc{val_acc:.4f}) if val_acc best_acc: best_acc, best_weights, bad_epochs val_acc, deepcopy(model.state_dict()), 0 else: bad_epochs 1 if bad_epochs patience: print(f早停触发于第 {epoch} 轮) break model.load_state_dict(best_weights) return model对代码逻辑做三点说明。第一clip_grad_norm_的max_norm1.0能有效防止 attention 层梯度爆炸但不要调小到0.1太小会让训练无法收敛第二scheduler.step()放在验证之后、打印之前顺序正确第三deepcopy保存权重而非直接赋值否则后续优化器更新会连带修改已保存的best_weights。4.2 类别不平衡下的损失权重计算多维时间序列分类里类别不平衡是常态比如一个设备 90% 时间处于“正常”状态故障状态只占 5%。直接优化 CrossEntropyLoss 会把模型推向“全预测正常”的极端。训练前用sklearn或者手写代码计算类别权重作为损失函数的weight参数即可。加权后的效果在线割分类中体现特别明显等价于给少数类样本更大的反向传播信号。from collections import Counter def compute_class_weight(labels): counter Counter(labels.tolist()) total sum(counter.values()) num_classes len(counter) weight torch.tensor([ total / (num_classes * counter[i]) for i in range(num_classes) ], dtypetorch.float) return weight / weight.min() # 归一化多数类权重为 1末尾的weight / weight.min()这步是常识之外的技巧。原公式的权重最大值可能达到几十倍直接喂给损失函数会让训练初期不稳定归一化到最小值是 1 之后学习率调度更容易对齐。4.3 分类评估不只是看 Accuracy精确率/召回率/F1 在时间序列分类里必须分开算因为很多工况类别是长尾分布Accuracy 在 97% 的时候少数类的召回率可能只有 20%。分类评估的代码要能输出每个类别的指标和混淆矩阵from sklearn.metrics import classification_report, confusion_matrix import numpy as np def full_report(model, loader, device, class_namesNone): model.eval() all_preds, all_labels [], [] with torch.no_grad(): for x, y in loader: x x.float().to(device) logits model(x) preds logits.argmax(dim-1).cpu().numpy() all_preds.extend(preds.tolist()) all_labels.extend(y.numpy().tolist()) print(classification_report(all_labels, all_preds, target_namesclass_names, digits4)) print(confusion_matrix(all_labels, all_preds))classification_report自带 per-class precision、recall、f1-score 与 support不需要自己循环计算。注意y.numpy()之前要保证y在 CPU。5. 训练稳定性的两个隐藏杀手种子固定与超参对拍训练 Transformer 不像 LSTM 那样“多跑几个 epoch 就自然变好”它对初始化和随机失活比较苛刻。第一个隐藏杀手是随机种子不固定。PyTorch 里torch.manual_seed只管 PyTorch 的随机数DataLoader 内部翻样本顺序用的是 Python 的random模块和 NumPy 的np.random三者必须同时设置import random import numpy as np import torch def set_seed(seed: int 42): 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 Falsecudnn.deterministic True会让卷积和注意力里的某些 cuDNN 算法退化为确定性的但速度稍慢的版本benchmark关闭是为了避免输入尺寸变化导致启动不同算法。很多公开源码不写这两行导致同一份代码在 GPU 上的期望表现不可复现这正是本次源码把种子固定作为独立模块的原因。第二个隐藏杀手是超参对拍。建议固定 patch_len、d_model、nhead 这三个结构参数后只对学习率先做一轮粗扫粗扫经验值是1e-4、5e-5、2e-5三档每档跑到第 10 个 epoch 看验证 loss 的下降速率而不是一次性 full epochs。学习率对 Transformer 的影响远大于隐藏层维度这个观感和 LSTM 训练完全不一样。验证方式上把在训练集上最后几个 epoch 的 loss 和验证集 loss 印在同一行如果前者一直往下掉而后者停滞优先恢复weight_decay而不是加 dropout。给 encoder layer 加dropout0.1是时间序列分类里最不容易出错的配置加了之后通常不需要同步调大num_layers。最后一个实用技巧是把训练好的模型权重和**预处理统计量mean、std**打包保存。很多项目只存权重测试时重新算 mean 和 std导致推理结果和训练报告对不上。保存一个包含权重、mean、std、class_names 的字典既省部署功夫也让整个源码的工作流留下完整闭环。本文还有配套的精品资源点击获取