简介：这份资源是一份面向环保科技从业者、数据分析师与大数据工程师的空气质量分析与预测系统技术文档，围绕机器学习与时间序列分析在环境科学中的应用展开，适合需要定期发布空气质量报告、开展环境政策研究或制定城市发展规划的政府机构与企业单位参考。压缩包内共1个docx文件，约595KB，内容涵盖系统需求分析、Django框架下的B/S架构设计、MySQL数据库E-R建模，以及斯皮尔曼相关性系数、贝叶斯算法等预测模型的实现细节，并配有大量案例研究与技术说明。已有162人学习下载，可作为同类项目的参考模板。读者可从中获取从数据采集、清洗预处理到模型训练与部署的完整流程思路，理解随机森林、支持向量机等算法在空气质量指标预测中的具体用法，并借鉴性能评估与改进方向，但需注意结合不同地区实际情况调整参数设置。
1. 从一张超标罚单说起：空气质量监测与预测系统到底在解决什么
去年冬天，一个做园区环保信息化的朋友找我，说他们装的几台微型空气质量监测站天天报警，PM2.5 一超标就推短信，结果运维人员跑过去一看，现场根本没异味，风一吹数值又掉下来了。问题不在传感器坏，而在于「监测」和「预测」被混为一谈：设备只告诉你此刻是多少，却没人告诉你未来两小时会不会持续升高、要不要提前让产线降负荷。这就是基于机器学习的空气质量监测与预测系统要补的那一环——用历史与实时的多污染物、气象数据，训练出能滚动输出未来浓度区间的模型，把「事后报警」变成「事前调度」。
这套系统适合三类人：一是做环保物联网平台、手里已经有监测数据的工程师；二是想拿一个完整机器学习项目练手的学生和转行者，空气质量数据公开、特征清晰、业务闭环完整，比很多玩具数据集更接近真实工程；三是园区、厂区里负责能耗与排放联动的运维人员。它不追求把预报做到气象台级别，而是追求在本地几十米到几公里尺度上，比「拍脑袋」和「单点阈值」更靠谱。下面我按自己搭过的一版方案，把数据、特征、模型、部署和踩过的坑讲清楚。
2. 数据从哪来、特征怎么造：空气质量监测的数据管线
2.1 监测数据的三个来源与选型取舍
做空气质量预测，第一步不是选模型，而是确定数据源。常见做法有三类：国控/省控站点的公开小时数据、自建微型站的分钟级数据、以及气象再分析或本地气象站数据。公开站点数据质量高、时间跨度长，但空间分辨率粗，一个城市可能只有几个点；微型站密度高，却容易受湿度、温度漂移影响，PM2.5 在湿度大时读数虚高是行业里公认的玄学问题。我的建议是：训练阶段用公开站点数据打底，保证标签可靠；上线阶段用微型站做空间插值补充，但必须做湿度校正。
选型时重点看三件事：时间分辨率是否统一到小时、缺失值比例是否低于 15%、是否有同步的气象字段（温度、湿度、风速、风向、气压）。如果只有污染物没有气象，风速一高浓度就掉，模型会学成「风速大就干净」的伪相关，换季就翻车。
2.2 用 pandas 把多源数据对齐成一张宽表
真实数据一定是脏的：时间戳时区不一致、整点缺测、单位混用。下面这段是我常用的对齐脚本，核心是把污染物和气象按小时外连接到同一张表，再做缺失标记。
import pandas as pd import numpy as np # 污染物数据：time, pm25, pm10, no2, so2, co, o3 poll = pd.read_csv("pollutant.csv", parse_dates=["time"]) # 气象数据：time, temp, rh, wind_speed, wind_dir, pressure met = pd.read_csv("meteo.csv", parse_dates=["time"]) # 统一到整点小时，避免分钟级抖动干扰 poll["time"] = poll["time"].dt.floor("h") met["time"] = met["time"].dt.floor("h") df = pd.merge(poll, met, on="time", how="outer").sort_values("time") df = df.set_index("time").asfreq("h") # 补齐缺失小时 # 缺失标记：告诉模型这一格是补出来的，不是真实观测 for col in ["pm25", "pm10", "no2", "temp", "rh", "wind_speed"]: df[f"{col}_isna"] = df[col].isna().astype(int) # 线性插值只补短缺口，长缺口保留 NaN 交给模型或丢弃 df["pm25"] = df["pm25"].interpolate(method="linear", limit=3) df["temp"] = df["temp"].interpolate(method="linear", limit=3) df = df.dropna(subset=["pm25"]) # 标签缺失的样本不能用于训练 df.to_csv("air_aligned.csv")逻辑说明：asfreq("h")保证时间轴连续，否则后面做滞后特征会错位；limit=3是关键参数，只补连续 3 小时以内的缺口，超过就保留 NaN，因为长缺口插值等于编数据。_isna标记列让模型知道哪些值是补的，树模型能利用这个信息降低对补值的信任。参数上，如果数据是分钟级，先聚合到小时再对齐，不要直接对分钟做插值，否则噪声会被放大。
2.3 滞后特征与滚动统计：让模型看到「趋势」
空气质量有强自相关，当前 PM2.5 和过去几小时高度相关，所以滞后特征是收益最高的一类特征。我一般会造 1、3、6、12、24 小时滞后，再加 6 小时和 24 小时滚动均值、滚动标准差。滚动标准差能刻画「波动剧烈程度」，沙尘或烟花时段这个值会突然变大，是很好的预警信号。
def add_lag_features(df, col="pm25", lags=(1, 3, 6, 12, 24)): for lag in lags: df[f"{col}_lag{lag}"] = df[col].shift(lag) df[f"{col}_roll6_mean"] = df[col].shift(1).rolling(6).mean() df[f"{col}_roll24_std"] = df[col].shift(1).rolling(24).std() return df df = add_lag_features(df) df["target_pm25_next3h"] = df["pm25"].shift(-3) # 预测未来3小时 df = df.dropna()注意shift(1)再 rolling，是为了避免把当前时刻的值算进滚动窗口，否则就是标签泄漏，线下指标好看、上线直接崩。预测目标我选未来 3 小时，是因为太短没有调度价值，太长误差又压不住，3 小时是园区调度能接受的窗口。
3. 模型怎么选、怎么训：从线性回归到梯度提升的落地路径
3.1 基线、树模型与序列模型的取舍
别一上来就上 LSTM。我踩过的坑是：数据量不到两年、特征工程没做透时，LSTM 训练慢、调参玄学、线上推理还要维护状态，收益还不如 LightGBM。合理的路径是先用线性回归或 Ridge 做基线，确认特征方向对不对；再用 LightGBM 或 XGBoost 做主模型，它们对缺失值、异常值鲁棒，训练快，特征重要性可解释；只有当数据超过三五年、且需要多步长序列输出时，再考虑 LSTM 或 Temporal Fusion Transformer。
选型判断标准很实在：如果运维团队要能看懂「为什么今天报高」，树模型的特征重要性直接能解释；如果只是追求榜单指标，序列模型可能略高，但工程成本翻倍。我一般会两个都训，用验证集对比，差距在 5% 以内就选树模型。
3.2 用 LightGBM 训练一个可解释的预测模型
下面是我常用的训练脚本，重点是时间序列切分不能随机打乱，否则未来信息会泄漏到训练集。
import lightgbm as lgb from sklearn.metrics import mean_absolute_error features = [c for c in df.columns if c not in ["pm25", "target_pm25_next3h"]] X, y = df[features], df["target_pm25_next3h"] # 时间序列切分：前80%训练，后20%验证，绝不 shuffle split = int(len(df) * 0.8) X_train, X_val = X.iloc[:split], X.iloc[split:] y_train, y_val = y.iloc[:split], y.iloc[split:] model = lgb.LGBMRegressor( n_estimators=800, learning_rate=0.03, num_leaves=31, min_child_samples=20, subsample=0.8, colsample_bytree=0.8, objective="regression_l1" # MAE 对极端值更稳 ) model.fit( X_train, y_train, eval_set=[(X_val, y_val)], eval_metric="l1", callbacks=[lgb.early_stopping(50)] ) pred = model.predict(X_val) print("MAE:", mean_absolute_error(y_val, pred))参数说明：learning_rate=0.03配n_estimators=800是慢学稳收敛的组合，num_leaves=31控制复杂度防止过拟合，min_child_samples=20保证叶子节点有足够样本。objective="regression_l1"用 MAE 而不是 MSE，是因为空气质量数据里偶发重污染极值会拉偏 MSE，导致模型整体偏高。early_stopping(50)在验证集 50 轮不降就停，省得手动试轮数。
3.3 评估指标不能只看 MAE
MAE 只告诉你平均差多少，但业务关心的是「超标那几小时有没有报准」。我一般会补两个指标：一是超标召回率，即真实 PM2.5 超过阈值时模型预测也超的比例；二是分位数损失，看模型对高值的低估程度。如果超标召回率低于 70%，说明模型在极端时段偏保守，需要给高值样本加权或单独训一个分类器做「是否超标」的二分类，再和回归结果融合。
import numpy as np thr = 75 # 示例阈值，按当地标准调整 mask = y_val > thr recall = ((pred > thr) & mask).sum() / max(mask.sum(), 1) print("超标召回率:", recall)这个召回率才是运维真正看的数字。MAE 降 1 微克但召回率掉 10 个点，对业务是负优化。
4. 从离线模型到在线服务：部署与实时预测的工程细节
4.1 特征一致性：离线训练和在线推理必须同一套代码
上线最常见的翻车是「训练时特征算一套，线上算另一套」。比如训练用 pandas rolling，线上用 SQL 窗口函数，边界处理不一致，结果线上预测系统性偏移。我的做法是把特征计算封装成一个函数，离线和在线都调它，在线时把最近 N 小时数据拼成 DataFrame 传入。
def build_features(recent_df): # recent_df 至少包含最近 48 小时，按时间升序 df = add_lag_features(recent_df.copy()) return df[features].iloc[[-1]] # 只取最新一行做推理 # 在线服务伪代码 latest = fetch_last_48h() # 从时序库取数 X_now = build_features(latest) y_hat = model.predict(X_now)[0]关键是fetch_last_48h必须和训练数据的清洗规则一致：同样的单位、同样的缺失处理、同样的时区。任何一处不同，模型都会「认不出」输入。
4.2 用 FastAPI 暴露预测接口并加缓存
在线服务我一般用 FastAPI，轻量、异步、好接监控。加一层缓存是因为同一小时内多次请求结果应该一致，没必要重复推理。
from fastapi import FastAPI import time app = FastAPI() cache = {"ts": 0, "value": None} @app.get("/predict/pm25") def predict(): now = int(time.time() // 3600) if cache["ts"] == now and cache["value"] is not None: return {"pm25_next3h": cache["value"], "cached": True} latest = fetch_last_48h() value = float(model.predict(build_features(latest))[0]) cache.update(ts=now, value=value) return {"pm25_next3h": value, "cached": False}参数上，缓存按整点小时失效，和监测数据更新频率对齐。返回值里带cached字段，方便排查是实时算的还是缓存命中。接口不要返回裸数组，带上时间戳和单位，前端和调度系统才不会用错。
4.3 监控与回滚：模型上线不是终点
上线后必须监控三件事：输入特征分布是否漂移、预测值与实测值的滚动偏差、接口延迟。我一般会每天算一次过去 24 小时的 MAE，如果连续三天比验证集 MAE 高出 50%，就触发告警并回滚到上一版模型。漂移检测用简单的 PSI 或均值对比即可，不必上复杂框架。回滚要能一键切，所以模型文件按版本号存，配置文件指向当前版本，别把模型路径写死在代码里。
5. 避坑与排查：空气质量预测里最容易翻车的五件事
5.1 现象：线下 MAE 很低，上线后预测值整体偏高
原因：训练集和线上数据的缺失值处理不一致，线上把缺失填成了 0，而 0 在污染物里是「极干净」的强信号，模型被带偏。解决：统一用同一套清洗函数，缺失保留 NaN 或填成训练集均值，并带上_isna标记。
5.2 现象：模型在冬季表现好，夏季一塌糊涂
原因：臭氧在夏季是主导污染物，而训练特征里只有 PM2.5 的滞后项，没有 O3 和温度交互。解决：把 O3、温度、日照时长纳入特征，或按季节分别训练子模型。我一般会加一个「月份」或「季节」类别特征，让树模型自己分裂。
5.3 现象：超标时段总是报不出来
原因：重污染样本占比低，回归模型被大量清洁样本主导，倾向于预测中间值。解决：对高值样本加权，或单独训一个超标二分类器，两个结果融合。加权时权重别超过 5 倍，否则模型会对噪声过拟合。
5.4 现象：接口偶尔超时，日志显示推理很慢
原因：每次请求都重新读全量历史数据算特征，IO 和计算都重。解决：把特征计算拆成定时任务，每小时预计算好最新特征存起来，接口只做推理。预计算和实时计算必须共用同一函数，避免不一致。
5.5 现象：风向特征用了数值编码，模型学不出规律
原因：风向是角度，359 度和 1 度实际很近，但数值差很大。解决：把风向拆成sin和cos两个分量，或者转成 16 方位类别。这个坑很隐蔽，但改完通常能带来几个点的提升。
6. 把预测接进调度：阈值联动与一个可复用的验证习惯
模型训完只是半成品，真正产生价值的是把它接进业务动作。我一般会设两级阈值：预测未来 3 小时 PM2.5 超过 75 触发「预警」，超过 115 触发「调度」，调度动作可以是通知产线降负荷、开启喷淋或调整通风。阈值不要写死在代码里，放配置文件，因为不同季节、不同园区标准不一样。联动逻辑要加「迟滞」：连续两次预测超标才触发，避免单次抖动导致频繁调度，这就是控制里的防抖。
验证习惯上，我坚持做「回测 + 影子运行」两步。回测是把模型在过去半年的数据上滚动预测，看每个月的 MAE 和召回率是否稳定；影子运行是上线后先只记录预测不触发动作，跑两周对比实测，确认没有系统性偏差再打开联动。这两步能挡掉大部分「线下好、线上崩」的问题。
def decide(pred, prev_pred, cfg): if pred > cfg["dispatch_thr"] and prev_pred > cfg["dispatch_thr"]: return "dispatch" if pred > cfg["warn_thr"]: return "warn" return "normal"这个函数里prev_pred就是迟滞用的上一次预测，cfg从配置文件读。别小看这几行，它决定了调度系统会不会被噪声折腾到没人信。
最后说个我自己的教训：我早期总想把模型指标刷到最好，后来发现运维只关心「报准的那几次有没有用」。与其花两周把 MAE 从 12 降到 11，不如花两天把超标召回率从 65% 提到 80%，后者才是这个系统值不值得做的分水岭。希望帮到你。
本文还有配套的精品资源，点击获取
