PyTorch实现Transformer多维时间序列分类完整指南

简介：这套基于PyTorch与Transformer的多维时间序列分类项目源码，适合具备一定深度学习基础、希望将注意力机制应用于时序数据建模的开发者学习参考。内容围绕Gated Transformer结构展开，覆盖数据处理、模型训练、热力图与特征图可视化、聚类分析等核心环节，难度适中，代码经本地编译验证可运行，并附有助教审定的说明文档，便于对照理解实现细节。压缩包共65个文件，约13.48MB，以Python脚本为主体（19个py），辅以编译生成的pyc文件、运行结果图像（jpg/png）、模型权重备份（zbak）及字体与说明文档，目录按模块分区，方便按数据处理、模型构建、结果可视化等阶段查阅。目前已有30人学习下载，适合入门Transformer时序分类任务，或作为课程设计与科研实验的参考实现。

1. 多维时间序列分类：为什么选 Transformer 而不是 LSTM/CNN

多维时间序列分类任务里，最常被问到的选型问题是：该用 LSTM 还是 Transformer，以及 PyTorch 源码怎么组织才能复现。真实场景往往是一台设备上四路传感器同时采样温度、振动、电流、压力，要把最近一段数据判成正常、磨损还是故障。LSTM 长序列梯度容易衰减，1D-CNN 感受野有限，跨时间步交互要靠堆层数；Transformer 的自注意力让任意两个时间步直接连通，多变量通道也能独立建模，配合 PyTorch 现成的 TransformerEncoder，几百行源码就能搭完从数据到分类评估的完整流程。

这篇文章面向两类读者：刚接触深度学习时间序列分类的人能照步骤复制全流程；写过不少分类模型的熟手，重点看窗口参数、位置编码与可复现性这三个易忽略的环节。

2. 多维时间序列的窗口化与张量化：喂给 Transformer 前先定好形状

2.1 滑窗的目标与窗口参数怎么定

原始采样流是一条很长的时间序列，分类通常要回答的是"过去这一段时间属于哪个状态"。所以第一步是用固定长度的窗口把流切成样本。窗口长度取多少，取决于语义粒度：振动信号周期约 0.1 秒时，窗口至少要装下 5～10 个周期，模型才能看到重复冲击；温度是慢变量，窗口太短会让模型只看到本底噪声；人工标注的事件样本，则按事件平均长度取窗。

滑窗允许重叠。步长小于窗口长度时，相邻窗口有重叠，既增加了样本量，也平滑了标签边界；步长等于窗口长度时样本互不重叠，适合事件本身离散且分布均匀的数据。直接给"窗口越大越好"是常见误解——窗口加长会线性增加训练成本，却不一定线性带来准确率提升。

2.2 用 numpy 一次切出所有窗口

按 Python for 循环逐窗口拷贝多维数组，样本到几万条时速度就会明显变慢。常见做法是用 numpy 的 stride_tricks 生成零拷贝窗口视图，再一次性 copy 成独立内存：

import numpy as np import pandas as pd def sliding_windows(data: np.ndarray, window_size: int, stride: int): """把 (T, C) 的多维时间序列切成 (N, window_size, C)。""" t_len = data.shape[0] n_windows = 1 + (t_len - window_size) // stride from numpy.lib.stride_tricks import as_strided shape = (n_windows, window_size, data.shape[1]) strides = (data.strides[0] * stride, data.strides[0], data.strides[1]) return as_strided(data, shape=shape, strides=strides).copy() df = pd.read_csv("sensor.csv") data = df.iloc[:, :4].to_numpy(dtype=np.float32) windows = sliding_windows(data, window_size=128, stride=64) print(windows.shape) # (N, 128, 4)

