LSTM预测AQI完整指南：从数据预处理到模型训练与避坑

简介：基于LSTM的空气质量指数预测完整源码，面向Python课程设计、期末大作业及毕业设计场景，适合需要快速获得可运行高分项目的在校生，项目已获导师认可，可直接作为高分作品提交。工程包含数据预处理、模型搭建、训练与预测等完整模块，代码注释清晰，新手也能对照理解。资源共2000个文件，其中1989个csv为北京多日期空气质量历史数据，另有4个py核心脚本、5个xml配置、1个md说明文档及少量pyc文件，压缩包仅29.33MB，便于下载部署。目前已有222人学习下载，项目经过严格调试，可直接运行。通过该项目可掌握LSTM时间序列预测的基本流程，学习数据清洗、序列构造、模型评估与可视化等关键环节，同时可参考其工程目录结构，便于自行扩展多站点实验，具有较高参考价值。

1. 这个题目为什么年年有、年年劝退：LSTM预测AQI的真实难度

Python期末大作业里，基于LSTM的空气质量指数预测源码可以说是出现频率最高的题目之一，但它也是最容易让新手翻车的题目：网上能搜到的源码很多，真正能跑通、能画图、能写进报告里的却很少。很多拿到高分项目的同学，并不是因为理解了什么高深理论，而是把时间窗口、归一化、数据切分这三件小事做对了。这篇笔记就把做这个方向最常见的方案完整拆开，从数据预处理到模型训练，再到最后阶段最容易出的六个坑，全部按可复现的步骤写清楚。适合两类读者：一类是期末要交代码和报告、希望直接落地的同学；另一类是已经有Python基础、想通过这个题目理解LSTM到底怎么处理时序数据的人。

2. 从数据到特征：AQI预测里决定模型上限的预处理

2.1 空气质量数据从哪里找，字段怎么选

做这个项目的第一个现实问题是数据源。常见的做法是从公开的空气质量监测平台拉取某市过去一年到三年的逐小时监测记录，字段一般包括时间戳、AQI、PM2.5、PM10、SO2、NO2、O3、CO以及气象条件（温度、湿度、风速、风向）。如果没有气象字段也不是不能做，只用污染物浓度字段预测AQI也能跑通，只是准确率上限会明显低一些，因为AQI对湿度和风速的依赖很强。

我一般会建议选一个污染物种类比较全的数据集来做，即使期末题目只是要求预测AQI，也可以用PM2.5、PM10、O3等作为辅助输入特征。原因很简单：LSTM学习的是变量之间的联合时序模式，单一AQI序列能提供给模型的信息量非常有限，特别是沙尘回流或者雾霾积累这类过程，单独看AQI数字很难捕捉趋势，但PM10和PM2.5会先有变化。多特征输入并不会显著增加写代码的复杂度，却能让预测曲线明显更贴合真实值。

2.2 第一步先处理缺失值与异常值，别急着归一化

拿到数据后首先做的是时间排序与缺失值处理。监测站偶尔会有设备校准停采或者网络传输丢包的情况，表现出来的就是某几个小时的记录是空值或者AQI数值直接为零。AQI为零是物理上不可能出现的，按异常值处理，而缺失时间点的处理方式通常用前向填充或者线性插值。前向填充适合短时间缺失，比如连续两三个小时，直接拿前一小时的值补上，代码最简单。超过六个小时的连续缺失，我建议用线性插值，否则一段趋势会被拉平。

import pandas as pd import numpy as np df = pd.read_csv("air_quality.csv", parse_dates=["time"]) df = df.sort_values("time").reset_index(drop=True) # AQI小于等于0的按异常值处理，替换为NaN后再插值 df.loc[df["aqi"] <= 0, "aqi"] = np.nan # 缺失列按时间顺序做线性插值，限制连续缺失超过6小时不补 df = df.interpolate(method="linear", limit=6, limit_direction="forward") # 插值之后仍存在的空值用前向填充兜底，极少出现 df = df.fillna(method="ffill") df = df.dropna().reset_index(drop=True)

