基于机器学习的空气质量监测与预测系统：从数据采集到模型部署的完整工程闭环

简介这份资源是一份面向环保科技从业者、空气污染研究人员、数据分析师与大数据工程师的技术文档围绕空气质量监测与预测系统的完整构建流程展开涉及数据挖掘、时间序列分析与预测建模等方向。内容从多源数据采集、清洗与预处理入手结合统计分析、斯皮尔曼相关性系数、贝叶斯算法以及随机森林、支持向量机等机器学习方法对AQI、PM2.5、PM10、SO₂等指标进行预测并给出系统架构、数据库E-R设计、模型训练与部署的实施方案同时讨论性能评估与改进方向。资源包内共1个docx文件约595KB以开题报告形式呈现包含需求分析、系统用例图、功能模块图、MTV模式说明及算法公式推导可作为类似项目的参考模板。目前已有162人学习适合需要定期发布空气质量报告、开展环境政策研究或制定城市发展规划的政府机构与企业单位参考读者可据此了解从数据到预测的完整技术链路并根据不同地区实际情况调整参数设置。1. 空气质量监测与预测系统从数据采集到模型落地的完整路径很多团队做空气质量项目第一反应是买传感器、搭看板结果数据飘得厉害模型跑出来还不如 persistence 基线。问题不在算法在于整条链路没打通传感器漂移没校准、缺失值用均值填充、特征里混入了未来信息、模型上线后没人监控衰减。基于机器学习的空气质量监测与预测系统核心不是某个花哨的模型而是把「采集—清洗—特征—建模—部署—迭代」串成可复现的工程闭环。它适合有基础 Python 能力、手头有监测站点数据或打算自建采集节点的开发者也适合需要给环保/园区/城市管理做预测预警的技术负责人。下面按落地顺序拆开讲每一步都给可抄的代码和参数。2. 数据采集与预处理把原始监测数据变成能喂给模型的表2.1 监测数据的来源与字段设计常见做法是三类数据源国控/省控站点公开 APIPM2.5、PM10、SO2、NO2、CO、O3、AQI、自建低成本传感器节点电化学激光散射、气象数据温度、湿度、风速、风向、气压。我一般会统一成一张宽表时间粒度对齐到小时字段命名固定为timestamp, station_id, pm25, pm10, so2, no2, co, o3, temp, humi, wind_speed, wind_dir, pressure。注意风向要做 sin/cos 分解否则 359° 和 1° 在数值上差很远但实际接近树模型会学出玄学分裂。采集频率建议 5 分钟原始值聚合到小时均值再入模。原因污染物浓度短时波动大小时均值既能降噪又和大多数公开预报口径一致。如果做实时预警可以保留 5 分钟粒度单独跑一套阈值规则不要和小时模型混用。2.2 缺失值、异常值与传感器漂移的处理缺失值不要直接fillna(mean)。污染物浓度有强自相关性正确做法是按站点做时间序列插值再对长缺口用同区域邻近站点回归填充。异常值用物理边界加滚动分位数双重判断PM2.5 超过 1000 或为负直接置缺滚动 24 小时中位数 ±3 倍 MAD 之外的标记为可疑。传感器漂移是自建节点最大的坑。低成本 PM 传感器受湿度影响大湿度高于 80% 时读数虚高。我一般会加一个湿度校正pm25_corr pm25_raw / (1 0.25 * max(0, humi - 60) / 40)系数用同期国控站点做线性回归拟合。没有国控对照的至少每季度用标准膜采样做一次现场比对。import pandas as pd import numpy as np def clean_air_quality(df, station_colstation_id, time_coltimestamp): df df.sort_values([station_col, time_col]).copy() # 物理边界置缺 for col, low, high in [(pm25, 0, 1000), (pm10, 0, 1500), (so2, 0, 2000), (no2, 0, 2000), (co, 0, 50), (o3, 0, 1000)]: df.loc[(df[col] low) | (df[col] high), col] np.nan # 按站点时间插值限制最大连续插补 3 小时 df df.groupby(station_col).apply( lambda g: g.set_index(time_col).interpolate( methodtime, limit3, limit_directionboth ).reset_index() ).reset_index(dropTrue) # 滚动 MAD 异常标记 for col in [pm25, pm10, so2, no2, co, o3]: med df.groupby(station_col)[col].transform( lambda s: s.rolling(24, min_periods6).median()) mad df.groupby(station_col)[col].transform( lambda s: (s - s.rolling(24, min_periods6).median()).abs().rolling(24, min_periods6).median()) df.loc[(df[col] - med).abs() 3 * mad, col] np.nan return df逻辑说明先按站点和时间排序保证插值方向正确物理边界用领域常识卡死避免传感器故障值污染interpolate(methodtime)按时间间隔加权比线性索引插值更合理limit3防止长缺口被过度平滑MAD 比标准差抗异常值适合污染物这种尖峰分布。参数上limit根据数据完整度调一般 2~6 小时MAD 倍数 3 是常用起点漏报多就降到 2.5误报多就升到 3.5。2.3 特征工程时间、气象与滞后项特征分四组时间特征小时、星期、月份、是否节假日、气象特征温湿度、风速风向分解、气压、滞后特征过去 1/3/6/12/24 小时的污染物浓度、滚动统计过去 6/12/24 小时均值、最大、最小。滞后和滚动特征必须用shift构造确保预测 t 时刻时只用到 t-1 及之前的信息。很多翻车案例就是滚动窗口没 shift把当前时刻的真实值漏进去了离线指标漂亮上线直接崩。def build_features(df, targetpm25, lags(1, 3, 6, 12, 24)): df df.sort_values([station_id, timestamp]).copy() df[hour] df[timestamp].dt.hour df[dow] df[timestamp].dt.dayofweek df[month] df[timestamp].dt.month df[is_weekend] (df[dow] 5).astype(int) df[wind_dir_sin] np.sin(np.deg2rad(df[wind_dir])) df[wind_dir_cos] np.cos(np.deg2rad(df[wind_dir])) for lag in lags: df[f{target}_lag{lag}] df.groupby(station_id)[target].shift(lag) for win in (6, 12, 24): df[f{target}_roll{win}_mean] df.groupby(station_id)[target].transform( lambda s: s.shift(1).rolling(win, min_periodswin // 2).mean()) df[f{target}_roll{win}_max] df.groupby(station_id)[target].transform( lambda s: s.shift(1).rolling(win, min_periodswin // 2).max()) return df.dropna(subset[f{target}_lag{max(lags)}])逻辑说明groupby(station_id)保证不同站点不串数据shift(1)是防泄漏的关键滚动统计先 shift 再 rollingmin_periods设窗口一半避免开头大量缺失。参数上滞后阶数看自相关函数一般 24 小时覆盖日周期滚动窗口 6/12/24 对应短中长趋势。如果做多步预测目标也要按步长 shift别只 shift 特征。3. 模型选型与训练从线性回归到梯度提升的取舍3.1 基线模型为什么必须先跑上来就上 LSTM 或 Transformer 是常见误区。空气质量预测里t-1时刻的 PM2.5 本身就是极强预测因子persistence 基线直接用上一小时值在很多站点能拿到 R² 0.7 以上。先跑线性回归和 persistence你才知道后面的复杂模型到底有没有增量价值。我一般用三个基线persistence、线性回归带滞后特征、随机森林。只有梯度提升或深度学习稳定超过基线 10% 以上 RMSE 改善才值得上。3.2 梯度提升树与超参数设置XGBoost/LightGBM 在表格型空气质量数据上通常是性价比最高的选择。关键参数n_estimators500~2000 配合early_stopping_rounds50learning_rate0.03~0.1max_depth4~8太深容易记住站点噪声subsample0.7~0.9colsample_bytree0.7~0.9min_child_weight5~20控制叶节点样本数。时间序列不能随机划分必须按时间切前 70% 训练中间 15% 验证调参最后 15% 测试。否则验证集泄漏调出来的参数上线就废。import lightgbm as lgb from sklearn.metrics import mean_squared_error, mean_absolute_error def time_split(df, time_coltimestamp, ratios(0.7, 0.15, 0.15)): df df.sort_values(time_col) n len(df) train_end int(n * ratios[0]) val_end int(n * (ratios[0] ratios[1])) return df.iloc[:train_end], df.iloc[train_end:val_end], df.iloc[val_end:] feature_cols [c for c in df.columns if c not in [timestamp, station_id, pm25, pm10, so2, no2, co, o3]] train, val, test time_split(df) model lgb.LGBMRegressor( n_estimators2000, learning_rate0.05, max_depth6, subsample0.8, colsample_bytree0.8, min_child_weight10, objectiveregression, metricrmse, random_state42) model.fit(train[feature_cols], train[pm25], eval_set[(val[feature_cols], val[pm25])], eval_metricrmse, callbacks[lgb.early_stopping(50), lgb.log_evaluation(100)]) pred model.predict(test[feature_cols]) print(RMSE, mean_squared_error(test[pm25], pred, squaredFalse)) print(MAE, mean_absolute_error(test[pm25], pred))逻辑说明time_split按时间顺序切杜绝未来信息泄漏early_stopping在验证集 RMSE 不再下降时停防止过拟合min_child_weight调大能抑制站点特异噪声。参数上learning_rate和n_estimators反向搭配0.05 配 2000 是稳妥起点max_depth超过 8 在站点数据上几乎必过拟合。评估除了 RMSE还要看分位数误差重污染时段的低估往往比整体 RMSE 更致命。3.3 深度学习什么时候值得上只有当你有多个城市、多年、多站点的大规模数据比如几十万条以上且需要做多步或多站点联合预测时LSTM/TCN/Transformer 才可能超过 GBDT。小数据上深度学习调参成本高、收益不稳定。如果上输入窗口一般 24~72 小时输出 1~24 小时损失函数用 Huber 或分位数损失别用纯 MSE否则对极端污染事件不敏感。训练时按站点做归一化别全局归一化否则不同浓度水平的站点会被拉平。4. 部署与监控模型上线后怎么不变成黑匣子4.1 推理服务与定时任务常见做法是 FastAPI 包一层/predict接口输入最近 24 小时特征输出未来 1~24 小时浓度。定时任务用 APScheduler 或 cron 每小时拉最新数据、构造特征、调模型、写回数据库。模型文件用 joblib 或 LightGBM 原生格式保存别用 pickle 跨版本容易翻车。接口要做输入校验字段缺失、时间戳不连续、数值越界直接返回 400别让脏数据进模型。from fastapi import FastAPI, HTTPException from pydantic import BaseModel import joblib import numpy as np app FastAPI() model joblib.load(lgb_pm25.pkl) FEATURES [...] # 与训练时顺序一致 class PredictRequest(BaseModel): station_id: str features: dict app.post(/predict) def predict(req: PredictRequest): missing [f for f in FEATURES if f not in req.features] if missing: raise HTTPException(400, fmissing features: {missing}) x np.array([[req.features[f] for f in FEATURES]]) if not np.isfinite(x).all(): raise HTTPException(400, non-finite feature value) y model.predict(x)[0] return {station_id: req.station_id, pm25_pred: float(y)}逻辑说明FEATURES顺序必须和训练时完全一致否则预测结果会静默错位输入校验挡住缺失和非有限值返回 float 避免 numpy 类型序列化问题。参数上接口超时设 2~5 秒模型推理本身通常毫秒级瓶颈在特征查询。定时任务失败要有重试和告警别静默丢一小时。4.2 监控指标与模型衰减上线后盯三个指标推理延迟、输入特征分布漂移PSI 或 KS、预测误差滚动均值。特征漂移比误差更早暴露问题比如传感器校准后 PM2.5 整体下移模型没重训就会系统性高估。我一般每周算一次 PSI超过 0.2 就触发重训评估。误差监控用最近 7 天实际值回填后的 MAE比训练集 MAE 涨 30% 以上就要查原因。重训频率看数据变化速度一般每月一次全量重训每周增量更新。5. 避坑与排查空气质量预测里最容易翻车的 5 个点现象离线 R² 0.95上线 R² 不到 0.5。原因特征构造时滚动窗口没 shift当前时刻目标值泄漏进特征。解决所有滚动和滞后特征统一先shift(1)再计算用时间切分验证别用随机切分。现象模型在重污染天严重低估。原因训练集里重污染样本少MSE 损失被正常样本主导。解决对高浓度样本加权或改用 Huber/分位数损失评估时单独看 PM2.5 150 区间的 MAE。现象不同站点预测精度差异巨大。原因全局归一化或全局模型没区分站点类型城市/背景/工业。解决按站点做归一化或把站点 ID 做 target encoding再或者分站点建模型。现象传感器湿度高时 PM2.5 虚高模型学出错误关联。原因原始数据没做湿度校正模型把湿度和高 PM2.5 绑定。解决入模前做湿度校正或把湿度及其交互项显式加入特征让模型自己学校正。现象接口偶尔返回 NaN 或报错。原因特征查询时某小时数据缺失插值后仍有 NaN或特征顺序与训练不一致。解决接口层做有限性校验和字段顺序校验特征管道里对剩余 NaN 用站点中位数兜底并记录日志。6. 进阶技巧用分位数回归给出预测区间点预测不够用环保预警需要「大概率不超过多少」。把 LightGBM 的objective换成quantilealpha设 0.5/0.9/0.95 分别训练三个模型就能给出中位数和上界。上界比点预测更适合触发预警能减少漏报。训练时每个分位数单独 early stopping评估用 pinball loss。我一般还会把 0.9 分位数的预测值做 3 小时滚动平滑避免单小时抖动导致误报。这套做法在园区预警里比单点模型实用得多代价是训练和推理成本翻三倍但换来的是可解释的风险边界。希望帮到你。本文还有配套的精品资源点击获取