strides 三个分量的含义要解释清楚：第一个分量是"窗口之间"移动的字节数，等于原始时间步长乘以 stride，算错会让窗口起点错位；第二个分量是窗口内相邻时间步的偏移，即原始数组的行步长；第三个分量是通道维步长。as_strided 产生的视图与原始数组共享内存，最后调用 .copy() 把数据真正复制出来，避免后续归一化等原地操作污染源数组。这里其实已经涉及 pytorch 张量基础的一个对齐点：后续模型接受的数据形状是 (batch, seq_len, feature_dim)，与 NLP 中 (batch, seq_len, embedding_dim) 完全一致，feature_dim 就是这里的通道数。

这段代码没有做边界校验。如果 t_len 小于 window_size，n_windows 会是 0 或负数，说明这批数据本身就不该参与训练；遇到这种样本直接丢弃，不要用 pad 硬凑。

2.3 按通道做 Z-score 归一化

通道间量纲差异很大时，电流可能是几十安，振动幅值只有 0.01，直接喂网络会让大数值通道主导注意力权重。常见做法是按通道独立做 Z-score，统计量只能来自训练集：

def fit_scaler(train_data: np.ndarray): mean = train_data.mean(axis=0, keepdims=True) std = train_data.std(axis=0, keepdims=True) + 1e-8 return mean, std mean, std = fit_scaler(x_train_raw) x_train = (x_train_raw - mean) / std x_val = (x_val_raw - mean) / std

验证集和测试集都用训练集统计量来转换，不能在各自集合上重新 fit。另一个容易踩的点：不要在切窗之前直接对整条序列做归一化，除非你确认这条序列是单个稳定过程的采样；否则某一段的离群值会把其他窗口全部压缩到接近 0。推荐顺序是"划分样本 → 求训练统计量 → 转换全部分段"。

2.4 类别不平衡与分层划分

故障样本天然稀缺是这类任务的常态。切完窗口之后先用 bincount 数一数每个类的窗口数，再做分层划分，保证训练、验证、测试的类别比例一致：

from sklearn.model_selection import train_test_split def split_by_class(X, y, val_ratio=0.15, test_ratio=0.15): X_train_val, X_test, y_train_val, y_test = train_test_split( X, y, test_size=test_ratio, stratify=y, random_state=42) val_size_in_train = val_ratio / (1 - test_ratio) X_train, X_val, y_train, y_val = train_test_split( X_train_val, y_train_val, test_size=val_size_in_train, stratify=y_train_val, random_state=42) return X_train, X_val, X_test, y_train, y_val, y_test

分两次划分而不是一次三路 split，是为了让训练集和验证集都保持分层。如果某些类别在训练集里样本数为 0，问题不在损失函数，而在数据采集或分层策略本身。

如果相邻窗口来自同一条采样序列，它们高度相关，直接随机 split 会让验证指标虚高。工业场景更严谨的做法是按设备记录 ID 做 GroupShuffleSplit，保证同一条记录的所有窗口只出现在一个集合里。

提示：窗口重叠比例越高，训练集与验证集之间的信息泄漏越严重；重叠率超过 50% 时建议强制按记录 ID 分组划分。

3. 基于 PyTorch 的 Transformer 分类器：从 Encoder 到分类头的完整实现

3.1 时间序列怎么映射成 Token

讲 transformer 架构及其工作原理时，核心是 token 的抽象。在时间序列分类里，最自然的映射是：把每个时间步当作一个 token，该时刻所有通道的采样值就是 token 的原始特征。通道数是 4，token 的输入维度就是 4；Transformer 内部多头注意力的统一维度叫 d_model，输入和 d_model 不一致时加一个线性投影对齐。

如果采样噪声大，也可以先在时间维上做一层短卷积再投影，既平滑又降采样，类似 Swin Transformer 里 patch embedding 的思路。不过工业多维时间序列的窗口长度通常不超过 1024，直接线性投影往往就够，先用最简配置跑通再优化不迟。

