1. 时间序列预测这件事,为什么CNN也能掺一脚
很多人一听到时间序列预测,脑子里蹦出来的第一个模型一定是LSTM或者Transformer,CNN似乎天生是干图像识别那行的。我最早也是这么想的,直到一次偶然的机会,我需要对一个工业设备的连续运行参数做短期趋势预测,手头没有GPU服务器,只有一台普通的办公电脑,LSTM跑起来慢得让人怀疑人生,Transformer更不用提。后来无意间试了一把一维卷积网络,效果居然出奇地好,而且训练速度快了不止一个量级。
CNN能处理时间序列的原理其实不复杂。图像是二维网格数据,时间序列是一维序列数据,卷积操作的数学本质是局部加权求和——用一个卷积核扫过数据,提取局部模式。放在时间序列的场景里,这个“局部模式”就是某一段历史窗口内的变化规律,比如连续几小时的上升趋势、周期性小波动、突然的尖刺等。多个卷积核叠加,就能从不同尺度捕捉不同的时间依赖模式。相比RNN那种按时间步逐个递推的结构,CNN是并行计算整段序列的,训练效率天然高出一截。
这篇文章打算讲清楚三件事:用PyTorch搭建一维CNN做时间序列预测的完整流程、每一步为什么要这么设计、以及我在实际项目中踩过的坑和最后的调优思路。内容面向有一定PyTorch基础、想找个轻量方案替代RNN做预测的读者,代码可以直接跑通,也欢迎你来挑毛病。
2. 数据准备:滑动窗口、标准化与数据集封装
2.1 时间窗口怎么切
时间序列预测的第一步永远是构建样本,这里面的核心概念是滑动窗口。简单说,我们用过去N个时刻的数据去预测未来M个时刻的值。窗口开多大直接影响模型能“看到”多长的历史信息。
我常用的经验是:先观察原始序列的周期性和自相关性。比如数据有24小时周期,那窗口至少包含一个完整周期,甚至翻倍给到48;如果数据没有明显周期,就通过自相关函数找到截尾点,以那个步长作为窗口的下限。切窗口时还要注意重叠率,一般用stride=1相邻窗口滑动,样本量充足,训练时也能看到更多模式之间的微小差异。
import numpy as np import torch from torch.utils.data import Dataset, DataLoader def create_sequences(data, seq_len, pred_len, stride=1): xs, ys = [], [] for i in range(0, len(data) - seq_len - pred_len + 1, stride): x = data[i:i + seq_len] y = data[i + seq_len:i + seq_len + pred_len] xs.append(x) ys.append(y) return np.array(xs), np.array(ys)注意越靠近末尾的数据越不能浪费,如果样本总量很少,可以适当调小stride做数据增强。我知道有人会问,窗口大了会不会引入过多噪声?会,但卷积网络有自动降噪的能力,后面会展开说。
2.2 标准化别偷懒
很多初学者在时间序列预测里最容易犯的错,就是直接把原始数值丢给模型。我的建议是无论如何先做标准化。常用的有两种:Z-score标准化和Min-Max归一化。
Z-score适合数据分布接近正态、有轻微离群值的场景:
from sklearn.preprocessing import StandardScaler scaler = StandardScaler() data_scaled = scaler.fit_transform(data.reshape(-1, 1)).flatten()Min-Max适合数据有明确上下界、且需要保持原始尺度关系的场景。不过我实测下来,Z-score在大多数情况下表现更稳,尤其是序列里有缓慢漂移趋势的时候,Z-score能把漂移变成相对变化,模型学起来更容易。
这里有一个极其关键的细节:标准化必须只用训练集的数据去fit,然后用这组参数去transform验证集和测试集。道理很简单——测试集代表未来未知的数据,你不能让模型在训练阶段就窥见测试集的均值和方差,否则验证效果会被虚假抬高,上线后的真实表现直接打折扣。这是数据泄漏的一种隐蔽形式,极其常见。
2.3 用TensorDataset封装训练/验证/测试数据
数据切好、标准化完成后,下一步是把numpy数组转成PyTorch能读的Dataset。
from torch.utils.data import TensorDataset X_train = torch.tensor(x_train, dtype=torch.float32).unsqueeze(1) # (样本数, 1, 窗口长度) y_train = torch.tensor(y_train, dtype=torch.float32) X_val = torch.tensor(x_val, dtype=torch.float32).unsqueeze(1) y_val = torch.tensor(y_val, dtype=torch.float32) X_test = torch.tensor(x_test, dtype=torch.float32).unsqueeze(1) y_test = torch.tensor(y_test, dtype=torch.float32) train_dataset = TensorDataset(X_train, y_train) val_dataset = TensorDataset(X_val, y_val) test_dataset = TensorDataset(X_test, y_test)这里unsqueeze(1)是为了给数据加上通道维度。Conv1d的输入形状是(batch, channels, length),单变量预测时channels=1;如果是多变量预测,把多个特征放在channels维度上就行。
3. PyTorch建模:从Conv1d到全连接输出
3.1 模型结构设计
网络结构我建议用三块:初始卷积层提取局部特征、中间卷积层逐步扩大感受野、最后接全连接层输出预测值。下面这个例子是我实战中调过很多次后沉淀下来的结构,适合大多数单变量序列预测场景。
import torch.nn as nn class CNN1D(nn.Module): def __init__(self, seq_len, pred_len, n_channels=1): super(CNN1D, self).__init__() self.conv1 = nn.Conv1d(n_channels, 64, kernel_size=3, padding=1) self.bn1 = nn.BatchNorm1d(64) self.conv2 = nn.Conv1d(64, 128, kernel_size=3, padding=1) self.bn2 = nn.BatchNorm1d(128) self.conv3 = nn.Conv1d(128, 128, kernel_size=5, padding=2) self.bn3 = nn.BatchNorm1d(128) self.relu = nn.ReLU() self.pool = nn.MaxPool1d(2) temp_input = torch.randn(1, n_channels, seq_len) with torch.no_grad(): temp_output = self._forward_conv(temp_input) flattened_size = temp_output.numel() self.fc1 = nn.Linear(flattened_size, 64) self.fc2 = nn.Linear(64, pred_len) def _forward_conv(self, x): x = self.relu(self.bn1(self.conv1(x))) x = self.pool(x) x = self.relu(self.bn2(self.conv2(x))) x = self.pool(x) x = self.relu(self.bn3(self.conv3(x))) x = self.pool(x) return x def forward(self, x): x = self._forward_conv(x) x = x.view(x.size(0), -1) x = self.relu(self.fc1(x)) x = self.fc2(x) return x第一层用64个卷积核,第二层128个,第三层保持128个不过度增加参数。每层卷积后面都跟了BatchNorm和ReLU,中间用MaxPooling降维。BatchNorm对训练的稳定性帮助很大,尤其是在数据分布有小幅变化的时候,能抑制特征分布漂移带来的训练震荡。
3.2 前向传播的维度变化
理解维度变化是写对模型的关键。假设输入是一批样本,形状为(batch=32, channels=1, seq_len=168),经过第一层卷积后变成(batch, 64, 168),MaxPool后变成(batch, 64, 84);经过第二层变成(batch, 128, 84),再到(batch, 128, 42);第三层后是(batch, 128, 42),池化后变成(batch, 128, 21)。展平后是128*21=2688维的特征向量,经过全连接层压缩输出pred_len个预测值。
关于全连接层输入维度的计算,我上面代码里用了动态推导的方式——构造一个假想输入过一次卷积部分,自动算出展平后的特征数。这样即使以后改了seq_len或者网络层数,也不需要手动心算维度,少掉很多麻烦。
3.3 损失函数与评估指标选择
时间序列回归任务的默认损失函数是MSE(均方误差),它对大误差的惩罚很大,模型会更倾向于避免大的预测偏差。MAE则对所有误差一视同仁,对离群值不那么敏感。我在实际项目中常用MSE训练,评估时同时看MSE和MAE,偶尔还会看MAPE(平均绝对百分比误差),方便向非技术同事解释准确率。
回归任务没有像分类那样的accuracy指标,所以很多人会忽略评估指标的选择。我的习惯是核心指标用RMSE,因为它的量纲和原始数据一致,能直观理解“平均预测偏差大概是多少”。如果业务对异常点更敏感,就换成MAE训练,配合一定的正则化。
4. 训练过程管理与回归任务的特殊处理
4.1 学习率、batch size与epoch的配合
训练CNN时间序列模型,优化器我首推AdamW,初始学习率设在1e-3到3e-3之间。如果训练时loss曲线震荡明显,把学习率降到5e-4左右。batch size方面,16到64之间比较合适。时间序列样本之间天然存在相关性,同一batch里的样本可能来自相邻时间段,这会让梯度估计有一定偏置,但也无伤大雅,不需要像某些场景那样刻意打乱样本顺序。
训练一个比较稳妥的做法是配合余弦退火调整学习率,让模型在后期逐步收敛到更平滑的极小值。下面的训练循环是我一直沿用的模板:
import torch.optim as optim from torch.optim.lr_scheduler import CosineAnnealingLR model = CNN1D(seq_len=168, pred_len=24) optimizer = optim.AdamW(model.parameters(), lr=1e-3, weight_decay=1e-4) scheduler = CosineAnnealingLR(optimizer, T_max=50) criterion = nn.MSELoss() for epoch in range(100): model.train() train_loss = 0.0 for x_batch, y_batch in train_loader: optimizer.zero_grad() y_pred = model(x_batch) loss = criterion(y_pred, y_batch) loss.backward() optimizer.step() train_loss += loss.item() scheduler.step() # 每个epoch后在验证集上评估一次,决定是否保存模型循环里的关键细节是每轮epoch结束要跑验证集评估,只看训练集loss极容易陷入过拟合。我把验证集loss当作模型选择的依据,后续只保留验证集loss最低的那一版参数。
4.2 训练中的早停与模型保存
早停法在时间序列模型里尤其实用,因为训练到一定阶段后验证集loss会先降后升,如果一直跑满全部epoch,保存的往往是过拟合后的模型。保险的做法是设置patience参数,比如连续10个epoch验证集loss没有刷新最低值,就终止训练。
best_val_loss = float('inf') patience = 10 trigger_times = 0 for epoch in range(100): # ... 训练代码 ... val_loss = evaluate(model, val_loader) if val_loss < best_val_loss: best_val_loss = val_loss trigger_times = 0 torch.save(model.state_dict(), 'best_model.pth') else: trigger_times += 1 if trigger_times >= patience: print(f'Early stopping at epoch {epoch}') break模型保存时不要整个序列化,只保存state_dict,方便之后加载和继续微调。加载时注意先实例化相同结构的模型,再load_state_dict。
4.3 预测后的反标准化
模型输出的预测值是标准化后的数值,要还原成实际业务数值才能做分析或展示。这一步必须用之前保存下来的scaler做inverse_transform,不要重新fit一遍。
y_pred_np = y_pred.detach().cpu().numpy() y_pred_inv = scaler.inverse_transform(y_pred_np.reshape(-1, 1)).flatten()很多时候我会把预测结果和真实值画在一张图上,通过可视化判断模型拟合的好坏。画图时同样要把真实值也反标准化回去,保持同一尺度对比。
5. 实测中容易踩的坑
5.1 数据泄漏问题
数据泄漏是最隐蔽也最致命的错误。除了前面说的标准化泄漏,还有一种情况是样本切分时的泄漏。比如你用前80%的数据做训练,后20%做测试,这没问题;但如果你用全量数据做了滑动窗口切分,再随机打乱划分训练集和测试集,那测试集里就会混入从时间上早于训练集某些样本的数据,模型相当于看到了“未来”的部分信息。时间序列必须严格按时间顺序划分,切分点之前的全部归训练,切分点之后归测试,中间可以留一段间隙作为缓冲。
我自己就栽过这个跟头。有一回某个项目的验证集效果一直漂亮得不正常,后来排查发现代码里顺手用了random_split,把相邻时间窗口的样本打散到了训练集和验证集,验证结果完全失真。改成按时间顺序切分后,效果掉了不少,但上线后的表现和验证结果才真正对得上。
5.2 卷积核大小与感受野
很多人会把图像领域的经验直接搬过来,卷积核一律用3x3。但一维时间序列里,kernel_size=3意味着每次只能看到3个时间步,感受野太小,模型无法捕捉稍长一点的趋势变化。我的经验是浅层用3或5捕捉局部变化,深层用7甚至更大的卷积核,配合stride或pooling逐层扩大感受野。感受野的计算公式是:
层数越深,单层卷积核越大,模型能关联到的历史上下文就越长。比如三层卷积,第一层kernel=5,第二层kernel=5,第三层kernel=7,各层stride=1,每层pooling=2,最终感受野可以覆盖到上百个时间步,对于多数日维度或者小时维度的序列预测已经完全够了。
5.3 周期性数据的处理
如果序列有明显的周期性,比如每天同一时段规律性变化,或者每周有周末效应,单纯把数值喂给CNN并不够。模型需要“知道”当前样本落在周期中的什么位置。两个常用手段:一是把时间特征(小时、星期、月份)作为额外通道拼接到输入;二是对序列做差分或者季节分解,剥离出趋势项和季节项,对残差项建模预测,最后再加回季节项。第二种做法在数据规律非常强烈的场景里提升效果显著。
实测中最简单的“低成本提效法”是把时间特征经过one-hot或者正弦编码后,直接拼到输入向量的尾部,也就是加长seq_len维度。CNN会因为多出来的特征通道自动学习不同时间位置的模式差异,效果通常优于纯靠数值序列硬学。
6. 效果对比与后续扩展
6.1 CNN与LSTM、MLP的对比
我拿某设备的连续运行数据做过一组对比实验,统一用前168小时预测后24小时,数据量大约有两万多条小时级记录。MLP(用同样窗口展平后过全连接网络)的RMSE大概在6.3左右;LSTM调了几轮之后能到4.8;一维CNN跑出来RMSE达到了4.2。训练时间上,同样的CPU环境,MLP用了不到2分钟,CNN大约6分钟,LSTM接近15分钟。
CNN能打赢LSTM的原因,在于时间序列里的许多模式是局部且重复出现的——比如尖锐的下降、缓慢的上升、短促的振荡。卷积核天然擅长提取这类局部模式,而且参数共享让模型不必为每个时间步单独学一套权重,泛化能力更强。LSTM的优势是记忆长距离依赖,但如果任务本身需要的依赖窗口不长,这个优势体现不出来,反而被顺序计算拖慢了速度。
6.2 多变量与CNN-LSTM混合方向
单变量预测只是入门,实际业务里更常见的场景是多变量输入、多变量输出。比如预测一台设备未来温度的同时,还要预测它的振动幅度和能耗。这时只要把输入从(batch, 1, seq_len)变成(batch, num_features, seq_len)就行,卷积会自动在特征维度上做混合。输出端则要保持多通道,让全连接层输出num_features×pred_len个数值再reshape。
再进一步,如果数据里既需要CNN的局部特征提取能力,又需要捕捉长期依赖,可以考虑CNN-LSTM混合结构:先用卷积层压缩序列长度,把降维后的特征序列喂给LSTM,最后接全连接输出。我在另一个短期预测项目里试过这个结构,比单独用CNN又降了约8%的误差,代价是训练时间增加约1.5倍,属于性价比很高的增强方案。
另外还有一个方向值得一提——把CNN替换成因果卷积,也就是带膨胀率的因果卷积网络,模拟RNN式的单向时序依赖,避免未来信息泄漏。PyTorch里可以用torch.nn.Conv1d配合dilation参数实现。这类结构在对序列依赖有严格因果要求的场景里更保险,比如实时在线预测系统。
我在实际使用中发现,CNN做时间序列预测的调试手感非常友好。调参无非就是卷积核大小、层数、通道数三个方向,比起LSTM那些门控单元的参数要直观得多。如果你当前的项目预测长度不长、数据量中等、训练资源有限,我建议先试一版一维CNN——很可能你会发现,自己之前花大力气调RNN的时间,本来可以省下一大半。