别再只盯着交通流量了！用STGNN搞定城市空气质量预测的完整实战指南（附PyTorch代码）

从气象站到神经网络：用STGNN重构城市空气质量预测的技术革命

清晨推开窗户，你是否曾疑惑今天的空气质量究竟如何？传统预测方法往往只能给出模糊的区域性判断，而现代人工智能技术正在彻底改变这一局面。时空图神经网络（STGNN）的出现，为空气质量预测带来了前所未有的精准度和实时性。本文将带你深入探索如何利用这一前沿技术，构建属于自己的城市空气质量预测系统。

1. 为什么传统方法在空气质量预测上举步维艰

空气质量预测从来就不是一个简单的问题。传统方法主要依赖两种途径：基于物理模型的大气化学传输模拟和基于统计学的时序预测。前者需要超级计算机支持，计算成本高昂；后者则难以捕捉复杂的空间关联性。

以北京市为例，分布在城市各处的35个空气质量监测站每天产生大量数据。传统LSTM模型在处理这些数据时，只能单独预测每个站点的数值，完全忽略了站点之间的空间关联。而实际上，当西北风来临时，位于上风向的监测站数据会显著影响下风向站点的数值变化。

传统方法的三大局限：

• 空间盲区：无法建模监测站点之间的动态影响关系
• 特征单一：难以融合气象、交通等多源异构数据
• 计算僵化：参数调整困难，无法自适应不同城市特性

2019年发表在《Atmospheric Environment》的一项研究显示，传统方法在PM2.5预测上的平均绝对误差（MAE）高达23.4μg/m³，而早期STGNN模型已经能将这一指标降低到15.7μg/m³。

2. 构建空气质量时空图：从原始数据到关系网络

构建合适的时空图是STGNN成功应用的关键。与交通预测不同，空气质量数据的空间关系更加复杂多变，需要综合考虑多种因素。

2.1 数据准备与预处理

典型的空气质量数据集包含以下要素：

import pandas as pd # 示例数据加载 air_quality_data = pd.read_csv('air_quality.csv') meteorological_data = pd.read_csv('weather.csv') # 数据预处理示例 def preprocess_data(df): # 处理缺失值 df = df.interpolate(method='linear') # 标准化 df = (df - df.mean()) / df.std() return df

关键数据源：

2.2 多图融合的时空关系建模

单一的空间关系图无法全面反映空气质量传播的复杂机制。我们提出基于三种关系的复合图结构：

1. 距离图：使用高斯核函数计算站点间的空间衰减效应

   def build_distance_graph(coords, sigma=0.1): n = len(coords) adj = np.zeros((n, n)) for i in range(n): for j in range(n): dist = haversine(coords[i], coords[j]) adj[i,j] = np.exp(-dist**2/(2*sigma**2)) return adj

2. 风向图：根据实时风向构建动态有向图

   def build_wind_graph(coords, wind_dir, wind_speed): # 实现风向影响关系构建 ...

3. 相似图：基于历史数据模式相似性构建

   def build_similarity_graph(data, threshold=0.7): corr = data.corr() return (corr > threshold).astype(float)

提示：实际应用中，这三种图的权重可以通过注意力机制动态调整，以适应不同天气条件下的主导因素变化。

3. STGNN模型实战：从理论到PyTorch实现

在众多STGNN架构中，我们选择改进版的Graph WaveNet作为基础模型，因其出色的自适应图学习能力和长期依赖捕捉特性。

3.1 模型架构设计

我们的定制模型包含以下核心组件：

1. 自适应图学习层：自动发现站点间的隐藏关系

   class AdaptiveGraphLearner(nn.Module): def __init__(self, node_num, dim): super().__init__() self.embed1 = nn.Parameter(torch.randn(node_num, dim)) self.embed2 = nn.Parameter(torch.randn(node_num, dim)) def forward(self): return torch.softmax(F.relu(torch.mm(self.embed1, self.embed2.T)), dim=1)