为什么不用 Transformer 的 Decoder？分类任务输出的是窗口级单一标签，不是逐时间步生成序列；Encoder 输出的每个位置表示已经编码了全局语义，配合池化即可。Decoder 的自回归结构在这里只会引入不必要的顺序依赖和训练不稳定因素。

3.2 位置编码取可学习的还是正弦固定编码

自注意力本身对位置不敏感，把时间步打乱重排，注意力结果在集合意义上是等价的，所以位置编码直接影响分类性能。两种主流做法：正弦位置编码由不同频率的 sin/cos 组成，优势是能外推到比训练时更长的序列；可学习位置编码是一张可训练的参数表，定长任务上训练更稳定，也少一个调参维度。

时间序列分类的窗口通常是固定的，不需要外推能力，所以一般项目里默认用可学习编码。参数表长度必须大于等于最大 seq_len，如果数据增强会变长输入，编码表更要覆盖到最长可能值：

import torch import torch.nn as nn class LearnablePositionalEncoding(nn.Module): def __init__(self, max_len: int, d_model: int): super().__init__() self.pe = nn.Parameter(torch.zeros(1, max_len, d_model)) nn.init.trunc_normal_(self.pe, std=0.02) def forward(self, x): return x + self.pe[:, :x.size(1), :]

初始化为什么用 trunc_normal_ 而不是全零？全零位置编码在训练最初若干 batch 内提供不了任何位置先验，模型只能靠数据硬学；std=0.02 的截断正态既给了初始位置结构，又不会压过通道特征的数量级。

3.3 核心模型：不需要手写 Q/K/V

PyTorch 的 nn.TransformerEncoderLayer 已经把多头自注意力、残差、LayerNorm、FFN 全部封装好，源码里不需要手写 Q/K/V 矩阵相乘，重点是把输入输出维度和池化方式组织对。完整分类器如下，需要把 3.2 的 LearnablePositionalEncoding 一并复制：

