1. 从标准到代码为什么我们需要亲手算一遍AQI大家好我是老张在数据分析和环境监测这个行当里摸爬滚打了十来年。今天想和大家聊聊一个既熟悉又陌生的东西——空气质量指数也就是我们每天在天气App里看到的那个AQI。说它熟悉是因为我们天天看说它陌生是因为绝大多数人甚至很多数据分析新手都不知道这个数字背后到底是怎么算出来的。你可能觉得这不过是个简单的数字转换有现成的API接口调用一下不就完了我以前也这么想直到有一次我接手一个本地的空气质量分析项目需要处理一批历史传感器数据。当我尝试用某个公开服务计算AQI时发现结果和官方发布的数据对不上偏差还不小。那一刻我才意识到如果不搞清楚国家标准里白纸黑字写的计算逻辑仅仅当一个“调包侠”你的数据分析根基就是不稳的结果可能失之毫厘谬以千里。所以这篇文章就是带你一起亲手把国家标准《环境空气质量指数AQI技术规定试行》HJ 633-2012这本“武功秘籍”给拆解了并用Python代码把它实现出来。这个过程就像学做饭不能只看菜谱得亲自下锅炒一遍。目标读者很明确如果你是环境科学、大气科学专业的学生或者是一位需要对环境数据进行深度分析、甚至想自己开发一个小型空气质量监测应用的数据分析师或开发者那这篇文章就是为你准备的。我们会从最基础的概念讲起一直讲到可以运行的代码保证你读完就能自己动手算。2. 理解基石拆解AQI的国家标准GB3095-2012与HJ 633-2012要写代码先得读懂规则。中国的AQI计算主要依据两个核心文件《环境空气质量标准》GB3095-2012和《环境空气质量指数AQI技术规定试行》HJ 633-2012。前者规定了六种基本污染物的浓度限值后者则详细说明了如何将这些浓度值转化为我们看到的AQI指数。2.1 六位“主角”污染物与他们的浓度限值首先我们得认识一下这六位影响我们呼吸健康的“主角”。它们分别是细颗粒物PM2.5直径小于等于2.5微米的颗粒物能深入肺泡危害最大。可吸入颗粒物PM10直径小于等于10微米的颗粒物。二氧化硫SO2二氧化氮NO2一氧化碳CO臭氧O3这里有个非常关键的点也是新手最容易混淆的地方不是所有污染物都同时参与“实时AQI”和“24小时AQI”的计算。标准里明确区分了不同时间尺度的评价指标。对于1小时平均的“实时AQI”参与计算的污染物是PM2.5、PM10、SO2、NO2、CO和O3这六项。而对于24小时平均的“日均AQI”O3的评价标准使用的是“日最大8小时平均”浓度而不是24小时平均浓度。这一点在写代码时必须严格区分用错了标准结果肯定不对。2.2 核心算法IAQI与AQI的递进关系AQI的计算不是一个公式直接得出它分为两步走理解这个逻辑至关重要。第一步计算单项污染物的空气质量分指数IAQI这是最核心的一步。对于每一种污染物国家标准都给出了一个浓度限值表格这个表格将浓度划分成几个区间每个区间对应一个IAQI范围0-5051-100101-150151-200201-300301-400401-500。这就像一把有刻度的尺子我们把测得的污染物浓度值“放”到这把尺子上看看它落在哪个区间。计算IAQI的公式是一个简单的线性插值公式IAQI_p [(IAQI_Hi - IAQI_Lo) / (BP_Hi - BP_Lo)] * (C_p - BP_Lo) IAQI_Lo我来翻译一下这个“天书”IAQI_p你要计算的某种污染物比如PM2.5的分指数。C_p你实际测量到的这种污染物的浓度值。BP_Hi和BP_Lo在标准表格中C_p所落区间的浓度上限值和浓度下限值。IAQI_Hi和IAQI_Lo在标准表格中对应BP_Hi和BP_Lo的IAQI指数上限值和下限值。举个例子假设我们测得的PM2.5 1小时平均浓度是88 μg/m³。我们去查标准表这个表我们后面会具体给出发现88落在浓度区间[76115]里这个区间对应的IAQI范围是[151200]。那么BP_Lo 76 BP_Hi 115IAQI_Lo 151 IAQI_Hi 200C_p 88套入公式IAQI_pm2.5 [(200-151)/(115-76)] * (88-76) 151 ≈ 166.1四舍五入后得到166。第二步从所有IAQI中选出最大值即为AQIAQI max(IAQI_1, IAQI_2, ..., IAQI_6) 也就是说最终的AQI值等于所有参与计算的污染物分指数IAQI中最大的那个。同时这个最大值对应的污染物就被确定为当日的“首要污染物”。这个设计理念很好理解空气质量是由最差的那项指标决定的是“木桶效应”的典型体现。3. 把标准“翻译”成数据构建浓度-指数对照表理论懂了接下来就要为写代码做准备了。我们需要把国家标准文档里的那些表格“翻译”成程序能理解的数据结构。这是连接理论和实践的关键桥梁。3.1 读懂标准表格的结构我们以最常用的1小时平均浓度限值为例来看看标准表格长什么样。下面这个表是我根据HJ 633-2012整理的简化版核心数据污染物项目浓度限值 (μg/m³CO为mg/m³) 对应的IAQI (0, 50, 100, 150, 200, 300, 400, 500)PM2.5 (1小时)0, 35, 75, 115, 150, 250, 350, 500PM10 (1小时)0, 50, 150, 250, 350, 420, 500, 600SO2 (1小时)0, 150, 500, 650, 800, 1600, 2100, 2620NO2 (1小时)0, 100, 200, 700, 1200, 2340, 3090, 3840CO (1小时mg/m³)0, 5, 10, 35, 60, 90, 120, 150O3 (1小时)0, 160, 200, 300, 400, 800, 1000, 1200怎么读这个表以PM2.5第一行为例。它表示当PM2.5的1小时浓度小于等于35 μg/m³时其IAQI值在0-50之间线性变化对应“优”等级。当浓度大于35且小于等于75时IAQI在51-100之间变化对应“良”。当浓度大于75且小于等于115时IAQI在101-150之间变化对应“轻度污染”。以此类推直到浓度大于500时IAQI直接取500严重污染上限。特别注意O3和COO3的24小时评价用的是“日最大8小时平均”其浓度限值表与1小时表完全不同。而CO的浓度单位是mg/m³不是μg/m³在计算和比较时务必统一单位这是一个常见的坑。3.2 在代码中如何表示这张表在Python里我们可以用一个列表的列表二维列表或者字典来优雅地存储这个对照关系。我更喜欢用列表的列表因为它顺序固定索引方便和污染物数组能很好地对应起来。# 1小时平均AQI计算对应的浓度限值表 (Breakpoints) # 每个子列表代表一种污染物顺序为[PM2.5, PM10, SO2, NO2, CO, O3] # 列表内的数字是浓度值分别对应IAQI为0, 50, 100, 150, 200, 300, 400, 500时的浓度阈值。 # 注意CO的单位是mg/m³其他为μg/m³。 IAQI_BREAKPOINTS_1H [ [0, 35, 75, 115, 150, 250, 350, 500], # PM2.5 [0, 50, 150, 250, 350, 420, 500, 600], # PM10 [0, 150, 500, 650, 800, 1600, 2100, 2620], # SO2 [0, 100, 200, 700, 1200, 2340, 3090, 3840], # NO2 [0, 5, 10, 35, 60, 90, 120, 150], # CO (mg/m³) [0, 160, 200, 300, 400, 800, 1000, 1200] # O3 ] # 对应的IAQI指数值这是一个公共的对照列表 IAQI_INDEX [0, 50, 100, 150, 200, 300, 400, 500]这样当我们想查询PM2.5第一种污染物在IAQI为100对应索引2时的浓度限值只需要访问IAQI_BREAKPOINTS_1H[0][2]即可得到75。这个数据结构是我们后续所有计算的基础。4. 从零开始手把手实现AQI计算函数现在我们进入最激动人心的环节——写代码。我会带你一步步构建一个健壮、清晰、可复用的AQI计算模块而不仅仅是跑通一个脚本。4.1 核心函数计算单项污染物IAQI这是整个计算的原子操作。它的任务很明确给定一个污染物浓度值conc以及该污染物对应的浓度限值列表bp_list返回计算出的IAQI值。def calculate_iaqi(conc, bp_list): 计算单项污染物的空气质量分指数IAQI。 参数: conc (float): 污染物浓度值。 bp_list (list): 该污染物对应的浓度限值列表长度应为8。 返回: int: 计算得到的IAQI值范围0-500。 # 1. 边界检查如果浓度小于等于最低阈值IAQI按比例计算0-50区间 if conc bp_list[1]: # bp_list[1]对应IAQI 50的浓度 # 在0-50区间进行线性插值注意bp_list[0]通常是0 iaqi (conc / bp_list[1]) * 50 return max(0, int(round(iaqi))) # 确保非负并四舍五入取整 # 2. 边界检查如果浓度超过最高阈值直接返回500 if conc bp_list[-1]: # bp_list[-1]对应IAQI 500的浓度 return 500 # 3. 核心逻辑找到浓度值所在的区间 for i in range(1, len(bp_list)): if bp_list[i] conc: # 找到第一个大于等于conc的浓度阈值 bp_lo, bp_hi bp_list[i-1], bp_list[i] iaqi_lo, iaqi_hi IAQI_INDEX[i-1], IAQI_INDEX[i] # 应用线性插值公式 iaqi ((iaqi_hi - iaqi_lo) / (bp_hi - bp_lo)) * (conc - bp_lo) iaqi_lo return int(round(iaqi)) # 四舍五入取整 # 理论上不会走到这里但为了代码完整性 return 500这个函数我加了详细的注释并处理了两种边界情况浓度过低和过高这是很多简易实现会忽略的地方但在处理真实数据时非常重要。round()函数用于四舍五入int()确保返回整数这符合标准规定。4.2 整合函数计算整体AQI与首要污染物有了单兵作战的能力现在来组建军团。这个函数接收一组污染物浓度数据调用上面的calculate_iaqi函数并找出最大值确定AQI和首要污染物。def calculate_aqi_from_concentrations(concentrations, breakpoints_table): 根据一组污染物浓度计算AQI和首要污染物。 参数: concentrations (list/dict): 污染物浓度列表或字典。 列表顺序需与breakpoints_table一致[PM2.5, PM10, SO2, NO2, CO, O3]。 字典则需包含对应的键名。 breakpoints_table (list): 浓度限值二维列表。 返回: tuple: (AQI值, 首要污染物名称, 各污染物IAQI字典) # 定义污染物名称顺序用于输出 pollutant_names [PM2.5, PM10, SO2, NO2, CO, O3] # 处理输入如果是字典则转换为列表 if isinstance(concentrations, dict): conc_list [concentrations.get(name, 0) for name in pollutant_names] else: conc_list concentrations # 计算各项IAQI iaqi_values [] for i, conc in enumerate(conc_list): if conc is None or conc 0: # 处理无效数据 iaqi 0 else: iaqi calculate_iaqi(conc, breakpoints_table[i]) iaqi_values.append(iaqi) # 找出最大的IAQI作为AQI aqi max(iaqi_values) # 找出首要污染物可能有多个并列 primary_pollutants [] for i, iaqi in enumerate(iaqi_values): if iaqi aqi: primary_pollutants.append(pollutant_names[i]) # 组装各污染物IAQI结果字典 iaqi_dict {name: value for name, value in zip(pollutant_names, iaqi_values)} return aqi, primary_pollutants, iaqi_dict这个函数的设计考虑了实用性。它既能接受列表输入也能接受字典输入更符合从JSON或数据库读取数据的场景。返回结果不仅包含AQI还包含了首要污染物列表和各污染物的详细IAQI值信息非常完整。5. 实战演练处理真实数据并可视化理论代码都有了不跑起来看看怎么行我们来模拟一个真实的数据处理场景。假设我们从某个空气质量监测站拿到了一小时的数据格式是JSON。5.1 读取与清洗数据首先我们准备一份模拟的JSON数据文件air_quality_data.json{ station_id: station_001, timestamp: 2023-10-27T14:00:00, readings: { PM2.5: 88.5, PM10: 120.3, SO2: 45.2, NO2: 78.9, CO: 1.2, O3: 110.5 } }然后写一个完整的脚本从数据读取、计算到结果展示import json import pandas as pd import matplotlib.pyplot as plt # 加载之前定义的核心函数和标准表假设它们在一个叫aqi_calculator的模块里 # 这里为了演示我们直接写在一个文件里 # ... (将前面第4节的calculate_iaqi和calculate_aqi_from_concentrations函数定义放在这里) # ... (将IAQI_BREAKPOINTS_1H和IAQI_INDEX的定义放在这里) def process_real_data(file_path): 处理单个时间点的监测数据 with open(file_path, r, encodingutf-8) as f: data json.load(f) timestamp data[timestamp] readings data[readings] # 计算AQI aqi, primary_pollutants, iaqi_dict calculate_aqi_from_concentrations( readings, IAQI_BREAKPOINTS_1H ) print(f 监测站 {data[station_id]} - {timestamp} ) print(f污染物浓度 (原始数据):) for pol, conc in readings.items(): unit μg/m³ if pol ! CO else mg/m³ print(f {pol}: {conc} {unit}) print(f\n各污染物IAQI分指数:) for pol, iaqi in iaqi_dict.items(): print(f {pol}: {iaqi}) print(f\n**最终AQI: {aqi}**) print(f首要污染物: {, .join(primary_pollutants)}) return aqi, primary_pollutants, iaqi_dict, readings # 执行计算 if __name__ __main__: result process_real_data(air_quality_data.json)运行这段代码你会得到类似这样的输出 监测站 station_001 - 2023-10-27T14:00:00 污染物浓度 (原始数据): PM2.5: 88.5 μg/m³ PM10: 120.3 μg/m³ SO2: 45.2 μg/m³ NO2: 78.9 μg/m³ CO: 1.2 mg/m³ O3: 110.5 μg/m³ 各污染物IAQI分指数: PM2.5: 166 PM10: 100 SO2: 15 NO2: 39 CO: 12 O3: 34 **最终AQI: 166** 首要污染物: PM2.5看计算出来了AQI是166首要污染物是PM2.5属于“轻度污染”。这个结果和我们之前手动估算的PM2.5 IAQI值是一致的。5.2 进阶处理时间序列数据并绘图实际项目中我们更常处理的是时间序列数据。比如一个监测站一天24小时的数据。我们可以用Pandas来高效处理并绘制趋势图。def process_time_series_data(csv_file_path): 处理CSV格式的时间序列数据 # 假设CSV文件有列timestamp, PM2.5, PM10, SO2, NO2, CO, O3 df pd.read_csv(csv_file_path, parse_dates[timestamp]) aqi_list [] primary_list [] print(开始批量计算时间序列AQI...) for index, row in df.iterrows(): # 提取浓度数据 conc_dict { PM2.5: row[PM2.5], PM10: row[PM10], SO2: row[SO2], NO2: row[NO2], CO: row[CO], O3: row[O3] } # 计算AQI aqi, primary, _ calculate_aqi_from_concentrations(conc_dict, IAQI_BREAKPOINTS_1H) aqi_list.append(aqi) # 取第一个首要污染物通常只有一个 primary_list.append(primary[0] if primary else N/A) # 将结果添加回DataFrame df[AQI] aqi_list df[Primary_Pollutant] primary_list # 简单可视化 plt.figure(figsize(12, 6)) plt.plot(df[timestamp], df[AQI], markero, linestyle-, linewidth2, markersize4) plt.title(24小时AQI变化趋势) plt.xlabel(时间) plt.ylabel(AQI) plt.grid(True, alpha0.3) plt.axhline(y100, colorr, linestyle--, alpha0.5, label良/轻度污染界限) plt.legend() plt.tight_layout() plt.savefig(aqi_trend_24h.png, dpi150) plt.show() # 输出统计信息 print(f\n 24小时数据统计 ) print(fAQI平均值: {df[AQI].mean():.1f}) print(fAQI最大值: {df[AQI].max()} (出现在 {df.loc[df[AQI].idxmax(), timestamp]})) print(f首要污染物出现频率:) print(df[Primary_Pollutant].value_counts()) return df # 假设你有一个名为‘hourly_data.csv’的文件 # df_result process_time_series_data(hourly_data.csv)通过这个时间序列处理你不仅能得到每个时间点的AQI还能分析全天的变化趋势、污染峰值时段、以及首要污染物的演变规律。这些洞察对于环境分析和报告至关重要。6. 避坑指南与高级话题在真实项目中摸爬滚打几年我踩过不少坑。这里分享几个关键点希望能帮你省点时间。坑1单位单位单位这是最最常见的错误。国家标准中CO的浓度单位是毫克/立方米mg/m³而其他污染物是微克/立方米μg/m³。很多数据源提供的CO单位可能是ppm或ppb需要进行换算。如果你的CO浓度值看起来特别小比如0.001那很可能单位是mg/m³如果看起来是几或者几十那可能就是ppm需要转换。在计算前务必确认并统一所有数据的单位。坑21小时平均 vs 24小时平均 vs O3-8h这是算法选择的核心。务必根据你的数据时间分辨率选择正确的浓度限值表。实时数据1小时平均使用我们上面定义的IAQI_BREAKPOINTS_1H。日均数据24小时平均需要使用另一套不同的浓度限值表特别是O3其24小时评价标准是“日最大8小时滑动平均”算法更复杂需要先计算8小时滑动平均值再查对应的限值表。千万不要混用。坑3数据有效性规定不是所有数据都能拿来算。标准规定计算1小时AQI时至少需要45分钟的采样数据计算24小时平均需要至少20个小时的有效数据。对于缺失或无效的数据在计算时需要谨慎处理比如置为0或标记为无效避免影响整体结果。坑4四舍五入与取整标准规定IAQI和AQI计算结果只保留整数并且是四舍五入。在Python中round()函数默认就是四舍五入注意Python 3的round规则是“银行家舍入法”但对于整数转换影响不大。确保你的最终输出是整数。关于扩展空气质量等级与健康提示计算出AQI后通常还需要将其转换为空气质量等级和健康提示。这个对照关系是固定的AQI数值范围空气质量等级表示颜色对健康影响0 - 50优绿色空气质量令人满意基本无空气污染51 - 100良黄色空气质量可接受但某些污染物可能对极少数异常敏感人群健康有较弱影响101 - 150轻度污染橙色易感人群症状有轻度加剧健康人群出现刺激症状151 - 200中度污染红色进一步加剧易感人群症状可能对健康人群心脏、呼吸系统有影响201 - 300重度污染紫色心脏病和肺病患者症状显著加剧运动耐受力降低健康人群普遍出现症状301 - 500严重污染褐红色健康人群运动耐受力降低有明显强烈症状提前出现某些疾病你可以写一个简单的函数来完成这个映射def aqi_to_category(aqi): if aqi 50: return 优, 绿色 elif aqi 100: return 良, 黄色 elif aqi 150: return 轻度污染, 橙色 elif aqi 200: return 中度污染, 红色 elif aqi 300: return 重度污染, 紫色 else: return 严重污染, 褐红色最后我想说理解并实现AQI的计算远不止是为了得到一个数字。它让你掌握了环境数据“翻译”成公共信息的钥匙。当你自己动手写完这些代码再去看那些空气质量App或平台的数据时感觉会完全不一样。你会知道166这个数字背后是PM2.5浓度在88.5微克每立方米时根据国家标准线性插值得出的结果。这种从黑盒到白盒的转变是技术人最大的乐趣和底气。下次当你需要做相关的数据分析、开发应用或者写报告时这套代码和思路应该能给你提供一个扎实的起点。如果遇到问题不妨回头再看看国家标准原文那才是最权威的答案。