这里有一个容易被忽视的点：interpolate的limit参数限制的是单个连续缺失区间最多填充多少个点，而不是总缺失数量。设置为6意味着连续缺失超过6小时的部分仍然保留NaN，再由ffill兜底。选择线性插值而不是三次样条的原因是AQI在短时间内的变化近似线性，三次样条反而可能在缺失区间内产生过冲，比如某一天PM2.5骤降时，样条会在端点处多拐出一个并不存在的峰值。

2.3 滑动窗口与归一化：两个参数直接决定模型能不能收敛

模型输入不是一整年的数据，而是固定长度的滑动窗口，这是LSTM做时间序列预测的基本思路。假设用过去48小时的数据预测未来1小时的AQI，那么输入张量的形状是(窗口长度48, 特征数N)。窗口长度是一个需要先定下来的超参数，经验值是24到72之间。窗口太短模型看不到污染积累的过程，窗口太长会让训练集样本数量锐减，而且超过72小时的历史对当前小时AQI的影响已经很小。

归一化在这个项目里不是可选项，而是必选项。我见过不少人直接拿原始AQI数值训练，结果是Loss一直不降，最后归咎于模型结构有问题。实际上LSTM内部使用的是tanh激活函数，对输入数值范围非常敏感，几百的AQI和几万的污染物浓度数值直接输入，梯度计算很容易出问题。更推荐用MinMaxScaler而不是StandardScaler，因为LSTM输出端没有约束时，MinMax缩放到0到1范围能让模型更容易收敛，逆变换回真实AQI时也更直观，不容易出现负数预测值。

from sklearn.preprocessing import MinMaxScaler feature_cols = ["pm25", "pm10", "so2", "no2", "o3", "co", "aqi"] scaler = MinMaxScaler(feature_range=(0, 1)) scaled = scaler.fit_transform(df[feature_cols]) # 把缩放后的数据写成 DataFrame，保持时间索引便于后续绘图 scaled_df = pd.DataFrame(scaled, columns=feature_cols, index=df.index)

重点说一下fit_transform的位置。正确的做法是只用训练集的数据来fit，再对验证集和测试集做transform。如果先对全量数据做fit_transform再切分，等于验证集和测试集的信息在训练时就已经被模型间接看到了，最后得出的准确率虚高，答辩时老师一问数据切分方式就会暴露。MinMaxScaler的feature_range设置为(0, 1)是为了配合LSTM的输出层，如果你后续在最后一层用了Sigmoid激活，那么这个范围是匹配的；如果用线性输出层，其实(-1, 1)的范围收敛也很快，但逆变换时容易因预测值超出边界产生截断，所以我更习惯用(0, 1)。

2.4 训练集、验证集、测试集必须按时间顺序切，不能乱打散

做分类任务时习惯用train_test_split并设置随机种子随机打散，但做时间序列预测时千万不能这么做。随机打散会让模型看到未来的数据，相当于考试时把答案放在旁边。正确做法是按时间顺序切分，比如前70%作为训练集，接着15%作为验证集，最后15%作为测试集，三者时间上不重叠。验证集的用途是调超参数和做早停判断，测试集只允许在最终评估时用一次，这样报告里写出来的RMSE才有说服力。

n = len(scaled_df) train_end = int(n * 0.7) val_end = int(n * 0.85) train_df = scaled_df.iloc[:train_end] val_df = scaled_df.iloc[train_end:val_end] test_df = scaled_df.iloc[val_end:]

