空气质量数据背后的秘密教你用Python爬取PM2.5历史数据并可视化分析最近几年每当秋冬季节来临朋友圈总会被各种晒蓝天和吐槽雾霾的内容刷屏。作为一名数据爱好者我常常好奇我们城市的空气质量到底是在变好还是变坏每天的PM2.5波动有什么规律不同季节的污染特征有何不同为了回答这些问题我决定自己动手用Python构建一个空气质量数据分析工具。本文将带你从零开始通过爬取权威空气质量数据建立本地数据库并进行多维度的可视化分析。不同于简单的数据展示我们会重点关注如何从原始数据中挖掘有价值的洞见。整个过程只需要基础的Python知识但最终你将获得一个可以持续监测和分析空气质量变化的实用工具。1. 数据获取构建稳定的爬虫系统1.1 选择可靠的数据源在开始编码之前首先要确定数据来源。经过对比测试我发现中国环境监测总站提供的实时发布系统数据最为权威和稳定。这个平台不仅提供实时数据还包含历史数据查询功能非常适合我们的需求。import requests from bs4 import BeautifulSoup def get_city_list(): url http://www.cnemc.cn/sssj/ response requests.get(url) soup BeautifulSoup(response.text, html.parser) city_options soup.select(select[namecity] option) return {opt.text: opt[value] for opt in city_options if opt[value]}提示在实际项目中建议设置合理的请求间隔如5-10秒一次避免给服务器造成过大压力。1.2 解析网页结构获取关键数据空气质量数据通常以表格形式呈现我们需要准确识别其中的关键字段字段名称说明数据类型AQI空气质量指数整数PM2.5细颗粒物浓度整数(μg/m³)PM10可吸入颗粒物浓度整数(μg/m³)SO2二氧化硫浓度整数(μg/m³)NO2二氧化氮浓度整数(μg/m³)CO一氧化碳浓度浮点(mg/m³)O3臭氧浓度整数(μg/m³)def parse_air_data(html): soup BeautifulSoup(html, html.parser) table soup.find(table, {class: report-table}) data {} for row in table.find_all(tr): cells row.find_all(td) if len(cells) 2: key cells[0].text.strip() value cells[1].text.strip() data[key] float(value) if . in value else int(value) return data1.3 构建完整的历史数据爬取流程为了获取长期数据我们需要设计一个可以自动翻页、处理异常的重试机制import time from datetime import datetime, timedelta def fetch_history_data(city_code, start_date, end_date): base_url http://www.cnemc.cn/sssj/history.aspx session requests.Session() all_data [] current_date start_date while current_date end_date: try: params { city: city_code, date: current_date.strftime(%Y-%m-%d) } response session.get(base_url, paramsparams) day_data parse_air_data(response.text) day_data[date] current_date.strftime(%Y-%m-%d) all_data.append(day_data) print(f成功获取 {current_date} 数据) time.sleep(8) # 礼貌性延迟 current_date timedelta(days1) except Exception as e: print(f获取 {current_date} 数据失败: {str(e)}) time.sleep(30) # 失败后延长等待时间 return all_data2. 数据存储与管理构建本地空气质量数据库2.1 设计合理的数据存储结构获取到的原始数据需要经过清洗和结构化处理才能用于分析。我推荐使用SQLite作为本地存储方案它轻量且无需额外配置import sqlite3 from contextlib import closing def init_database(db_pathair_quality.db): with closing(sqlite3.connect(db_path)) as conn: cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS air_quality ( date TEXT PRIMARY KEY, aqi INTEGER, pm25 INTEGER, pm10 INTEGER, so2 INTEGER, no2 INTEGER, co REAL, o3 INTEGER, primary_pollutant TEXT, quality_level TEXT ) ) conn.commit()2.2 实现高效的数据更新机制为了避免重复爬取已经存在的数据我们需要实现增量更新功能def update_database(new_data, db_pathair_quality.db): with closing(sqlite3.connect(db_path)) as conn: cursor conn.cursor() # 获取已有日期列表 cursor.execute(SELECT date FROM air_quality) existing_dates {row[0] for row in cursor.fetchall()} # 只插入新数据 inserted 0 for record in new_data: if record[date] not in existing_dates: cursor.execute( INSERT INTO air_quality VALUES ( :date, :aqi, :pm25, :pm10, :so2, :no2, :co, :o3, :primary_pollutant, :quality_level ) , record) inserted 1 conn.commit() print(f成功插入 {inserted} 条新记录)2.3 数据质量控制与异常处理原始数据中可能存在缺失值或异常值我们需要建立自动化的数据清洗流程def clean_air_data(raw_data): cleaned [] for record in raw_data: # 处理缺失值 clean_record { date: record.get(date, ), aqi: record.get(AQI, -1), pm25: record.get(PM2.5, -1), pm10: record.get(PM10, -1), so2: record.get(SO2, -1), no2: record.get(NO2, -1), co: record.get(CO, -1.0), o3: record.get(O3, -1), primary_pollutant: record.get(首要污染物, ), quality_level: record.get(空气质量级别, ) } # 验证数据有效性 if clean_record[date] and clean_record[aqi] 0: cleaned.append(clean_record) return cleaned3. 数据分析挖掘空气质量变化规律3.1 基础统计分析了解数据全貌使用pandas可以快速计算各种统计指标import pandas as pd def basic_analysis(db_pathair_quality.db): with closing(sqlite3.connect(db_path)) as conn: df pd.read_sql(SELECT * FROM air_quality, conn) # 计算各污染物年度平均值 annual_avg df.groupby(pd.to_datetime(df[date]).dt.year).mean() # 计算月度变化 df[month] pd.to_datetime(df[date]).dt.month monthly_avg df.groupby(month).mean() return { annual_trend: annual_avg, monthly_pattern: monthly_avg }3.2 可视化分析直观展示数据特征matplotlib和seaborn组合可以创建丰富的可视化图表import matplotlib.pyplot as plt import seaborn as sns def plot_pm25_trend(df): plt.figure(figsize(12, 6)) df[date] pd.to_datetime(df[date]) df df.set_index(date).sort_index() # 使用移动平均平滑数据 df[pm25_30d] df[pm25].rolling(30).mean() ax sns.lineplot(datadf, xdf.index, ypm25_30d, colorred, linewidth2) ax.set(titlePM2.5浓度30日移动平均趋势, ylabelPM2.5浓度 (μg/m³), xlabel日期) plt.xticks(rotation45) plt.tight_layout() plt.show()3.3 高级分析污染物相关性研究不同污染物之间往往存在关联我们可以通过热力图展示它们的关系def plot_pollutant_correlation(df): pollutants [pm25, pm10, so2, no2, co, o3] corr_matrix df[pollutants].corr() plt.figure(figsize(10, 8)) sns.heatmap(corr_matrix, annotTrue, cmapcoolwarm, vmin-1, vmax1, center0) plt.title(主要空气污染物相关性分析) plt.tight_layout() plt.show()4. 应用扩展构建实用的空气质量监测系统4.1 自动化数据更新与预警我们可以将爬虫部署到服务器实现每日自动更新import schedule import time def daily_update_job(): today datetime.now().strftime(%Y-%m-%d) data fetch_history_data(101010100, today, today) # 北京城市代码 cleaned clean_air_data(data) update_database(cleaned) # 每天上午10点执行 schedule.every().day.at(10:00).do(daily_update_job) while True: schedule.run_pending() time.sleep(60)4.2 开发简单的Web展示界面使用Flask可以快速构建一个数据展示页面from flask import Flask, render_template import pandas as pd app Flask(__name__) app.route(/) def dashboard(): with closing(sqlite3.connect(air_quality.db)) as conn: df pd.read_sql( SELECT date, aqi, pm25, quality_level FROM air_quality ORDER BY date DESC LIMIT 30 , conn) return render_template(dashboard.html, datadf.to_dict(records)) if __name__ __main__: app.run(debugTrue)4.3 空气质量预测模型初探基于历史数据我们可以尝试构建简单的预测模型from sklearn.ensemble import RandomForestRegressor from sklearn.model_selection import train_test_split def build_prediction_model(df): # 特征工程 df[date] pd.to_datetime(df[date]) df[day_of_year] df[date].dt.dayofyear df[month] df[date].dt.month df[year] df[date].dt.year # 准备特征和目标变量 features [day_of_year, month, year, pm10, so2, no2, co, o3] X df[features] y df[pm25] # 划分训练测试集 X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.2, random_state42) # 训练模型 model RandomForestRegressor(n_estimators100, random_state42) model.fit(X_train, y_train) return model, X_test, y_test在实际项目中我发现PM2.5数据有明显的季节性特征冬季浓度通常比夏季高2-3倍。通过持续监测可以明显看到近几年空气质量改善的趋势特别是在实施严格环保政策的时期数据变化尤为显著。这个系统不仅帮助我了解了空气质量的宏观变化还让我养成了关注环境数据的习惯。
空气质量数据背后的秘密：教你用Python爬取PM2.5历史数据并可视化分析
空气质量数据背后的秘密教你用Python爬取PM2.5历史数据并可视化分析最近几年每当秋冬季节来临朋友圈总会被各种晒蓝天和吐槽雾霾的内容刷屏。作为一名数据爱好者我常常好奇我们城市的空气质量到底是在变好还是变坏每天的PM2.5波动有什么规律不同季节的污染特征有何不同为了回答这些问题我决定自己动手用Python构建一个空气质量数据分析工具。本文将带你从零开始通过爬取权威空气质量数据建立本地数据库并进行多维度的可视化分析。不同于简单的数据展示我们会重点关注如何从原始数据中挖掘有价值的洞见。整个过程只需要基础的Python知识但最终你将获得一个可以持续监测和分析空气质量变化的实用工具。1. 数据获取构建稳定的爬虫系统1.1 选择可靠的数据源在开始编码之前首先要确定数据来源。经过对比测试我发现中国环境监测总站提供的实时发布系统数据最为权威和稳定。这个平台不仅提供实时数据还包含历史数据查询功能非常适合我们的需求。import requests from bs4 import BeautifulSoup def get_city_list(): url http://www.cnemc.cn/sssj/ response requests.get(url) soup BeautifulSoup(response.text, html.parser) city_options soup.select(select[namecity] option) return {opt.text: opt[value] for opt in city_options if opt[value]}提示在实际项目中建议设置合理的请求间隔如5-10秒一次避免给服务器造成过大压力。1.2 解析网页结构获取关键数据空气质量数据通常以表格形式呈现我们需要准确识别其中的关键字段字段名称说明数据类型AQI空气质量指数整数PM2.5细颗粒物浓度整数(μg/m³)PM10可吸入颗粒物浓度整数(μg/m³)SO2二氧化硫浓度整数(μg/m³)NO2二氧化氮浓度整数(μg/m³)CO一氧化碳浓度浮点(mg/m³)O3臭氧浓度整数(μg/m³)def parse_air_data(html): soup BeautifulSoup(html, html.parser) table soup.find(table, {class: report-table}) data {} for row in table.find_all(tr): cells row.find_all(td) if len(cells) 2: key cells[0].text.strip() value cells[1].text.strip() data[key] float(value) if . in value else int(value) return data1.3 构建完整的历史数据爬取流程为了获取长期数据我们需要设计一个可以自动翻页、处理异常的重试机制import time from datetime import datetime, timedelta def fetch_history_data(city_code, start_date, end_date): base_url http://www.cnemc.cn/sssj/history.aspx session requests.Session() all_data [] current_date start_date while current_date end_date: try: params { city: city_code, date: current_date.strftime(%Y-%m-%d) } response session.get(base_url, paramsparams) day_data parse_air_data(response.text) day_data[date] current_date.strftime(%Y-%m-%d) all_data.append(day_data) print(f成功获取 {current_date} 数据) time.sleep(8) # 礼貌性延迟 current_date timedelta(days1) except Exception as e: print(f获取 {current_date} 数据失败: {str(e)}) time.sleep(30) # 失败后延长等待时间 return all_data2. 数据存储与管理构建本地空气质量数据库2.1 设计合理的数据存储结构获取到的原始数据需要经过清洗和结构化处理才能用于分析。我推荐使用SQLite作为本地存储方案它轻量且无需额外配置import sqlite3 from contextlib import closing def init_database(db_pathair_quality.db): with closing(sqlite3.connect(db_path)) as conn: cursor conn.cursor() cursor.execute( CREATE TABLE IF NOT EXISTS air_quality ( date TEXT PRIMARY KEY, aqi INTEGER, pm25 INTEGER, pm10 INTEGER, so2 INTEGER, no2 INTEGER, co REAL, o3 INTEGER, primary_pollutant TEXT, quality_level TEXT ) ) conn.commit()2.2 实现高效的数据更新机制为了避免重复爬取已经存在的数据我们需要实现增量更新功能def update_database(new_data, db_pathair_quality.db): with closing(sqlite3.connect(db_path)) as conn: cursor conn.cursor() # 获取已有日期列表 cursor.execute(SELECT date FROM air_quality) existing_dates {row[0] for row in cursor.fetchall()} # 只插入新数据 inserted 0 for record in new_data: if record[date] not in existing_dates: cursor.execute( INSERT INTO air_quality VALUES ( :date, :aqi, :pm25, :pm10, :so2, :no2, :co, :o3, :primary_pollutant, :quality_level ) , record) inserted 1 conn.commit() print(f成功插入 {inserted} 条新记录)2.3 数据质量控制与异常处理原始数据中可能存在缺失值或异常值我们需要建立自动化的数据清洗流程def clean_air_data(raw_data): cleaned [] for record in raw_data: # 处理缺失值 clean_record { date: record.get(date, ), aqi: record.get(AQI, -1), pm25: record.get(PM2.5, -1), pm10: record.get(PM10, -1), so2: record.get(SO2, -1), no2: record.get(NO2, -1), co: record.get(CO, -1.0), o3: record.get(O3, -1), primary_pollutant: record.get(首要污染物, ), quality_level: record.get(空气质量级别, ) } # 验证数据有效性 if clean_record[date] and clean_record[aqi] 0: cleaned.append(clean_record) return cleaned3. 数据分析挖掘空气质量变化规律3.1 基础统计分析了解数据全貌使用pandas可以快速计算各种统计指标import pandas as pd def basic_analysis(db_pathair_quality.db): with closing(sqlite3.connect(db_path)) as conn: df pd.read_sql(SELECT * FROM air_quality, conn) # 计算各污染物年度平均值 annual_avg df.groupby(pd.to_datetime(df[date]).dt.year).mean() # 计算月度变化 df[month] pd.to_datetime(df[date]).dt.month monthly_avg df.groupby(month).mean() return { annual_trend: annual_avg, monthly_pattern: monthly_avg }3.2 可视化分析直观展示数据特征matplotlib和seaborn组合可以创建丰富的可视化图表import matplotlib.pyplot as plt import seaborn as sns def plot_pm25_trend(df): plt.figure(figsize(12, 6)) df[date] pd.to_datetime(df[date]) df df.set_index(date).sort_index() # 使用移动平均平滑数据 df[pm25_30d] df[pm25].rolling(30).mean() ax sns.lineplot(datadf, xdf.index, ypm25_30d, colorred, linewidth2) ax.set(titlePM2.5浓度30日移动平均趋势, ylabelPM2.5浓度 (μg/m³), xlabel日期) plt.xticks(rotation45) plt.tight_layout() plt.show()3.3 高级分析污染物相关性研究不同污染物之间往往存在关联我们可以通过热力图展示它们的关系def plot_pollutant_correlation(df): pollutants [pm25, pm10, so2, no2, co, o3] corr_matrix df[pollutants].corr() plt.figure(figsize(10, 8)) sns.heatmap(corr_matrix, annotTrue, cmapcoolwarm, vmin-1, vmax1, center0) plt.title(主要空气污染物相关性分析) plt.tight_layout() plt.show()4. 应用扩展构建实用的空气质量监测系统4.1 自动化数据更新与预警我们可以将爬虫部署到服务器实现每日自动更新import schedule import time def daily_update_job(): today datetime.now().strftime(%Y-%m-%d) data fetch_history_data(101010100, today, today) # 北京城市代码 cleaned clean_air_data(data) update_database(cleaned) # 每天上午10点执行 schedule.every().day.at(10:00).do(daily_update_job) while True: schedule.run_pending() time.sleep(60)4.2 开发简单的Web展示界面使用Flask可以快速构建一个数据展示页面from flask import Flask, render_template import pandas as pd app Flask(__name__) app.route(/) def dashboard(): with closing(sqlite3.connect(air_quality.db)) as conn: df pd.read_sql( SELECT date, aqi, pm25, quality_level FROM air_quality ORDER BY date DESC LIMIT 30 , conn) return render_template(dashboard.html, datadf.to_dict(records)) if __name__ __main__: app.run(debugTrue)4.3 空气质量预测模型初探基于历史数据我们可以尝试构建简单的预测模型from sklearn.ensemble import RandomForestRegressor from sklearn.model_selection import train_test_split def build_prediction_model(df): # 特征工程 df[date] pd.to_datetime(df[date]) df[day_of_year] df[date].dt.dayofyear df[month] df[date].dt.month df[year] df[date].dt.year # 准备特征和目标变量 features [day_of_year, month, year, pm10, so2, no2, co, o3] X df[features] y df[pm25] # 划分训练测试集 X_train, X_test, y_train, y_test train_test_split( X, y, test_size0.2, random_state42) # 训练模型 model RandomForestRegressor(n_estimators100, random_state42) model.fit(X_train, y_train) return model, X_test, y_test在实际项目中我发现PM2.5数据有明显的季节性特征冬季浓度通常比夏季高2-3倍。通过持续监测可以明显看到近几年空气质量改善的趋势特别是在实施严格环保政策的时期数据变化尤为显著。这个系统不仅帮助我了解了空气质量的宏观变化还让我养成了关注环境数据的习惯。