2. 时空卷积块：整合扩散卷积和门控TCN

   class STConvBlock(nn.Module): def __init__(self, in_channels, spatial_channels, out_channels): super().__init__() self.spatial_conv = DiffusionGraphConv(...) self.temporal_conv = GatedTCN(...) def forward(self, x, adj): x = self.spatial_conv(x, adj) x = self.temporal_conv(x) return x

3. 多图融合模块：动态整合不同关系图的信息

   class MultiGraphFusion(nn.Module): def __init__(self, graph_num, node_num): super().__init__() self.weights = nn.Parameter(torch.ones(graph_num)/graph_num) self.attention = nn.MultiheadAttention(...) def forward(self, graphs): # 加权融合基础图 base_graph = sum(w*g for w,g in zip(self.weights, graphs)) # 注意力增强 attn_graph, _ = self.attention(base_graph, base_graph, base_graph) return 0.5*(base_graph + attn_graph)

3.2 气象特征融合策略

气象因素对空气质量影响显著，我们设计了一种门控融合机制：

class MeteorologyFusion(nn.Module): def __init__(self, air_dim, weather_dim): super().__init__() self.gate = nn.Linear(air_dim + weather_dim, air_dim) self.transform = nn.Linear(weather_dim, air_dim) def forward(self, air_feat, weather_feat): gate = torch.sigmoid(self.gate(torch.cat([air_feat, weather_feat], -1))) return gate * air_feat + (1-gate) * self.transform(weather_feat)

4. 训练技巧与实战调优

在实际项目中，我们发现以下几个技巧能显著提升模型性能：

4.1 多任务学习框架

同时预测多种污染物指标可以提升模型泛化能力：

class MultiTaskHead(nn.Module): def __init__(self, hidden_dim, tasks): super().__init__() self.shared = nn.Sequential( nn.Linear(hidden_dim, hidden_dim//2), nn.ReLU() ) self.heads = nn.ModuleDict({ name: nn.Linear(hidden_dim//2, 1) for name in tasks }) def forward(self, x): shared = self.shared(x) return {name: head(shared) for name, head in self.heads.items()}

4.2 课程学习策略

逐步增加预测时间跨度，让模型先学会短期预测，再挑战长期预测：

def curriculum_schedule(epoch): if epoch < 10: return 3 # 预测未来3小时 elif epoch < 20: return 6 # 预测未来6小时 else: return 12 # 预测未来12小时

4.3 损失函数设计

结合多种损失函数的优势：

def hybrid_loss(pred, target): mae = F.l1_loss(pred, target) mse = F.mse_loss(pred, target) # 加入峰值误差惩罚 peak_err = F.l1_loss(pred[target > 75], target[target > 75]) return 0.5*mae + 0.3*mse + 0.2*peak_err

典型训练参数配置：

5. 部署落地与持续优化

将STGNN模型投入实际生产环境需要考虑以下关键因素：

5.1 边缘-云端协同推理

为平衡实时性和计算成本，我们采用分层处理架构：

1. 边缘设备：运行轻量级模型处理实时数据
2. 云端服务器：定期运行完整模型进行校正
3. 结果融合：使用卡尔曼滤波整合不同来源的预测

5.2 模型解释性增强

通过以下方法提升模型透明度：

def analyze_importance(model, data): # 计算图注意力权重 attn_weights = model.get_attention_weights() # 特征重要性分析 feature_importance = calculate_shap_values(model, data) return attn_weights, feature_importance

5.3 持续学习机制

设计模型更新策略应对数据分布变化：

class ContinualLearner: def __init__(self, base_model): self.model = base_model self.memory = DataBuffer(capacity=10000) def update(self, new_data): self.memory.add(new_data) # 从记忆库中采样进行微调 batch = self.memory.sample(batch_size=32) loss = train_step(self.model, batch) return loss

在实际部署中，我们遇到的一个典型挑战是春季沙尘暴期间的预测异常。通过分析发现，模型对这类罕见事件学习不足。解决方案是引入对抗样本生成技术，主动合成极端天气条件下的模拟数据增强训练集。