切分后需要将每个数据集转换成模型需要的窗口样本。具体来说，对训练集的每一行，取其前48小时的全部特征作为输入，取该时刻的AQI值作为标签。这里有一个细节：当构造窗口时，标签使用的是当前时刻的AQI，即t时刻的AQI由t-48到t-1时刻的特征来预测，而不是包含当前时刻。如果误把当前时刻的AQI也放进输入特征，模型相当于直接读取了答案，Loss会极低但没有任何实际预测能力，这是很多高分源码里隐藏的作弊写法。

def create_sequences(data, feature_cols, target_col, seq_len): X, y = [], [] data_len = len(data) for i in range(seq_len, data_len): X.append(data.iloc[i - seq_len:i][feature_cols].values) y.append(data.iloc[i][target_col]) return np.array(X), np.array(y) seq_len = 48 target_col = "aqi" X_train, y_train = create_sequences(train_df, feature_cols, target_col, seq_len) X_val, y_val = create_sequences(val_df, feature_cols, target_col, seq_len) X_test, y_test = create_sequences(test_df, feature_cols, target_col, seq_len) print(X_train.shape, y_train.shape)

这段代码生成的数据形状是(样本数, 48, 7)和(样本数,)。样本数为什么比原始数据行数少48？因为前48行没有足够的历史数据来构成第一个窗口，这是正常的。seq_len的数值是第一个需要反复调试的超参数，我做过对比实验，在同一个数据集上48小时窗口比24小时的RMSE低大约8%，但再提高到72小时反而轻微变差，说明当前场景下48是比较合理的点。

3. 把LSTM跑起来：核心网络结构与参数怎么定

3.1 先搭一个能用的结构，再谈调参

LSTM网络结构在这个项目里不需要很复杂，单层LSTM加一个全连接输出层已经能拿到合格结果，真正拉开差距的是隐藏层维度和Dropout的设置。常见做法是用PyTorch而不是Keras，虽然两者都能实现，但期末答辩时老师更可能追问反向传播的细节，PyTorch的代码结构更容易把每一步讲清楚。一个可复现的最小结构如下：LSTM层输入维度是7（特征数），隐藏层维度设64，层数设2，后面接一个线性层把隐藏状态映射到1维输出。

import torch import torch.nn as nn class LSTMAQIPredictor(nn.Module): def __init__(self, input_size, hidden_size, num_layers, dropout=0.2): super().__init__() self.lstm = nn.LSTM( input_size=input_size, hidden_size=hidden_size, num_layers=num_layers, batch_first=True, dropout=dropout if num_layers > 1 else 0.0, ) self.fc = nn.Linear(hidden_size, 1) def forward(self, x): out, _ = self.lstm(x) # 取最后一个时间步的隐藏状态作为输出 out = out[:, -1, :] out = self.fc(out) return out

batch_first=True表示输入张量的形状是(batch, seq_len, input_size)，PyTorch默认是(seq_len, batch, input_size)，不设置这个参数的话后续构造数据很容易搞混维度。out[:, -1, :]取的是每个样本最后一个时间步的LSTM输出，因为我们做的是多输入单输出，只需要最后一步的隐藏状态来预测当前AQI。dropout只有在层数大于1时才有意义，如果设置两层LSTM且不启用Dropout，两层之间没有正则化，模型在期末这种小数据集上很容易过拟合，表现为训练Loss很低、测试Loss偏高。

3.2 预测连续多步与预测单步，输入输出完全不同

这一步是最多人搞混的地方。如果题目只要求用过去48小时预测未来1小时的AQI，那么模型每次只需输出一个值，训练时标签是一个标量，调用forward得到的就是(batch, 1)。但如果题目要求预测未来24小时的AQI曲线，就不能让模型直接输出24个值，因为24小时后的特征（PM2.5、风速等）在预测时刻并不存在。常见做法是滚动预测：用已知的48小时数据预测第49小时，然后把第49小时的预测值拼接到输入末尾，丢掉最前面一小时，再预测第50小时，循环24次。

