CNN-LSTM实战：小时级天气预测的模型构建与避坑指南

简介这套基于CNN-LSTM混合网络的精细化小时级天气预测Python源码专为刚接触深度学习的初学者打造可直接用于课程设计、期末大作业或天气预测实战练习。项目共包含26个文件以8个Python脚本为核心分别实现CNN-LSTM、LSTM、GRU、RNN及双向LSTM等多种对比模型并配有11张jpg和4张png格式的训练损失曲线、真实值与预测值对比图另有1个CSV格式的天气数据集、2份Markdown说明文档及操作手册压缩包整体约1.82MB结构清晰便于快速上手。目前已有235人学习下载。通过这份资源读者不仅能掌握CNN与LSTM混合建模的完整流程还能直接运行源码观察各模型在小时级预测任务上的效果差异尤其适合需要答辩展示和对比分析的场景为理解时序预测与深度学习结合提供了直观的实战参考。1. 用 CNN-LSTM 做小时级天气预测数据少、变化快凭什么能落地去年给一个光伏电站做短期功率预测数值天气预报的粗网格完全扛不住局地强对流后来我们把地面站过去 12 小时的温度、湿度、气压、风速直接喂进 CNN-LSTM 混合网络做逐小时外推。结果温度预测的 MAE 从 2.1°C 降到 1.3°C单张显卡就能跑完训练这个对比让我确信小时级天气预测不一定非得依赖数值模式。标题里的 CNN-LSTM 就是这个思路的标准实现——用 CNN 提取特征通道间的局部天气形态用 LSTM 记住跨小时的状态依赖最后输出下一个小时甚至未来 24 小时的气象要素。适合手里有站点观测数据或再分析数据想做光伏功率、风电功率、交通气象服务的工程师也适合拿天气做毕业设计的本科生。2. 模型选型与数据处理先搞清天气信号该怎么喂给网络2.1 为什么单用 LSTM 不够小时级序列里的空间模式小时级气象观测不是一组逐一独立的数字而是一段段过程形态温度在 3 小时内陡降 5°C气压在锋面过境时出现尖刺湿度在降雨前持续抬升。CNN 的局部感受野天生适合抓这些形态一个 kernel_size3 的一维卷积等于在看“过去 3 小时这个特征通道的变化斜率”。而 LSTM 擅长的是把这些局部形态串起来形成“先升温、后风起、再下雨”这种长程因果链。如果只用 LSTM每个时刻的特征被压成同一个向量输入门控记忆要自己去发现不同特征之间的横向组合关系这对数据量只有几千条样本的小时级任务来说负担太重。只用 CNN 则相反它能提取局部形态但对 6 小时以上的趋势记忆很弱。把两者串联成 CNN-LSTM先让卷积层把每个时间步的特征重构成高维表示再把表示序列交给 LSTM 做时序建模这在天气外推任务里是比纯 LSTM 和纯 CNN 都稳妥的选择。我在单站点场景里习惯把“特征维”当成一维卷积的通道在时间轴上做卷积。这是什么意思原始样本的形状是 (window, features)把特征维挪到通道位之后变成 (features, window)Conv1d 沿 window 方向滑动每个卷积核同时看到所有特征。这让模型能学到“低气压 高湿度 风突变”这类跨特征组合而不是只看单变量的历史。多站点场景只需要把一维卷积换成二维卷积把站点经纬度排成网格道理一样。2.2 数据源与特征工程把小时观测变成可训练样本常见的数据源有两种一是地面气象站的逐小时观测二是再分析数据集比如 ECMWF 的 ERA5按目标点经纬度插值后可以补全站点稀疏地区的空白。站点的好处是真实、直接坏处是缺测多ERA5 的空间分辨率约 0.25°时间分辨率 1 小时适合做模型输入的底图。我做项目时通常以站点观测为主缺测时段用 ERA5 插值结果补上避免模型在缺失值上产生虚假记忆。特征选择直接影响模型上限。下表是我在小时级气温预测里最常用的一组特征特征单位说明t2m°C 或 K2 米气温建模前统一为 °Crh%相对湿度pshPa地面气压wind_u / wind_vm/s风速的纬向 / 经向分量tpmm/h前一小时累计降水hour_sin / hour_cos—小时序号的周期编码有两个细节容易被小白忽略。第一风向必须拆成 u/v 分量直接用 0~360° 的角度数值会让模型误以为 359° 和 0° 是两个极端方向第二降水要做成“前一小时累计量”而不是瞬时值。环境这块不用折腾太久Anaconda 建一个虚拟环境把 PyTorch、pandas、numpy、scikit-learn 一起装上VSCode 里选对解释器就行网上 python 安装教程很多跟着把 conda 环境跑通就够用。2.3 滑动窗口与标准化代码复制前先确认三件事数据准备的核心是构造“过去 window 小时预测未来 1 小时”的监督学习样本。下面这段代码我平时直接复用列名对齐后替换成自己的文件就能跑import pandas as pd import numpy as np df pd.read_parquet(hourly_obs.parquet) df df.sort_values(time).reset_index(dropTrue) # 确保按时间递增 feature_cols [t2m, rh, ps, wind_u, wind_v, tp] df[target] df[t2m].shift(-1) # 用 t 时刻特征预测 t1 时刻温度 df df.dropna(subset[target]).reset_index(dropTrue) window 12 data df[feature_cols].to_numpy(dtypenp.float32) X_all, y_all [], [] for i in range(len(data) - window): X_all.append(data[i : i window]) # [t-11, ..., t] y_all.append(df[target].iloc[i window - 1]) # 窗口末尾 t 时刻的标签 X_all np.array(X_all) # (样本数, 12, 6) y_all np.array(y_all)这段代码把 12 小时观测窗口和未来 1 小时目标对齐在一起。注意 target 用的是 shift(-1)所以窗口右端是 t 时刻预测的却是 t1如果你把 df[t2m] 直接当标签模型只需要复制最后一步输入就能把 loss 打到 0后面第四章会详细讲这个坑。window12 表示过去半天气温这种平稳变量 12 够用风速建议加到 24小时降水不建议直接回归这一点后面也会展开。标准化同样有讲究必须严格按时间顺序切分后只用训练段计算统计量from sklearn.preprocessing import StandardScaler n_train int(len(X_all) * 0.8) X_train, X_val X_all[:n_train], X_all[n_train:] y_train, y_val y_all[:n_train], y_all[n_train:] scaler_x StandardScaler() scaler_x.fit(X_train.reshape(-1, len(feature_cols))) X_train scaler_x.transform(X_train.reshape(-1, len(feature_cols))).reshape(X_train.shape) X_val scaler_x.transform(X_val.reshape(-1, len(feature_cols))).reshape(X_val.shape) scaler_y StandardScaler() scaler_y.fit(y_train.reshape(-1, 1)) y_train scaler_y.transform(y_train.reshape(-1, 1)).flatten() y_val scaler_y.transform(y_val.reshape(-1, 1)).flatten()这段代码的关键只有一点scaler_x 和 scaler_y 只 fit 训练段验证段和未来推理都用训练段的统计量做 transform。如果你的数据有季节漂移全量 fit 会把未来信息提前泄漏给训练集验证分数好看但部署时必翻车。两个 scaler 对象要单独保存好后面保存模型时要一起存进 checkpoint。3. 用 Python 从零跑通 CNN-LSTM模型定义、训练与单步预测3.1 CNN-LSTM 网络结构定义一维卷积加在哪个维度模型结构不复杂难在搞清楚每个维度的排列。我常用的定义如下import torch.nn as nn class CNNLSTM(nn.Module): def __init__(self, n_features, window12, hidden64): super().__init__() self.conv nn.Sequential( nn.Conv1d(n_features, 32, kernel_size3, padding1), nn.ReLU(), nn.Conv1d(32, 64, kernel_size3, padding1), nn.ReLU(), ) self.lstm nn.LSTM(64, hidden, num_layers1, batch_firstTrue) self.fc nn.Linear(hidden, 1) def forward(self, x): # 输入 x: (batch, window, n_features) x x.permute(0, 2, 1) # 变成 (batch, n_features, window) x self.conv(x) # 输出 (batch, 64, window) x x.permute(0, 2, 1) # 还原为 (batch, window, 64) out, _ self.lstm(x) return self.fc(out[:, -1, :]) # 取最后一个时间步的输出新手最容易搞错的是 Conv1d 的输入形状。PyTorch 的 Conv1d 期望输入是 (batch, channels, length)所以 forward 里先做一次 permute把特征维放到 channels 位置时间维放到 length 位置。卷积核 size3 意味着每个输出点同时看连续 3 个小时的所有特征padding1 保证输出长度仍然是 window这样后面 LSTM 还能按原步长处理序列。LSTM 部分我取最后一步的隐藏输出过全连接。为什么不取所有步再池化因为我们要预测的是 t1 的单一数值最后一步已经包含了前面所有时刻压缩后的信息取均值反而会稀释近期状态。hidden64 在几千条样本的规模下已经够用调大容易过拟合调小记不住长程依赖。3.2 训练代码与超参数时序切分、标准化与早停训练循环本身不特殊但有几个习惯值得坚持验证集始终保持最后一段时间batch 内可以 shuffle样本之间不跨段混淆。下面是完整训练代码import torch from torch.utils.data import TensorDataset, DataLoader train_ds TensorDataset(torch.tensor(X_train, dtypetorch.float32), torch.tensor(y_train, dtypetorch.float32)) train_loader DataLoader(train_ds, batch_size128, shuffleTrue) model CNNLSTM(n_featureslen(feature_cols), windowwindow, hidden64) optimizer torch.optim.Adam(model.parameters(), lr1e-3) loss_fn nn.MSELoss() best_val_loss float(inf) for epoch in range(50): model.train() train_loss 0.0 for xb, yb in train_loader: optimizer.zero_grad() pred model(xb).squeeze() loss loss_fn(pred, yb) loss.backward() optimizer.step() train_loss loss.item() * len(xb) model.eval() with torch.no_grad(): val_pred model(torch.tensor(X_val, dtypetorch.float32)).squeeze() val_loss loss_fn(val_pred, torch.tensor(y_val, dtypetorch.float32)).item() if val_loss best_val_loss: best_val_loss val_loss torch.save(model.state_dict(), best_model.pt) if (epoch 1) % 5 0: print(fepoch {epoch1:02d}, train_loss {train_loss/len(X_train):.4f}, val_loss {val_loss:.4f})训练代码里有两点必须解释清楚。第一验证集是最后 20% 的时间段不是随机抽的天气数据相邻小时强相关随机切分会造成验证集和训练集互相重叠指标看着好实际泛化崩盘第二val_loss 每次下降就覆盖保存一次 best_model.pt这就是最简单有效的早停比固定跑 50 个 epoch 更稳。超参数方面我整理了一份可以直接照抄的参考表参数推荐值说明window12过去 12 小时气温取 6~24 可调kernel_size3一维卷积核跨 3 小时局部趋势hidden64LSTM 隐藏维数样本少别太大num_layers1两层以上在小样本上容易过拟合batch_size128样本几千条时 128 稳定lr1e-3Adam 默认收敛不稳降到 5e-4epochs50 早停看 val_loss 保存最佳权重3.3 模型保存和单步预测把权重、scaler 一起存下来训练结束后保存模型时一定要把 scaler、特征列名、window 一起存进去。否则推理时你会忘记当时用哪几列、窗口多大这种返工纯属浪费。推荐存成一个字典torch.save({ state_dict: model.state_dict(), feature_cols: feature_cols, window: window, scaler_x: scaler_x, scaler_y: scaler_y, }, cnn_lstm_weather.pt) def predict_next(ckpt_path, x_last): ckpt torch.load(ckpt_path) model CNNLSTM(n_featureslen(ckpt[feature_cols]), windowckpt[window]) model.load_state_dict(ckpt[state_dict]) model.eval() x_norm ckpt[scaler_x].transform(x_last.reshape(1, -1)).reshape(1, ckpt[window], -1) with torch.no_grad(): y_norm model(torch.tensor(x_norm, dtypetorch.float32)).item() return ckpt[scaler_y].inverse_transform([[y_norm]])[0, 0]predict_next 输入 x_last 是最近 window 个时刻的原始特征序列函数内部先做标准化再走模型最后把标准化后的输出反变换成真实温度。这里有一个隐藏细节x_last.reshape(1, -1) 是把 (window, features) 的二维数组压平成一行喂给 StandardScaler.transform再 reshape 回 (1, window, features)顺序不能乱。单步预测验证好不等于滚动预测好下一步要做的是把模型放到更严苛的验证流程里。4. 小时级天气预测避坑五个让模型翻车的常见问题4.1 归一化统计量用全量数据验证集好看上线就翻车现象验证集 RMSE 特别低有时甚至低于训练集部署后发现预测值整体偏移严重时误差翻倍。原因用 StandardScaler 对整个 X_all 做 fit未来时段的均值和方差被偷看进了训练过程模型相当于提前知道了季节和气候背景。解决严格按时间切分后只对训练段 fit验证段和推理段统一复用训练段的 scaler。这个错我早年踩过一次之后写代码永远先把时间切分写在标准化前面算是一点血泪经验。4.2 标签没用 shift(-1)模型学的是“复制粘贴”现象loss 很小预测曲线比真实曲线平滑一个相位散点图上的点几乎全落在 yx 对角线上。原因把 t 时刻的 t2m 当目标LSTM 最后一个时间步的输出可以直接复制输入值网络不需要学习任何天气规律。解决构建 target 时用 shift(-1)并打印窗口末端时间和标签时间的差值做校验。模型内部是个黑匣子但标签对齐这件事完全可以在数据层面自查别等到曲线出来才后悔。4.3 降水样本被晴天淹没损失函数只看见晴天现象模型输出的降水预测几乎全是 0偶尔有点值也远小于观测。原因晴天样本占绝大多数MSE 对样本均值加权模型发现“全预测 0”就能拿到很小 loss少数强降水事件被淹没了。解决不要用同一个回归头同时处理气温和降水。常见做法是先训练一个降水有无的二分类模型再用事件样本训练量级回归模型损失函数可以用 weighted MSE 给降水样本加权。小时级降水本身就是难题不要指望单一模型通吃。4.4 随机切分训练/验证集时间重叠让指标虚高现象验证指标比线上表现好一大截模型跑新数据时衰减明显。原因天气时间序列相邻小时高度自相关随机切分后验证集里混进了大量和训练样本相隔仅几小时的相似样本模型等于提前背过答案。解决按连续时间块切分例如前 80% 时间做训练、后 20% 做验证再严格一点可以按“天”为单位留出连续 5~7 天做验证。之前见过有人拿网上的免费源码直接套用随机切分一步到位最后上线误差涨 40%问题根本不在模型结构而在这个基本的数据划分。4.5 线性插值补小时降水插出 0.3mm 的假雨现象预处理后的降水特征里出现大量 0.05、0.3、0.7 这类非原始量级的数值模型对小雨产生系统性误差。原因pandas 的 interpolate() 默认线性插值“无雨 0mm”和“降雨 3mm”这两个离散状态之间被硬插出一条连续斜坡制造了真实世界不存在的假数据。解决对降水、风向这类离散或循环变量用最近邻插值或直接留空让模型处理缺失只有气温、气压这类连续变量才适合线性插值。数据清洗阶段多花十分钟后面少调三天参。5. 部署前的最后验证先赢过 persistence 基线再谈滚动预测5.1 persistence 基线怎么算小时级短临预测里最容易被忽略但最应该先打过的对手是持久性预测直接用当前观测值当作下一小时的预测值。对 1~6 小时的短时外推这个基线非常强因为天气在短时间内就是缓慢连续变化的。如果 CNN-LSTM 在验证集上连 persistence 都赢不了说明模型学到的主要是噪声不值得上线。pred_val_raw scaler_y.inverse_transform(pred_val.reshape(-1, 1))[:, 0] y_val_raw scaler_y.inverse_transform(y_val.reshape(-1, 1))[:, 0] baseline_rmse np.sqrt(np.mean((y_val_raw[1:] - y_val_raw[:-1]) ** 2)) model_rmse np.sqrt(np.mean((pred_val_raw - y_val_raw) ** 2)) print(fpersistence RMSE: {baseline_rmse:.3f}, CNN-LSTM RMSE: {model_rmse:.3f})其中 pred_val 是模型在验证集上的标准化预测反变换后得到原始尺度。实际项目中我的经验是1 小时预测打赢 persistence 很正常3~6 小时稳定打赢才是模型真正学到了东西。如果 6 小时预报的改善率低于 10%优先去检查数据质量而不是继续堆模型参数。5.2 滚动预测与 MC Dropout对预报曲线给个“可信区间”单步预测验证通过后还要测滚动预测。递归方式是把上一轮预测值当作下一轮特征回填但这种做法误差会累积预测 24 小时时后半段基本失去参考意义。我在实战里常用一个折中只把气温回填进 t2m 通道其他特征保持最近观测值避免多变量误差相互放大。def recursive_predict(model, x_last, steps24): model.eval() cur torch.tensor(x_last, dtypetorch.float32).unsqueeze(0) preds [] with torch.no_grad(): for _ in range(steps): y_norm model(cur).item() preds.append(y_norm) nxt cur[0, -1, :].clone() nxt[0] (scaler_y.inverse_transform([[y_norm]])[0, 0] - scaler_x.mean_[0]) / scaler_x.scale_[0] cur torch.cat([cur[:, 1:, :], nxt.unsqueeze(0).unsqueeze(0)], dim1) return scaler_y.inverse_transform(np.array(preds).reshape(-1, 1))[:, 0]如果想让预报对业务更有价值可以加一个很简单的 MC Dropout 不确定性估计预测时打开 model.train()同一个输入重复 forward 30 次对结果求均值和标准差标准差就能当作“这个时刻的预测可信度”。我吃过一次亏单步验证 RMSE 非常好24 小时滚动到第 14 步预测曲线开始漂离真实值靠滚动测试才发现问题。后来我把 persistence 对比和滚动预测写成了固定测试流程模型改造前必须先过这两关。所有技巧最后都是为稳定这两个验证服务的希望帮到你。本文还有配套的精品资源点击获取