3. 数据采集与预处理：环境监测数据来源与清洗实战
各位同学，今天我们来聊聊环境评估中最磨人、也最关键的一步——数据采集与预处理。说实话，我见过太多模型跑得飞起，结果一查数据全是垃圾的案例。数据质量不行，再牛的算法也是白搭。
我个人习惯把数据预处理比作「做饭前的洗菜切菜」。菜不洗干净，厨艺再好也白费。咱们环境监测的数据更是如此，卫星、传感器、气象站，来源五花八门，格式千奇百怪。今天我就把这些年踩过的坑、总结的经验，一股脑倒给你们。
3.1 环境监测数据的主要来源
环境数据从哪来？说白了就三大类：天上看、地上测、中间站。
3.1.1 卫星遥感数据
卫星数据覆盖面广，但分辨率有限。我记得2018年做长三角PM2.5溯源时，用的就是MODIS的AOD产品。卫星数据有个特点——它不直接给你污染物浓度，给的是气溶胶光学厚度，你得自己反演。
- 常见卫星平台： Landsat（30m分辨率）、MODIS（250m-1km）、Sentinel系列（10m-60m）
- 常用产品： NDVI植被指数、AOD气溶胶、LST地表温度
- 获取方式： USGS EarthExplorer、NASA LAADS DAAC、欧空局Copernicus
3.1.2 地面传感器数据
传感器数据是环境监测的「地面真相」。我在项目里用过PM2.5激光散射传感器、电化学气体传感器、温湿度探头。这些家伙便宜但娇气，动不动就漂移、中毒。
- 常见传感器类型： 颗粒物传感器（PM2.5/PM10）、气体传感器（SO₂、NO₂、O₃、CO）、气象传感器（风速风向、温湿度、气压）
- 数据频率： 从1秒到1小时不等，看应用场景
- 通信方式： 4G/NB-IoT/LoRa，我建议用NB-IoT，功耗低、覆盖好
3.1.3 气象站数据
气象站数据是环境模型的「背景场」。风速风向直接影响污染物扩散，温湿度影响化学反应速率。我一般从中国气象数据网或NOAA的NCDC获取。
| 数据来源 | 典型参数 | 时间分辨率 | 空间覆盖 |
|---|---|---|---|
| 国家气象站 | 温度、湿度、风速、气压 | 1小时 | 全国约2400站 |
| 区域自动站 | 降水、温度、风向 | 10分钟 | 乡镇级 |
| 探空站 | 垂直廓线（温压湿风） | 12小时 | 全国约120站 |
3.2 数据清洗——别让脏数据毁了你的模型
数据清洗，说白了就是「去伪存真」。我见过最离谱的数据——某传感器连续三个月读数都是25.0，纹丝不动。这明显是死机了，但如果不处理，模型会以为这个地方空气质量好得离谱。
3.2.1 格式统一与时间对齐
不同来源的数据格式五花八门。卫星数据是HDF或NetCDF，传感器是CSV或JSON，气象站可能是TXT。第一步，全部转成统一的DataFrame格式。
import pandas as pd
import numpy as np
# 读取传感器CSV数据
sensor_df = pd.read_csv('sensor_data.csv', parse_dates=['timestamp'])
# 读取气象站数据（固定宽度格式）
met_df = pd.read_fwf('met_data.txt',
widths=[4,2,2,6,6,6],
names=['year','month','day','temp','humid','wind'])
# 时间对齐到整点
sensor_df['hour'] = sensor_df['timestamp'].dt.floor('H')
met_df['datetime'] = pd.to_datetime(met_df[['year','month','day']])
met_df['hour'] = met_df['datetime'].dt.floor('H')
# 合并
merged_df = pd.merge(sensor_df, met_df, on='hour', how='left')
3.2.2 重复值与异常时间戳
传感器网络经常出现重复上报。比如信号不好时，设备会缓存数据，等信号恢复后一股脑发过来。这时候就会出现同一时间戳多条记录。
# 检查重复
print(f"重复行数: {merged_df.duplicated(subset=['hour','station_id']).sum()}")
# 去重策略：取平均值或保留第一条
merged_df = merged_df.drop_duplicates(subset=['hour','station_id'], keep='first')
# 检查时间连续性
time_gaps = merged_df['hour'].diff().dt.total_seconds()
print(f"最大时间间隔: {time_gaps.max()/3600:.1f}小时")
3.3 缺失值处理——别乱填，要讲道理
缺失值处理是门艺术。我见过有人直接删掉所有含NaN的行，结果数据量少了三分之一。也见过有人用0填充，直接把模型带偏了。
3.3.1 缺失模式分析
先搞清楚缺失是随机的，还是系统性的。比如夜间传感器供电不足导致数据缺失，这就是系统性的，不能简单用均值填充。
# 可视化缺失模式
import missingno as msno
msno.matrix(merged_df.sample(1000))
# 如果看到明显的条纹，说明是系统性缺失
3.3.2 填充策略选择
| 缺失类型 | 推荐方法 | 适用场景 |
|---|---|---|
| 随机缺失（<5%） | 线性插值 | 温度、湿度等连续变量 |
| 连续缺失（<3小时） | 前向填充+后向填充 | 风速、风向 |
| 长时间缺失 | KNN插补或模型预测 | PM2.5、气体浓度 |
| 空间缺失 | 反距离加权（IDW） | 多个监测站之间的插值 |
# 线性插值（适合短时间缺失）
merged_df['pm25_interp'] = merged_df['pm25'].interpolate(method='linear', limit=3)
# 前向填充（适合传感器短暂离线）
merged_df['pm25_ffill'] = merged_df['pm25'].fillna(method='ffill', limit=6)
# KNN插补（适合长时间缺失）
from sklearn.impute import KNNImputer
imputer = KNNImputer(n_neighbors=5)
features = ['pm25','temp','humid','wind']
merged_df[features] = imputer.fit_transform(merged_df[features])
3.4 异常值检测——揪出那些「离谱」的数据
异常值检测，说白了就是找那些「不合理」的数据点。比如PM2.5突然飙到1000，但周围站点都只有50，这明显是传感器故障。
3.4.1 基于统计的方法
最常用的是Z-score和IQR。但要注意，环境数据往往不是正态分布，直接用Z-score会误判。
# IQR方法（对偏态分布更鲁棒）
Q1 = merged_df['pm25'].quantile(0.25)
Q3 = merged_df['pm25'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR
outliers = merged_df[(merged_df['pm25'] < lower_bound) |
(merged_df['pm25'] > upper_bound)]
print(f"检测到 {len(outliers)} 个异常值")
3.4.2 基于时间序列的方法
环境数据有很强的时间相关性。比如PM2.5不会在1分钟内从50跳到500。我常用移动平均+残差检测。
# 移动平均残差法
window = 6 # 6小时窗口
merged_df['pm25_ma'] = merged_df['pm25'].rolling(window=window, center=True).mean()
merged_df['residual'] = np.abs(merged_df['pm25'] - merged_df['pm25_ma'])
# 残差超过3倍标准差视为异常
threshold = 3 * merged_df['residual'].std()
anomalies = merged_df[merged_df['residual'] > threshold]
3.4.3 基于空间一致性的方法
这是我最喜欢的方法。如果某个站点的读数与周围站点差异过大，那很可能是异常。我在做京津冀空气质量项目时，就用这个方法揪出了3个故障传感器。
# 空间一致性检测（假设有经纬度）
from sklearn.neighbors import NearestNeighbors
# 找到最近的5个站点
nbrs = NearestNeighbors(n_neighbors=5, algorithm='ball_tree').fit(coords)
distances, indices = nbrs.kneighbors(coords)
# 计算该站点与邻居的中位数差异
def spatial_anomaly(row_idx, data):
neighbors = indices[row_idx][1:] # 排除自身
neighbor_median = data.iloc[neighbors].median()
return abs(data.iloc[row_idx] - neighbor_median)
merged_df['spatial_diff'] = merged_df.index.map(
lambda i: spatial_anomaly(i, merged_df['pm25'])
)
3.5 本章知识体系
下面这张图是我自己画的，把数据采集与预处理的整个流程串起来了。你想想看，从原始数据到干净可用的训练集，中间要经过多少道关卡？
嗯，到这里数据采集与预处理的核心内容就讲完了。记住，数据预处理没有银弹，每个项目都要根据数据特点灵活调整。我刚开始做环境数据时也犯过不少错，但正是这些坑让我学会了如何「驯服」这些调皮的数据。