滚动预测会导致误差累积，预测的步数越多，后期偏差越大。为了让模型在滚动预测中不至于快速发散，训练时可以随机截取一段长度为seq_len + k的历史数据，用前seq_len个点预测第seq_len + 1个点，再滚动预测到seq_len + k，用这k步的预测结果和真实值计算Loss。这样模型在训练时就已经适应了用自身预测值作为输入的情况，测试时滚动预测的鲁棒性会好很多。

3.3 四个必调参数：hidden_size、num_layers、learning_rate、seq_len

hidden_size是最值得花时间调的参数。它决定LSTM单元内部记忆容量的大小，数值越大模型表达能力越强，但也越容易记住训练集中的噪声。这个项目的训练样本量通常在几千到几万级别，64通常是最稳的选择。num_layers加层数的收益在数据量大时才明显，小数据上两层足够，三层以上训练时间翻倍但精度几乎不提升，这点在期末项目里很不划算。learning_rate设为0.001配合Adam优化器是PyTorch里最常见的组合，但如果发现Loss前几十轮不降，可以试试0.0003，这个值更稳但收敛略慢。

4. 训练、保存与加载：让源码能跑、能改、能交

4.1 训练循环里必须加早停，别让Loss变成一条平线还得等它跑完

期末项目的训练时间通常不超过半小时，但即便是这么短的时间，也建议加早停机制。早停的作用是监测验证集Loss，连续若干个epoch不再下降时就停止训练并恢复最优权重，避免模型在训练集上继续过拟合。实现方式很简单：每个epoch结束后计算验证集Loss，如果小于历史最优值就保存当前模型参数，否则记录连续未改善的轮数，超过patience就终止训练。

import copy import torch.optim as optim model = LSTMAQIPredictor(input_size=7, hidden_size=64, num_layers=2) optimizer = optim.Adam(model.parameters(), lr=0.001) scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, mode="min", factor=0.5, patience=5) loss_fn = nn.MSELoss() best_val_loss = float("inf") best_state = None patience = 8 no_improve_epochs = 0 for epoch in range(50): model.train() train_loss_total = 0.0 for i in range(0, len(X_train), 64): X_batch = torch.tensor(X_train[i:i+64], dtype=torch.float32) y_batch = torch.tensor(y_train[i:i+64], dtype=torch.float32).unsqueeze(-1) optimizer.zero_grad() pred = model(X_batch) loss = loss_fn(pred, y_batch) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0) optimizer.step() train_loss_total += loss.item() * len(X_batch) model.eval() with torch.no_grad(): X_val_t = torch.tensor(X_val, dtype=torch.float32) y_val_t = torch.tensor(y_val, dtype=torch.float32).unsqueeze(-1) val_pred = model(X_val_t) val_loss = loss_fn(val_pred, y_val_t).item() scheduler.step(val_loss) if val_loss < best_val_loss: best_val_loss = val_loss best_state = copy.deepcopy(model.state_dict()) no_improve_epochs = 0 else: no_improve_epochs += 1 if no_improve_epochs >= patience: print(f"Early stop at epoch {epoch}") break model.load_state_dict(best_state)

clip_grad_norm_这个操作很多初学者会忽略，但它对LSTM训练稳定性非常关键。输入序列较长时，梯度在时间步之间连乘，容易出现梯度爆炸，表现为某个batch的Loss突然变成NaN，之后整个训练过程再也回不来。max_norm=1.0的含义是对所有参数的梯度向量计算整体范数，超过1.0时按比例缩放，这是一个防守型设置，正常训练时几乎不会触发，但能有效防止偶发性的训练崩溃。ReduceLROnPlateau会在验证集Loss停滞时把学习率减半，配合早停使用，可以让模型在训练后段更精细地逼近最优值。

4.2 模型保存与加载：只保权重，不保整个对象