import torch import torch.nn as nn class TSClassifier(nn.Module): def __init__(self, n_channels: int, seq_len: int, d_model: int = 64, nhead: int = 4, num_layers: int = 3, dim_feedforward: int = 128, num_classes: int = 3, dropout: float = 0.1): super().__init__() self.input_proj = nn.Sequential( nn.Linear(n_channels, d_model), nn.GELU(), nn.Dropout(dropout), ) self.pos_enc = LearnablePositionalEncoding(seq_len, d_model) encoder_layer = nn.TransformerEncoderLayer( d_model=d_model, nhead=nhead, dim_feedforward=dim_feedforward, dropout=dropout, batch_first=True, activation="gelu") self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers) self.pool = nn.AdaptiveAvgPool1d(1) self.cls_head = nn.Sequential( nn.Linear(d_model, d_model // 2), nn.GELU(), nn.Dropout(dropout), nn.Linear(d_model // 2, num_classes), ) def forward(self, x): x = self.input_proj(x) x = self.pos_enc(x) x = self.encoder(x) # (batch, seq_len, d_model) x = x.transpose(1, 2) # (batch, d_model, seq_len) x = self.pool(x).squeeze(-1) # 平均池化 -> (batch, d_model) return self.cls_head(x)

模型里几个核心参数的调法：d_model 取 64 还是 128，取决于数据量和通道数，样本量小就 32；nhead 必须整除 d_model，64/4、128/8 是稳定组合；num_layers 在时间序列任务上取 2～4 层，超过 6 层无预训练时基本必然过拟合；dim_feedforward 取 d_model 的 2～4 倍。

dropout 需要结合数据量：训练样本上万时 0.1 足够，标签噪声大时调到 0.3 观察验证集。batch_first=True 这个参数值得强调：它让输入形状直接是 (batch, seq_len, d_model)，和 DataLoader 出来的张量顺序一致，不必在 forward 里做转置。如果用的是旧版本 PyTorch，默认 batch 在第二维，需要手动转成 (seq_len, batch, d_model)，接口没有新版本干净。

3.4 池化方式与分类头设计：不是简单取最后一步

Encoder 输出的是每个时间步的 d_model 维表示。有人习惯性取最后一个时间步做分类，这在时间序列任务里是错误的：没有语言模型式的 next-token 预测约束，序列末尾位置的表示并不保证携带全局信息。常见池化方式对比：

平均池化把注意力混合后的全部时间步一视同仁，分类头拿到的是整个窗口的综合表示，和 self-attention 的全局聚合目标一致。cls_token 在 NLP 里效果好，但在多维时间序列上很少带来超过平均池化的提升，还多出位置、长度上的特殊处理，所以一般项目默认用 AdaptiveAvgPool1d。

4. 训练循环与分类评估：Transformer 的损失函数、学习率与指标聚合

4.1 DataLoader 与训练循环模板

窗口样本装进 TensorDataset 就能直接构造 DataLoader。无论你是用 anaconda 配置 pytorch 环境，还是用 pip 直接安装 PyTorch，下面的代码都通用，注意把 torch、numpy、sklearn 版本锁在 requirements 里：

import torch import torch.nn as nn import numpy as np from torch.utils.data import TensorDataset, DataLoader def make_loader(X, y, batch_size=64, shuffle=True): X = torch.from_numpy(X).float() y = torch.from_numpy(y).long() return DataLoader(TensorDataset(X, y), batch_size=batch_size, shuffle=shuffle) train_loader = make_loader(X_train, y_train, batch_size=128) val_loader = make_loader(X_val, y_val, batch_size=256, shuffle=False) model = TSClassifier( n_channels=X_train.shape[-1], seq_len=X_train.shape[1], num_classes=len(np.unique(y_train)), ) optimizer = torch.optim.AdamW(model.parameters(), lr=1e-3, weight_decay=1e-4) criterion = nn.CrossEntropyLoss() scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=30)

训练循环里有一个从 LSTM 时间序列预测转过来的同学容易漏掉的点：梯度裁剪。Transformer 叠层多，反向梯度范数波动大，单步异常值可能直接把前期学到的分布冲掉。max_norm=1.0 是多数分类任务的安全值：

def train_one_epoch(model, loader, optimizer, criterion, device): model.train() total_loss, correct, total = 0.0, 0, 0 for xb, yb in loader: xb, yb = xb.to(device), yb.to(device) optimizer.zero_grad() logits = model(xb) loss = criterion(logits, yb) loss.backward() nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() total_loss += loss.item() * xb.size(0) correct += (logits.argmax(1) == yb).sum().item() total += xb.size(0) return total_loss / total, correct / total

注意：clip_grad_norm_ 的 max_norm 针对全部参数拼接后的整体范数。如果训练 loss 降得快但验证指标不动，先检查裁剪阈值是不是设得过大，不要第一反应就怀疑模型结构。

学习率调度上，CosineAnnealingLR 配 AdamW 是 Transformer 类模型的主流组合；小规模样本可以不加热身，数据量大或 d_model 超过 256 时，建议按 sqrt(d_model) 缩小基准学习率，或在最初 5% 的 iter 里线性 warmup。

4.2 损失函数：类别不平衡时的配置

默认 CrossEntropyLoss 对每个样本一视同仁。故障类只占 5% 时，模型学到"全预测正常"就已经有 95% 准确率。先看类别分布再决定是否加权：

counts = np.bincount(y_train) weights = 1.0 / (counts / counts.sum()) criterion = nn.CrossEntropyLoss(weight=torch.tensor(weights, dtype=torch.float))

weight 的含义是低频类单样本错分时施加更大惩罚。也有人用 Focal Loss 做难例挖掘，但工业分类类别数一般不超过 10，weighted CE 调整成本最低，先跑通再换。做基线对比时，有人会拿 xgboost 二分类模型先在手工特征上跑一版，树模型对 tabular 特征友好，但原始多维时间序列的特征工程成本高，Transformer 的端到端优势恰在这里。

4.3 分类评估指标：不能只看 accuracy

多分类评估里 accuracy 在类别不均衡时有迷惑性。正确姿势是每个类的 precision、recall、F1 都看，再加 macro-F1 与混淆矩阵：

from sklearn.metrics import classification_report, confusion_matrix, f1_score def evaluate(model, loader, device): model.eval() y_true, y_pred = [], [] with torch.no_grad(): for xb, yb in loader: logits = model(xb.to(device)) y_pred.extend(logits.argmax(1).cpu().numpy()) y_true.extend(yb.cpu().numpy()) print(classification_report(y_true, y_pred, digits=3)) print(confusion_matrix(y_true, y_pred)) return y_true, y_pred

指标选择要跟着业务目标走：

4.4 早停与最佳模型保存

Transformer 参数多，训练后期验证指标经常是涨一段然后崩。按验证集 macro-F1 做早停，保存历史最佳权重而不是最后一轮权重：

best_score, bad_epochs = 0.0, 0 for epoch in range(30): _, _ = train_one_epoch(model, train_loader, optimizer, criterion, device) y_true, y_pred = evaluate(model, val_loader, device) score = f1_score(y_true, y_pred, average="macro") if score > best_score: best_score = score torch.save(model.state_dict(), "best_model.pt") bad_epochs = 0 else: bad_epochs += 1 if bad_epochs >= 5: break scheduler.step()

patience 取 5 轮是起步值，验证集噪声大就放宽到 8～10。正式工程里 evaluate 里的 print 要换成 logging，否则每轮刷屏会影响日志可读性。早停同时承担了防止过拟合和自动确定 epoch 数的职责，省掉它之后手动调 epoch 反而更不可控。

5. 把 PyTorch 源码整理成可复现工程：seed、模型状态与文档边界

模型能跑通只是第一步。交付一份"源码及文档说明"，真正要提交的是别人拉下来能原样复现的实验环境与结果，最容易翻车的两个技术点是随机种子和模型保存格式，最容易被低估的是文档里的数据接口边界。

5.1 全局锁定随机种子

窗口切分、模型初始化、DataLoader 的 shuffle 都引入随机性。复现的第一步是在训练一开始就把所有随机源锁死：

import random import numpy as np import torch def seed_everything(seed=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 = False

cudnn.deterministic 会牺牲少量性能换取算子稳定；cudnn.benchmark 关掉是为了防止 PyTorch 按输入尺寸动态换算法导致实验间微小抖动。两个开关都必须在构建模型之前设置，否则初始化顺序不同照样复现不出来。

5.2 序列化配置而不是只存 state_dict

4.4 里的 torch.save(state_dict) 适合快速验证，但换人接手时会卡在"当时 d_model 到底设的多少"。可复现工程的推荐做法是把构造参数与指标一起存：

SAVE = { "config": {"n_channels": 4, "seq_len": 128, "d_model": 64, "nhead": 4, "num_layers": 3, "num_classes": 3}, "state_dict": model.state_dict(), "macro_f1": best_score, } torch.save(SAVE, "checkpoints/ts_classifier.pt")

推理时先按 config 重建模型再加载权重，训练超参不丢失，指标也能追溯。

5.3 数据格式约定是文档的边界

README 至少讲清四件事：数据文件格式（每列是什么，时间戳放哪列）、窗口划分方式（window_size / stride / 是否重叠）、训练与评估的启动命令、依赖版本列表。其中数据格式约定直接决定源码能否被别人复用——别人拿到的数据第一列若是时间戳，而源码假设第一列是通道，整个预处理链都会错位。

文档里放一张窗口划分示意图，画清楚原始序列、窗口、步长与标签对齐关系，比写十行文字都直观，这是提升这份源码实际使用率最快的投入。

本文还有配套的精品资源，点击获取