空气质量数据背后的秘密：教你用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[name="city"] 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 data

1.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, params=params)
            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(days=1)
            
        except Exception as e:
            print(f"获取 {current_date} 数据失败: {str(e)}")
            time.sleep(30)  # 失败后延长等待时间
    
    return all_data

2. 数据存储与管理：构建本地空气质量数据库

2.1 设计合理的数据存储结构

获取到的原始数据需要经过清洗和结构化处理才能用于分析。我推荐使用SQLite作为本地存储方案，它轻量且无需额外配置：

import sqlite3
from contextlib import closing

def init_database(db_path='air_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_path='air_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 cleaned

3. 数据分析：挖掘空气质量变化规律

3.1 基础统计分析：了解数据全貌

使用pandas可以快速计算各种统计指标：

import pandas as pd

def basic_analysis(db_path='air_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(data=df, x=df.index, y='pm25_30d', 
                     color='red', linewidth=2)
    ax.set(title='PM2.5浓度30日移动平均趋势', 
          ylabel='PM2.5浓度 (μg/m³)',
          xlabel='日期')
    plt.xticks(rotation=45)
    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, annot=True, cmap='coolwarm', 
               vmin=-1, vmax=1, center=0)
    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', 
                         data=df.to_dict('records'))

if __name__ == '__main__':
    app.run(debug=True)

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_size=0.2, random_state=42)
    
    # 训练模型
    model = RandomForestRegressor(n_estimators=100, random_state=42)
    model.fit(X_train, y_train)
    
    return model, X_test, y_test

在实际项目中，我发现PM2.5数据有明显的季节性特征，冬季浓度通常比夏季高2-3倍。通过持续监测，可以明显看到近几年空气质量改善的趋势，特别是在实施严格环保政策的时期，数据变化尤为显著。这个系统不仅帮助我了解了空气质量的宏观变化，还让我养成了关注环境数据的习惯。