用Python构建城市AQI实时监控系统从数据爬取到可视化分析实战空气质量指数AQI已经成为现代城市居民日常关注的重要指标。作为数据分析师我们不仅需要理解AQI的计算原理更需要掌握如何从原始数据到最终可视化呈现的完整流程。本文将带你用Python构建一个完整的城市AQI监控系统涵盖数据采集、清洗、计算和可视化全流程。1. 环境准备与数据源分析在开始编码前我们需要做好充分的准备工作。空气质量数据通常可以从以下几个公开渠道获取中国环境监测总站官方网站地方环保部门开放数据平台第三方环境数据API如AQICN气象数据服务平台对于本实战项目我们将重点使用requests库进行网页数据抓取同时需要准备以下Python库# 必需库安装命令 pip install requests pandas numpy matplotlib beautifulsoup4注意在实际爬取数据时请务必遵守网站的robots.txt协议控制请求频率避免对目标服务器造成过大压力。数据源的选择至关重要。经过对比测试我们发现中国环境监测总站提供的城市点位数据具有以下特点数据特点优势局限性实时性每小时更新历史数据获取不便覆盖面全国主要城市部分小城市数据缺失指标完整包含6项污染物原始数据需要清洗2. 数据爬取与清洗实战2.1 网页数据抓取策略环境监测网站通常采用动态加载技术我们需要分析其数据接口。以下是一个典型的抓取函数实现import requests from bs4 import BeautifulSoup import json def fetch_aqi_data(city_code北京市): headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } base_url http://www.cnemc.cn/sssj/ try: response requests.get(base_url, headersheaders) soup BeautifulSoup(response.text, html.parser) # 解析数据脚本 script_data soup.find(script, {type: text/javascript}).string # 这里需要根据实际网页结构调整解析逻辑 return parse_raw_data(script_data) except Exception as e: print(f数据获取失败: {str(e)}) return None2.2 数据清洗与标准化原始数据往往存在以下问题需要处理字段缺失或异常值时间格式不统一污染物浓度单位不一致数据重复或错误我们使用pandas进行数据清洗import pandas as pd def clean_aqi_data(raw_df): # 处理缺失值 df raw_df.dropna(subset[PM2.5, PM10, SO2, NO2, CO, O3]) # 统一单位转换 df[CO] df[CO] * 1000 # 将CO从mg/m³转为μg/m³ # 时间标准化 df[timestamp] pd.to_datetime(df[time], format%Y-%m-%d %H:%M) df.set_index(timestamp, inplaceTrue) # 去除异常值 for col in df.columns: if col in [PM2.5, PM10]: df df[(df[col] 0) (df[col] 1000)] return df3. AQI计算核心算法实现3.1 理解HJ 633-2012标准根据《环境空气质量指数(AQI)技术规定》AQI计算需要以下步骤对每种污染物浓度进行单独评价计算单项空气质量指数(IAQI)从各项污染物的IAQI中选取最大值作为当前AQI确定首要污染物污染物浓度限值分为24小时平均和1小时平均两种标准污染物单位24小时平均限值(μg/m³)1小时平均限值(μg/m³)PM2.5μg/m³35/75/115/150/250/350/500同24小时PM10μg/m³50/150/250/350/420/500/600同24小时SO2μg/m³150/500/650/800150/500/650/800NO2μg/m³100/200/700/1200/2340/3090/3840同24小时COmg/m³5/10/35/60/90/120/150同24小时O3μg/m³160/200/300/400/800100/160/215/265/8003.2 Python实现AQI计算以下是完整的AQI计算函数实现import numpy as np def calculate_iaqi(cp, bp_lo, bp_hi, i_lo, i_hi): 计算单项污染物IAQI iaqi (i_hi - i_lo) * (cp - bp_lo) / (bp_hi - bp_lo) i_lo return round(iaqi) def calculate_aqi(row): 计算单条数据的AQI # AQI分级限值 aqi_breakpoints [0, 50, 100, 150, 200, 300, 400, 500] # 各污染物浓度限值(24小时平均) bp_pm25 [0, 35, 75, 115, 150, 250, 350, 500] bp_pm10 [0, 50, 150, 250, 350, 420, 500, 600] bp_so2 [0, 150, 500, 650, 800] bp_no2 [0, 100, 200, 700, 1200, 2340, 3090, 3840] bp_co [0, 5, 10, 35, 60, 90, 120, 150] bp_o3 [0, 160, 200, 300, 400, 800] # 计算各项IAQI iaqi_values [] # PM2.5 cp row[PM2.5] for i in range(1, len(bp_pm25)): if cp bp_pm25[i] or i len(bp_pm25)-1: iaqi calculate_iaqi(cp, bp_pm25[i-1], bp_pm25[i], aqi_breakpoints[i-1], aqi_breakpoints[i]) iaqi_values.append(iaqi) break # 其他污染物类似处理... # 返回AQI和首要污染物 aqi max(iaqi_values) primary_pollutant [PM2.5, PM10, SO2, NO2, CO, O3][iaqi_values.index(aqi)] return pd.Series([aqi, primary_pollutant], index[AQI, PrimaryPollutant])4. 数据可视化与分析4.1 时间序列分析使用matplotlib绘制AQI变化趋势import matplotlib.pyplot as plt import seaborn as sns def plot_aqi_trend(df, city_name): plt.figure(figsize(15, 6)) sns.lineplot(datadf, xdf.index, yAQI, huePrimaryPollutant, stylePrimaryPollutant, markersTrue, dashesFalse) plt.title(f{city_name} AQI变化趋势, fontsize15) plt.xlabel(时间) plt.ylabel(AQI指数) plt.xticks(rotation45) plt.grid(True, linestyle--, alpha0.6) plt.tight_layout() plt.show()4.2 污染物相关性分析通过热力图展示各污染物之间的相关性def plot_correlation(df): plt.figure(figsize(10, 8)) corr df[[PM2.5, PM10, SO2, NO2, CO, O3]].corr() sns.heatmap(corr, annotTrue, cmapcoolwarm, center0) plt.title(污染物浓度相关性分析, fontsize15) plt.tight_layout() plt.show()4.3 地理空间可视化进阶使用folium库创建交互式地图import folium def create_aqi_map(df, city_center): m folium.Map(locationcity_center, zoom_start12) for idx, row in df.iterrows(): # 根据AQI值设置颜色 color #00E400 if row[AQI] 50 else \ #FFFF00 if row[AQI] 100 else \ #FF7E00 if row[AQI] 150 else \ #FF0000 if row[AQI] 200 else \ #99004C if row[AQI] 300 else #7E0023 folium.CircleMarker( location[row[lat], row[lng]], radiusrow[AQI]/50, colorcolor, fillTrue, fill_colorcolor, popupfAQI: {row[AQI]}br首要污染物: {row[PrimaryPollutant]} ).add_to(m) return m5. 系统优化与生产部署5.1 性能优化技巧当处理大量城市或多点位数据时我们需要考虑性能优化使用多线程/异步IO提高爬取效率采用数据库存储替代CSV文件实现增量更新而非全量抓取使用缓存减少重复计算import concurrent.futures def fetch_multiple_cities(city_codes): 多线程抓取多个城市数据 with concurrent.futures.ThreadPoolExecutor(max_workers5) as executor: futures {executor.submit(fetch_aqi_data, code): code for code in city_codes} results [] for future in concurrent.futures.as_completed(futures): city_code futures[future] try: data future.result() if data: results.append(data) except Exception as e: print(f{city_code}数据获取失败: {str(e)}) return pd.concat(results)5.2 自动化监控系统设计将整个流程封装为自动化系统数据采集层定时爬取各城市数据数据处理层清洗、计算AQI、存储分析展示层生成可视化报告预警通知层当AQI超过阈值时发送警报import schedule import time def job(): print(开始执行AQI数据更新...) df fetch_aqi_data() clean_df clean_aqi_data(df) aqi_df clean_df.apply(calculate_aqi, axis1) plot_aqi_trend(aqi_df, 北京市) print(数据更新完成) # 每小时执行一次 schedule.every().hour.do(job) while True: schedule.run_pending() time.sleep(1)5.3 常见问题与解决方案在实际项目中我们可能会遇到以下典型问题数据缺失建立数据补全机制如使用前后时间点平均值异常值处理设置合理的上下限阈值反爬策略使用代理IP池、随机User-Agent计算误差严格对照HJ 633-2012标准验证算法对于大规模部署建议使用Airflow等工具构建完整的数据管道数据采集 → 数据清洗 → AQI计算 → 数据存储 → 可视化 → 预警通知通过这个实战项目我们不仅掌握了AQI的计算方法更重要的是构建了一个完整的空气质量监控系统原型。在实际应用中可以根据需求扩展更多功能如添加历史数据分析功能实现多城市对比开发移动端应用接入气象数据综合分析这个系统的核心价值在于将原始环境数据转化为直观、可操作的空气质量信息帮助决策者和公众更好地理解城市空气质量状况。
用Python爬取并计算城市AQI：一个数据分析师的空气质量监控实战（附完整代码）
发布时间：2026/8/1 1:16:53
用Python构建城市AQI实时监控系统从数据爬取到可视化分析实战空气质量指数AQI已经成为现代城市居民日常关注的重要指标。作为数据分析师我们不仅需要理解AQI的计算原理更需要掌握如何从原始数据到最终可视化呈现的完整流程。本文将带你用Python构建一个完整的城市AQI监控系统涵盖数据采集、清洗、计算和可视化全流程。1. 环境准备与数据源分析在开始编码前我们需要做好充分的准备工作。空气质量数据通常可以从以下几个公开渠道获取中国环境监测总站官方网站地方环保部门开放数据平台第三方环境数据API如AQICN气象数据服务平台对于本实战项目我们将重点使用requests库进行网页数据抓取同时需要准备以下Python库# 必需库安装命令 pip install requests pandas numpy matplotlib beautifulsoup4注意在实际爬取数据时请务必遵守网站的robots.txt协议控制请求频率避免对目标服务器造成过大压力。数据源的选择至关重要。经过对比测试我们发现中国环境监测总站提供的城市点位数据具有以下特点数据特点优势局限性实时性每小时更新历史数据获取不便覆盖面全国主要城市部分小城市数据缺失指标完整包含6项污染物原始数据需要清洗2. 数据爬取与清洗实战2.1 网页数据抓取策略环境监测网站通常采用动态加载技术我们需要分析其数据接口。以下是一个典型的抓取函数实现import requests from bs4 import BeautifulSoup import json def fetch_aqi_data(city_code北京市): headers { User-Agent: Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 } base_url http://www.cnemc.cn/sssj/ try: response requests.get(base_url, headersheaders) soup BeautifulSoup(response.text, html.parser) # 解析数据脚本 script_data soup.find(script, {type: text/javascript}).string # 这里需要根据实际网页结构调整解析逻辑 return parse_raw_data(script_data) except Exception as e: print(f数据获取失败: {str(e)}) return None2.2 数据清洗与标准化原始数据往往存在以下问题需要处理字段缺失或异常值时间格式不统一污染物浓度单位不一致数据重复或错误我们使用pandas进行数据清洗import pandas as pd def clean_aqi_data(raw_df): # 处理缺失值 df raw_df.dropna(subset[PM2.5, PM10, SO2, NO2, CO, O3]) # 统一单位转换 df[CO] df[CO] * 1000 # 将CO从mg/m³转为μg/m³ # 时间标准化 df[timestamp] pd.to_datetime(df[time], format%Y-%m-%d %H:%M) df.set_index(timestamp, inplaceTrue) # 去除异常值 for col in df.columns: if col in [PM2.5, PM10]: df df[(df[col] 0) (df[col] 1000)] return df3. AQI计算核心算法实现3.1 理解HJ 633-2012标准根据《环境空气质量指数(AQI)技术规定》AQI计算需要以下步骤对每种污染物浓度进行单独评价计算单项空气质量指数(IAQI)从各项污染物的IAQI中选取最大值作为当前AQI确定首要污染物污染物浓度限值分为24小时平均和1小时平均两种标准污染物单位24小时平均限值(μg/m³)1小时平均限值(μg/m³)PM2.5μg/m³35/75/115/150/250/350/500同24小时PM10μg/m³50/150/250/350/420/500/600同24小时SO2μg/m³150/500/650/800150/500/650/800NO2μg/m³100/200/700/1200/2340/3090/3840同24小时COmg/m³5/10/35/60/90/120/150同24小时O3μg/m³160/200/300/400/800100/160/215/265/8003.2 Python实现AQI计算以下是完整的AQI计算函数实现import numpy as np def calculate_iaqi(cp, bp_lo, bp_hi, i_lo, i_hi): 计算单项污染物IAQI iaqi (i_hi - i_lo) * (cp - bp_lo) / (bp_hi - bp_lo) i_lo return round(iaqi) def calculate_aqi(row): 计算单条数据的AQI # AQI分级限值 aqi_breakpoints [0, 50, 100, 150, 200, 300, 400, 500] # 各污染物浓度限值(24小时平均) bp_pm25 [0, 35, 75, 115, 150, 250, 350, 500] bp_pm10 [0, 50, 150, 250, 350, 420, 500, 600] bp_so2 [0, 150, 500, 650, 800] bp_no2 [0, 100, 200, 700, 1200, 2340, 3090, 3840] bp_co [0, 5, 10, 35, 60, 90, 120, 150] bp_o3 [0, 160, 200, 300, 400, 800] # 计算各项IAQI iaqi_values [] # PM2.5 cp row[PM2.5] for i in range(1, len(bp_pm25)): if cp bp_pm25[i] or i len(bp_pm25)-1: iaqi calculate_iaqi(cp, bp_pm25[i-1], bp_pm25[i], aqi_breakpoints[i-1], aqi_breakpoints[i]) iaqi_values.append(iaqi) break # 其他污染物类似处理... # 返回AQI和首要污染物 aqi max(iaqi_values) primary_pollutant [PM2.5, PM10, SO2, NO2, CO, O3][iaqi_values.index(aqi)] return pd.Series([aqi, primary_pollutant], index[AQI, PrimaryPollutant])4. 数据可视化与分析4.1 时间序列分析使用matplotlib绘制AQI变化趋势import matplotlib.pyplot as plt import seaborn as sns def plot_aqi_trend(df, city_name): plt.figure(figsize(15, 6)) sns.lineplot(datadf, xdf.index, yAQI, huePrimaryPollutant, stylePrimaryPollutant, markersTrue, dashesFalse) plt.title(f{city_name} AQI变化趋势, fontsize15) plt.xlabel(时间) plt.ylabel(AQI指数) plt.xticks(rotation45) plt.grid(True, linestyle--, alpha0.6) plt.tight_layout() plt.show()4.2 污染物相关性分析通过热力图展示各污染物之间的相关性def plot_correlation(df): plt.figure(figsize(10, 8)) corr df[[PM2.5, PM10, SO2, NO2, CO, O3]].corr() sns.heatmap(corr, annotTrue, cmapcoolwarm, center0) plt.title(污染物浓度相关性分析, fontsize15) plt.tight_layout() plt.show()4.3 地理空间可视化进阶使用folium库创建交互式地图import folium def create_aqi_map(df, city_center): m folium.Map(locationcity_center, zoom_start12) for idx, row in df.iterrows(): # 根据AQI值设置颜色 color #00E400 if row[AQI] 50 else \ #FFFF00 if row[AQI] 100 else \ #FF7E00 if row[AQI] 150 else \ #FF0000 if row[AQI] 200 else \ #99004C if row[AQI] 300 else #7E0023 folium.CircleMarker( location[row[lat], row[lng]], radiusrow[AQI]/50, colorcolor, fillTrue, fill_colorcolor, popupfAQI: {row[AQI]}br首要污染物: {row[PrimaryPollutant]} ).add_to(m) return m5. 系统优化与生产部署5.1 性能优化技巧当处理大量城市或多点位数据时我们需要考虑性能优化使用多线程/异步IO提高爬取效率采用数据库存储替代CSV文件实现增量更新而非全量抓取使用缓存减少重复计算import concurrent.futures def fetch_multiple_cities(city_codes): 多线程抓取多个城市数据 with concurrent.futures.ThreadPoolExecutor(max_workers5) as executor: futures {executor.submit(fetch_aqi_data, code): code for code in city_codes} results [] for future in concurrent.futures.as_completed(futures): city_code futures[future] try: data future.result() if data: results.append(data) except Exception as e: print(f{city_code}数据获取失败: {str(e)}) return pd.concat(results)5.2 自动化监控系统设计将整个流程封装为自动化系统数据采集层定时爬取各城市数据数据处理层清洗、计算AQI、存储分析展示层生成可视化报告预警通知层当AQI超过阈值时发送警报import schedule import time def job(): print(开始执行AQI数据更新...) df fetch_aqi_data() clean_df clean_aqi_data(df) aqi_df clean_df.apply(calculate_aqi, axis1) plot_aqi_trend(aqi_df, 北京市) print(数据更新完成) # 每小时执行一次 schedule.every().hour.do(job) while True: schedule.run_pending() time.sleep(1)5.3 常见问题与解决方案在实际项目中我们可能会遇到以下典型问题数据缺失建立数据补全机制如使用前后时间点平均值异常值处理设置合理的上下限阈值反爬策略使用代理IP池、随机User-Agent计算误差严格对照HJ 633-2012标准验证算法对于大规模部署建议使用Airflow等工具构建完整的数据管道数据采集 → 数据清洗 → AQI计算 → 数据存储 → 可视化 → 预警通知通过这个实战项目我们不仅掌握了AQI的计算方法更重要的是构建了一个完整的空气质量监控系统原型。在实际应用中可以根据需求扩展更多功能如添加历史数据分析功能实现多城市对比开发移动端应用接入气象数据综合分析这个系统的核心价值在于将原始环境数据转化为直观、可操作的空气质量信息帮助决策者和公众更好地理解城市空气质量状况。