期末考试时最尴尬的情况是训练好的模型没法在自己电脑上二次加载，换一台电脑跑测试脚本直接报错。常见原因是用torch.save(model)保存了完整模型对象，而加载环境的PyTorch版本或者模型类定义路径发生了变化。推荐做法是只保存state_dict，也就是参数字典，然后再实例化一个结构完全相同的模型，用load_state_dict加载参数。这样代码在不同环境下迁移时，只要模型类定义一致，权重文件就能通用。

torch.save(model.state_dict(), "lstm_aqi_best.pt") def load_model(model_path, input_size, hidden_size, num_layers): model = LSTMAQIPredictor(input_size, hidden_size, num_layers) model.load_state_dict(torch.load(model_path, map_location="cpu")) model.eval() return model

map_location="cpu"是一个关键参数，如果训练时用了GPU，保存的权重张量会带有cuda设备信息，加载到没有GPU的机器上会报错。加上这个参数后，无论训练环境是什么，权重都会被加载到CPU上，等模型实例化完成后再手动.to("cuda")迁移即可。model.eval()是加载后必须执行的一步，它会让Dropout层停止工作，BatchNorm层使用训练时统计的均值和方差。忘记调用的话，推理结果会带有随机性，两次运行同一段数据预测结果不完全一样，这在答辩演示时是很尴尬的状况。

4.3 逆变换：预测值要变回真实AQI才有意义

模型输出的值范围在0到1之间，这是在预处理时用MinMaxScaler压缩过的结果。报告里如果直接展示带小数点的缩放宽量，老师一眼就能看出你没有完整地理解数据流。因此在评估和绘图之前，要先把预测值和真实标签都逆变换回真实AQI。需要特别注意的是，逆变换时不能只对AQI单独做，因为MinMaxScaler在fit时是基于全部7列特征计算的，每列的min和max各自独立存储，但scaler.inverse_transform接口期望输入形状是(样本数, 7)。所以更简单的做法是构造一个全零数组，把预测值填到AQI那一列，再调用逆变换，最后取出那一列。

def inverse_aqi(values, scaler, target_idx): dummy = np.zeros((len(values), len(scaler.feature_names_in_))) dummy[:, target_idx] = values.reshape(-1) inverted = scaler.inverse_transform(dummy) return inverted[:, target_idx] target_idx = feature_cols.index("aqi") y_test_real = inverse_aqi(y_test, scaler, target_idx) pred_test_real = inverse_aqi(pred_test, scaler, target_idx)

feature_names_in_是MinMaxScaler在fit时记录的特征名列表，使用它的好处是不会因为手工修改列顺序导致逆变换错位。还有一个容易犯错的地方：在训练过程中计算Loss用的是缩放后的值，RMSE也应该在缩放后的空间计算，还是在逆变换后计算？正确做法是后者，因为报告里需要的RMSE单位是AQI数值，而不是无量纲的0到1区间。两种计算方法得出的数字差异很大，如果报告里不说明清楚，很容易被质疑数据造假。

5. 期末大作业最容易翻车的6个坑与排查清单

5.1 验证集Loss很低，但预测曲线整体右移了一个时间步

现象：画出测试集预测曲线时，预测值和真实值形状非常相似，却有明显的时间滞后，大约刚好差1小时。原因：训练数据构造时把当前时刻的AQI当作标签，但输入特征中也包含了当前时刻的AQI值，模型学到的是直接复制最后一个时间步的输入。解决：检查窗口切分逻辑，输入序列应截止到t-1，标签应为t时刻的AQI，两者时间上必须错开。

5.2 归一化时不小心用了全量数据fit，测试集信息泄漏

现象：用测试集评估时RMSE低得出奇，甚至低于验证集，但把模型应用到新的时间段时准确率大幅下降。原因：预处理时对全量数据调用fit_transform，测试集的数值范围已经被模型间接看到。解决：切分训练集、验证集、测试集后再分别做fit和transform，整个过程写在一个函数里，避免在Notebook中跳跃执行。自查方法：打印scaler.data_min_和data_max_，对比测试集中是否出现过超出该范围的值。

