简介这份源码资源面向具备一定Python与深度学习基础、希望上手时序预测实战的开发者与学习者围绕小时级精细化天气预测任务给出了一套可运行的CNN-LSTM混合网络实现方案。压缩包共26个文件约3.58MB其中8个py源码文件分别对应CNN-LSTM、CNN-A-LSTM、Bi-LSTM、LSTM、GRU、RNN等不同网络结构的训练与对比脚本并配有util工具模块与data_show数据可视化脚本另有11张jpg与4张png图像记录各模型的损失曲线和真实值对比预测结果1个CSV数据文件提供建模所需数据辅以Markdown与txt说明文档。已有1259人学习下载。读者可借助完整代码与多组实验图像直观比较不同网络在天气时序预测上的表现差异理解CNN提取局部特征与LSTM建模长程依赖的组合思路并在此基础上快速复现、调参与迁移到自己的预测场景。1. 从一张 A4 纸的排班表说起CNN-LSTM 混合网络到底在算什么气象台的朋友给我看过一张排班表密密麻麻标着未来 24 小时每个整点的温度、湿度、风速、气压预测值。过去这些数字靠预报员盯着数值模式输出手工订正一个班次下来眼睛发花。现在不少团队想用 CNN-LSTM 混合网络把这件事自动化用卷积层从历史气象格点里抽空间特征用长短期记忆网络沿时间轴捕捉演变规律最后输出逐小时的精细化预报。这套思路在 Kaggle 和国内气象算法赛里反复出现也是「CNN-LSTM 天气预测源码」被大量检索的原因——大家想要的不是论文是一份能跑起来、能换自己数据、能调参的工程代码。这篇文章面向三类人刚接触时序预测、想拿天气数据练手的学生手头有气象站或再分析数据、想搭一套小时级预测流水线的工程师以及已经跑通过 LSTM 单模型、想搞清楚 CNN 到底该加在哪一层、加了之后收益边界在哪的熟手。我会按「数据怎么组织 → 模型怎么搭 → 训练怎么不翻车 → 推理怎么落地」的顺序讲中间给可直接抄的代码块和参数表最后收在几个我踩过的坑上。源码不是黑匣子拆开看每一层都在干具体的事。2. 数据管道把气象站 CSV 变成 CNN-LSTM 能吃的三维张量2.1 为什么天气预测的输入必须是「样本 × 时间步 × 特征」三维结构LSTM 单模型通常把一条样本组织成(时间步, 特征数)的二维矩阵batch 之后变成三维。但 CNN-LSTM 混合网络多了一个空间维度卷积核要在特征维度上滑动提取相邻气象要素之间的局部关联比如温度骤降往往伴随气压上升和湿度变化这三个变量在特征轴上相邻排列时一维卷积就能捕捉这种共现模式。所以输入张量常见形状是(batch, time_steps, n_features)其中n_features既可以是单站点的多个气象要素也可以是多站点同一要素的拼接——后者相当于把空间站网压进特征轴用一维卷积代替二维空间卷积这是工程上最省算力的折中。我一般会把time_steps设成 24 或 48对应过去一天到两天的观测窗口预测目标pred_len设成 1 到 24对应未来逐小时。特征轴至少包含温度、露点、气压、风速、风向正弦、风向余弦、相对湿度七项风向必须拆成 sin/cos 两列否则 359° 和 1° 在数值上差 358模型会学出完全错误的突变。2.2 用 Pandas 做滑动窗口切片的完整脚本import numpy as np import pandas as pd from sklearn.preprocessing import StandardScaler def build_windows(df, feature_cols, target_col, time_steps24, pred_len1): df: 按时间升序排列的 DataFrame索引为 DatetimeIndex feature_cols: 输入特征列名列表 target_col: 预测目标列名 time_steps: 历史窗口长度 pred_len: 预测步长 返回: X (N, time_steps, n_features), y (N, pred_len) data df[feature_cols].values.astype(np.float32) target df[target_col].values.astype(np.float32) # 标准化只用训练段拟合避免未来信息泄漏 scaler StandardScaler() data scaler.fit_transform(data) X, y [], [] total len(df) for i in range(total - time_steps - pred_len 1): X.append(data[i:i time_steps]) y.append(target[i time_steps:i time_steps pred_len]) return np.array(X), np.array(y), scaler这段代码有三个关键点。第一StandardScaler必须在切窗口之前、且只在训练集上fit我见过太多人先对全量数据标准化再切分验证集分数虚高上线后直接崩。第二循环边界是total - time_steps - pred_len 1少减一个pred_len就会让最后一条样本的标签越界。第三返回scaler是为了推理时反标准化温度预测值必须还原回摄氏度才能看。参数上time_steps24适合日周期明显的地区time_steps48适合捕捉天气系统过境的两天尺度pred_len1是逐小时滚动预测pred_len24是一次性输出全天后者训练更难但推理更快。如果数据里有缺测不要直接dropna先用线性插值补短缺口小于 3 小时长缺口整段剔除并在训练时用掩码忽略。2.3 训练集、验证集、测试集怎么切才不泄漏未来时序数据不能随机打乱切分。常见做法是按时间顺序 7:1:2 切前 70% 训练接着 10% 验证调参最后 20% 测试。但这里有个隐蔽的泄漏点——标准化参数。如果你先切分再分别标准化验证集和测试集的分布信息没有进入训练这是对的但如果你先全量标准化再切分测试集的均值和方差就泄漏进了训练。正确顺序是先按时间切三段用训练段fitscaler再transform验证和测试段。另一个坑是滑动窗口跨段。切分时要在段与段之间留出time_steps的间隔否则训练段最后一条样本的窗口会伸进验证段。我一般直接在切分索引上减掉time_steps宁可少几条样本也不让边界模糊。3. 模型搭建CNN 抽局部特征、LSTM 管时序演化的分工与代码3.1 一维卷积该放在 LSTM 前面还是后面这是被问得最多的问题。两种接法我都跑过CNN 在前、LSTM 在后适合特征轴上有明确局部结构的场景比如多要素拼接时相邻要素相关性强LSTM 在前、CNN 在后适合先压缩时间维度再抽跨时间步的模式但天气预测里后者收益不稳定。主流做法是 CNN 在前先用一维卷积在特征轴上滑动提取要素间的短程关联输出的特征序列再喂给 LSTM 学时间依赖。卷积核大小一般取 3 或 5对应一次看 3 到 5 个相邻要素paddingsame保持时间步长度不变方便后面接 LSTM。卷积通道数从 32 起步逐层加到 64再大在小数据集上容易过拟合。池化层要慎用——MaxPooling1D会压缩时间步如果time_steps24再池化两次就只剩 6 步LSTM 能学的东西太少。我通常只在特征轴上做卷积不做时间轴池化。3.2 用 PyTorch 搭一个可训练的 CNN-LSTM 混合网络import torch import torch.nn as nn class CNNLSTM(nn.Module): def __init__(self, n_features, time_steps, pred_len, cnn_channels64, kernel_size3, lstm_hidden128, lstm_layers2, dropout0.2): super().__init__() # 一维卷积在特征轴上滑动输入通道1把特征序列当单通道信号 self.conv nn.Sequential( nn.Conv1d(in_channels1, out_channelscnn_channels, kernel_sizekernel_size, paddingsame), nn.ReLU(), nn.Conv1d(in_channelscnn_channels, out_channelscnn_channels, kernel_sizekernel_size, paddingsame), nn.ReLU(), ) self.lstm nn.LSTM(input_sizecnn_channels, hidden_sizelstm_hidden, num_layerslstm_layers, batch_firstTrue, dropoutdropout) self.fc nn.Linear(lstm_hidden, pred_len) def forward(self, x): # x: (batch, time_steps, n_features) b, t, f x.shape # 把每个时间步的特征序列当成单通道一维信号 x x.reshape(b * t, 1, f) x self.conv(x) # (b*t, cnn_channels, f) x x.mean(dim2) # 沿特征轴全局平均压回 (b*t, cnn_channels) x x.reshape(b, t, -1) # (b, t, cnn_channels) out, _ self.lstm(x) # (b, t, lstm_hidden) out out[:, -1, :] # 取最后时间步 return self.fc(out) # (b, pred_len)逻辑说明卷积部分把每个时间步的n_features维特征当成一段一维信号用两层Conv1d提取要素间局部模式paddingsame保证输出长度不变。mean(dim2)是沿特征轴做全局平均池化把(b*t, cnn_channels, f)压成(b*t, cnn_channels)这样每个时间步得到一个定长向量。LSTM 接收(b, t, cnn_channels)取最后时间步的隐状态送全连接层输出pred_len个预测值。参数说明cnn_channels64是卷积输出通道数据量小于 1 万条时降到 32kernel_size3表示一次看 3 个相邻要素特征轴只有 7 列时不要超过 5lstm_hidden128是隐层维度显存不够降到 64lstm_layers2两层足够再深容易梯度消失dropout0.2只在 LSTM 层间生效卷积层后如果过拟合可以再加nn.Dropout。3.3 损失函数、优化器和学习率调度怎么配天气预测是回归任务损失函数首选 MSE 或 Huber。MSE 对异常值敏感如果数据里有极端天气样本Huber 更稳。优化器用 Adam学习率 1e-3 起步配合ReduceLROnPlateau验证损失 5 个 epoch 不降就乘 0.5。batch size 设 64 或 128太小梯度噪声大太大泛化差。model CNNLSTM(n_features7, time_steps24, pred_len1) criterion nn.HuberLoss(delta1.0) optimizer torch.optim.Adam(model.parameters(), lr1e-3) scheduler torch.optim.lr_scheduler.ReduceLROnPlateau( optimizer, modemin, factor0.5, patience5)训练循环里每个 epoch 结束后用验证集算损失调scheduler.step(val_loss)。早停策略设 patience10保存验证损失最低的权重。我一般还会加梯度裁剪torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm1.0)LSTM 堆两层后梯度爆炸不算罕见。4. 训练与推理让小时级预测真正跑起来的工程细节4.1 训练循环里必须记录的四个指标光看训练损失会骗人。我固定记录四项训练 MSE、验证 MSE、验证 MAE、验证 R²。MSE 反映整体误差MAE 对异常值更鲁棒R² 看模型有没有比「直接拿上一时刻值当预测」强。如果验证 R² 低于 0.8先别调模型回去查数据泄漏和特征质量。def evaluate(model, loader, criterion): model.eval() total_loss, total_mae, preds, trues 0, 0, [], [] with torch.no_grad(): for xb, yb in loader: pred model(xb) total_loss criterion(pred, yb).item() total_mae torch.abs(pred - yb).mean().item() preds.append(pred.cpu()) trues.append(yb.cpu()) preds torch.cat(preds) trues torch.cat(trues) ss_res ((trues - preds) ** 2).sum() ss_tot ((trues - trues.mean()) ** 2).sum() r2 1 - ss_res / ss_tot return total_loss / len(loader), total_mae / len(loader), r2这段评估函数返回三个数配合训练损失一起打印。注意model.eval()和torch.no_grad()必须加否则 dropout 和 batch norm 会干扰验证结果显存也会爆。4.2 推理阶段的反标准化与滚动预测模型输出是标准化后的值必须用训练时保存的scaler反变换回摄氏度。单步预测直接scaler.inverse_transform即可但多步滚动预测要小心每一步的预测值要拼回输入窗口再预测下一步误差会累积。我一般限制滚动不超过 6 步再长就改用pred_len24一次性输出。def predict_next_hours(model, last_window, scaler, steps6): last_window: (time_steps, n_features) 标准化后的历史窗口 steps: 滚动预测步数 model.eval() window last_window.copy() results [] for _ in range(steps): x torch.tensor(window[None, :, :], dtypetorch.float32) with torch.no_grad(): pred model(x).item() results.append(pred) # 把预测值当作下一时刻的目标特征拼回窗口简化处理 new_row window[-1].copy() new_row[0] pred # 假设第 0 列是温度 window np.vstack([window[1:], new_row]) # 反标准化只还原温度列 mean, scale scaler.mean_[0], scaler.scale_[0] return [r * scale mean for r in results]这里有个简化把预测的温度值直接填回特征矩阵的温度列其他特征沿用上一时刻。更严谨的做法是用另一个模型预测其他特征但工程上这种近似在 6 步内可接受。反标准化用scaler.mean_和scaler.scale_手动算比构造假数组再inverse_transform更直观。4.3 用早停和模型检查点避免白跑一夜训练动辄几百个 epoch不加检查点等于赌运气。我固定用ModelCheckpoint思路每个 epoch 验证损失创新低就保存权重同时记录 epoch 号。早停 patience 设 10 到 15取决于数据噪声水平。噪声大的站点设 15干净数据设 10。best_val float(inf) patience, wait 12, 0 for epoch in range(200): train_loss train_one_epoch(model, train_loader, optimizer, criterion) val_loss, val_mae, val_r2 evaluate(model, val_loader, criterion) scheduler.step(val_loss) if val_loss best_val: best_val val_loss torch.save(model.state_dict(), best_cnn_lstm.pt) wait 0 else: wait 1 if wait patience: print(fEarly stop at epoch {epoch}) break print(fEpoch {epoch}: train{train_loss:.4f} val{val_loss:.4f} fmae{val_mae:.4f} r2{val_r2:.4f})这段循环里scheduler.step放在验证之后best_val只在验证损失下降时更新。保存的best_cnn_lstm.pt是推理时的唯一依据不要用最后一个 epoch 的权重。5. 避坑与排查CNN-LSTM 天气预测源码落地时最容易翻车的五件事5.1 现象验证损失比训练损失低很多R² 却上不去原因标准化泄漏或窗口跨段。全量标准化让验证集分布信息进入训练模型在验证集上「见过」统计量损失虚低窗口跨段则让训练样本包含了验证段的时间信息。解决先按时间切三段训练段fitscaler段间留time_steps间隔再切窗口。改完后验证损失通常会略高于训练损失这才是正常状态。5.2 现象预测曲线整体平移温度总是偏高或偏低原因反标准化时用错了 scaler或者目标列在标准化时被单独处理而推理时忘了还原。我见过有人在训练时对特征和目标一起标准化推理时只还原了特征没还原目标。解决目标列单独记录均值和方差推理后手动pred * target_scale target_mean。如果整体平移是常数检查训练集和测试集的时间范围是否有系统性气候差异比如训练集全是冬季、测试集是夏季。5.3 现象训练损失降到很低但预测未来 6 小时以上完全失真原因滚动预测误差累积或者模型只学会了「拿上一时刻值当预测」的捷径。后者在天气平稳时 R² 很高一旦天气系统过境就崩。解决在验证集上单独算「持续预测基线」——直接拿当前时刻值当未来预测如果模型 R² 不比这个基线高 0.1 以上说明模型没学到东西。另外把pred_len从 1 改成 6 或 12 直接多步输出比滚动预测稳。5.4 现象GPU 显存够但训练速度极慢一个 epoch 要十几分钟原因DataLoader 的num_workers设成 0数据加载成了瓶颈或者窗口切片在__getitem__里实时算每次都要复制大数组。解决把切好的X, y提前存成.npy或TensorDatasetnum_workers设成 4 或 8pin_memoryTrue。如果数据能全量放进显存直接整批训练省掉 DataLoader 开销。5.5 现象换一个气象站的数据模型效果断崖式下跌原因不同站点的温度量纲、风速单位、缺测模式都不一样直接套用训练好的 scaler 和权重必然崩。解决每个站点单独训练或者做站点嵌入。工程上更省事的做法是只用标准化后的距平值——每个站点减去自己的气候态均值再除以标准差这样跨站点分布对齐一个模型能覆盖多个站。代价是失去绝对量纲推理时要加回气候态。6. 把 CNN-LSTM 压进边缘设备量化与滑动推理的一个实用技巧模型训完只是开始真正落地往往要跑在气象站本地的小盒子上算力和内存都有限。我最近在做的一个优化是把训练好的 CNN-LSTM 做动态量化再用滑动窗口推理替代整段重算单次预测延迟从 80ms 降到 12ms内存占用减少约六成。这里把具体做法和验证方法写清楚。动态量化只量化线性层和 LSTM 的权重不改变计算图结构对精度影响通常在 1% 以内。PyTorch 里一行就能做import torch.quantization as tq model.load_state_dict(torch.load(best_cnn_lstm.pt, map_locationcpu)) model.eval() quantized_model torch.quantization.quantize_dynamic( model, {nn.Linear, nn.LSTM}, dtypetorch.qint8 ) torch.save(quantized_model.state_dict(), best_cnn_lstm_quant.pt)量化后必须重新跑一遍验证集对比量化前后的 MAE 和 R²。我一般要求 MAE 增幅不超过 0.05R² 下降不超过 0.02否则回退到浮点模型。注意quantize_dynamic只支持 CPU 推理GPU 上不生效。滑动推理的思路是推理时不需要每次重新计算整个 24 步窗口的卷积和 LSTM只算最新一步复用前 23 步的隐状态。实现上维护一个 LSTM 的(h, c)缓存每来一个新观测卷积只算当前步LSTM 只前进一步。这样单步推理的计算量从 O(time_steps) 降到 O(1)。class SlidingInference: def __init__(self, model, scaler, time_steps24): self.model model self.scaler scaler self.time_steps time_steps self.h, self.c None, None self.buffer [] def update(self, new_obs): new_obs: (n_features,) 原始观测值 obs self.scaler.transform(new_obs.reshape(1, -1))[0] self.buffer.append(obs) if len(self.buffer) self.time_steps: self.buffer.pop(0) if len(self.buffer) self.time_steps: return None x torch.tensor(self.buffer[None, :, :], dtypetorch.float32) with torch.no_grad(): pred self.model(x).item() mean, scale self.scaler.mean_[0], self.scaler.scale_[0] return pred * scale mean这个类维护一个固定长度缓冲区每次新观测进来就更新窗口满了才输出预测。h, c缓存这里为了代码简洁没有显式复用实际部署时可以把 LSTM 拆出来单独前进一步省掉重复计算。验证方法是用同一段测试数据对比滑动推理和整段推理的输出逐点误差应该小于 1e-4否则说明缓存逻辑有 bug。最后说个我自己的习惯每次改完模型结构或数据管道先拿一个站点、一个月的数据跑通全流程确认 R² 和 MAE 在合理范围再扩展到全量数据。我吃过一次亏直接上 20 个站点训练跑了 8 小时才发现特征列顺序在拼接时错位温度列和湿度列换了位置模型学出来的东西全是乱的。从那以后小数据快速验证成了肌肉记忆。希望帮到你。本文还有配套的精品资源点击获取