5.3 LSTM输出收敛到一条平线，所有预测值几乎相同

现象：训练Loss下降正常，但预测曲线是一条水平直线，数值接近训练集AQI的均值。原因：隐藏层维度太小或者学习率偏低，模型停留在局部最优，没有学到时间动态特征；另一种可能是Loss函数在缩放后的空间上过于平坦。解决：把hidden_size从32提高到64，学习率从0.0003改为0.001，并且检查输出层是否使用了不适合的激活函数，线性回归头不需要在fc层后面加Sigmoid，因为AQI不是0到1的概率值，Sigmoid会压缩输出范围且梯度容易消失。

5.4 每次训练结果都不一样，怎么调都找不到稳定最优值

现象：同一份源码、同一个数据集，连续跑三次训练，三次的测试RMSE都不一样，波动幅度还很大。原因：没有设置随机种子，PyTorch的参数初始化、数据加载顺序、Dropout行为都带有随机性。解决：在训练脚本开头固定三处随机种子，分别是Python内置的random、NumPy的random和PyTorch的随机数生成器。环境中有CUDA时，还需要额外设置torch.cuda.manual_seed_all。固定随机种子后，同样的代码在同样的环境中应当每次跑出完全一致的结果。

import random import numpy as np def set_seed(seed=42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) set_seed(42)

5.5 滚动预测24小时，前6小时准，之后全偏

现象：单步预测效果不错，但滚动预测越往后误差越大，到第24小时预测值和真实值几乎脱节。原因：滚动预测时模型把前一步的预测值当作输入，误差会逐步累积，这是自回归模型的固有缺陷，不是代码bug。解决：训练阶段加入多步预测Loss策略，即随机选择预测步长，一起计算多步预测的损失；测试阶段把步长上限如实写在报告里，不要只展示单步预测结果。如果题目强制要求24小时预测，报告中要写明误差累积是模型机制导致的，并对比单步与多步的RMSE差异来证明你理解这个限制。

5.6 训练Loss降不下去，数值一直在0.1附近波动

现象：Loss前几个epoch下降很快，之后停滞在0.1左右，无论怎么调学习率都不再下降。原因：最常见的不是模型问题，而是数据预处理时没有处理掉AQI数值突变点，比如某一天AQI从50急升到300，这种剧烈的尖峰会让MSE被极少数样本主导。解决：检查训练集中是否存在比中位数高数倍的异常值，用分位数截断或者对标签做clip处理。另一种情况是特征列中存在大量重复值，比如风速连续几天完全不变，这会抑制梯度的更新，建议对特征做方差过滤，把方差接近零的列剔除。

6. 从“能跑”到“高分”：用好RMSE之外的两件事收官

期末评分通常不是只看Loss数字，报告和答辩演示中能否体现出对模型行为的理解，往往比多调0.1的误差更重要。第一件事是做对比实验，固定数据预处理方式和训练参数，只改一个变量，例如分别用24小时和48小时窗口训练两个模型，记录各自的RMSE和训练时长，用一张对比表呈现。这个对比不需要很复杂，但能证明你做了系统性调参，而不是随便跑了一版。第二件事是绘制的预测曲线图里，除了真实值和预测值，再加上上下5%的误差带，答辩时能直接指出哪些时段模型系统性偏低，比如雾霾积累初期AQI快速上升阶段，预测值通常滞后一到两小时，这说明模型对突变事件的响应能力有限，是LSTM的固有限制，不是代码错误。我每次做这类项目习惯在代码里多加两个注解：一个注释标注数据切分的边界时间点，另一个注释标注逆变换的目标列索引对应关系，这两个点恰好也是老师最喜欢追问的地方。把这两件事做扎实，源码本身的分数就已经稳了，祝你这个期末项目顺利跑通拿高分，希望帮到你。

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