簡(jiǎn)介城市空氣質(zhì)量數(shù)據(jù)分析平臺(tái)是一套面向Python初學(xué)者與數(shù)據(jù)實(shí)踐者的輕量級(jí)空氣質(zhì)量監(jiān)控與分析系統(tǒng)適用于環(huán)境數(shù)據(jù)分析、課程設(shè)計(jì)或小型科研項(xiàng)目場(chǎng)景。資源聚焦數(shù)據(jù)采集、清洗、統(tǒng)計(jì)與可視化全流程幫助用戶掌握requests網(wǎng)絡(luò)請(qǐng)求、SQLAlchemy數(shù)據(jù)庫(kù)操作、matplotlib圖表繪制等核心技能。壓縮包共9個(gè)文件6個(gè)Python主程序模塊、2個(gè)配置/說明文本、1個(gè)Markdown文檔總大小僅18KB結(jié)構(gòu)清晰main.py為入口data_collector/collector.py負(fù)責(zé)AQI數(shù)據(jù)抓取db.py實(shí)現(xiàn)數(shù)據(jù)庫(kù)建表與存儲(chǔ)processor.py完成缺失值處理與異常值清洗visualizer.py生成趨勢(shì)圖與熱力圖配合requirements.txt與README.md便于快速部署。目前已有125人學(xué)習(xí)下載讀者可直接運(yùn)行獲得全國(guó)主要城市空氣質(zhì)量變化分析結(jié)果獲取完整可執(zhí)行代碼、模塊化設(shè)計(jì)思路及典型環(huán)境數(shù)據(jù)處理排錯(cuò)要點(diǎn)。1. 城市空氣質(zhì)量數(shù)據(jù)分析平臺(tái)不是又一個(gè)Dashboard而是從原始監(jiān)測(cè)數(shù)據(jù)到可信結(jié)論的閉環(huán)流水線你手頭有一堆來自國(guó)控站點(diǎn)的 hourly.csv 文件字段名五花八門——有的叫PM25有的叫PM2.5(μg/m3)還有的帶空格、中文括號(hào)、單位后綴時(shí)間戳格式不統(tǒng)一有的是2023-01-01 00:00:00有的是2023/01/01 00:00甚至混著2023010100這種無分隔純數(shù)字更頭疼的是某站點(diǎn)連續(xù)72小時(shí)SO2值恒為0.0另一站點(diǎn)在暴雨天PM10反而飆升到 800——這些不是噪聲是真實(shí)業(yè)務(wù)中每天要啃的硬骨頭。這個(gè)「城市空氣質(zhì)量數(shù)據(jù)分析平臺(tái)」不是封裝好的 Web 應(yīng)用而是一套可復(fù)現(xiàn)、可審計(jì)、可嵌入 CI/CD 的 Python 工程化腳本集它把數(shù)據(jù)庫(kù)建模、臟數(shù)據(jù)定位、多源時(shí)間對(duì)齊、濃度統(tǒng)計(jì)歸因、空間熱力映射全鏈路串起來專治“數(shù)據(jù)已下載但報(bào)告寫不出”的典型翻車現(xiàn)場(chǎng)。適合剛接手環(huán)保類數(shù)據(jù)項(xiàng)目的工程師、需要交付可驗(yàn)證分析結(jié)果的課題組成員以及想把 Excel 手工清洗升級(jí)為自動(dòng)化 pipeline 的環(huán)境監(jiān)測(cè)員。2. 數(shù)據(jù)庫(kù)設(shè)計(jì)與初始化為什么用 SQLite 而非 MySQL三個(gè)現(xiàn)實(shí)約束下的選型邏輯2.1 為什么 SQLite 是中小型空氣質(zhì)量分析的默認(rèn)起點(diǎn)這不是技術(shù)情懷而是三重硬約束倒逼的選擇第一數(shù)據(jù)源常以離線 CSV/Excel 形式下發(fā)無實(shí)時(shí) API部署 PostgreSQL 意味著額外維護(hù)服務(wù)進(jìn)程和權(quán)限體系第二分析任務(wù)多為單機(jī)批量處理如月度趨勢(shì)回溯、站點(diǎn)異常聚類并發(fā)寫入壓力幾乎為零第三交付物常需打包給第三方復(fù)核——SQLite 的.db文件即數(shù)據(jù)庫(kù)雙擊可查、拖拽即走、Git 可追蹤配合.gitattributes設(shè)置*.db filterlfs。我們實(shí)測(cè)過10 年、300 個(gè)站點(diǎn)、每小時(shí) 6 項(xiàng)指標(biāo)的全量數(shù)據(jù)約 2.6 億行SQLite 插入速度達(dá) 12,000 行/秒SSD查詢響應(yīng) 200ms加索引后完全覆蓋常規(guī)分析場(chǎng)景。若未來需接入 Kafka 流或支撐百人并發(fā)看板再平滑遷移到 TimescaleDB但起步階段過度設(shè)計(jì)就是給自己挖坑。2.2 核心表結(jié)構(gòu)設(shè)計(jì)緊扣“控制質(zhì)量數(shù)據(jù)”本質(zhì)的四張表平臺(tái)數(shù)據(jù)庫(kù)僅含 4 張表拒絕冗余建模。所有字段命名遵循snake_case單位統(tǒng)一標(biāo)注在注釋中非字段名避免pm25_ug_m3這類自解釋型命名帶來的長(zhǎng)度災(zāi)難表名字段主鍵標(biāo) *說明約束stationsid*,code,name,lat,lng,type,status站點(diǎn)元信息code為國(guó)控站唯一編碼如1001Atype區(qū)分urban,suburban,roadsidecode唯一索引parametersid*,code,name,unit,min_limit,max_limit污染物參數(shù)定義code如PM25,NO2,O3_8Hmin_limit/max_limit為國(guó)標(biāo)限值μg/m3code唯一索引raw_recordsid*,station_id,param_id,dt,value,qc_flag原始監(jiān)測(cè)記錄dt為TEXT類型存 ISO86012023-01-01 00:00:00不使用 DATETIME避坑見 3.3qc_flag為整數(shù)0原始值1人工修正2設(shè)備故障標(biāo)記復(fù)合索引(station_id, param_id, dt)cleaned_recordsid*,station_id,param_id,dt,value,reason清洗后權(quán)威數(shù)據(jù)reason記錄清洗動(dòng)作如interpolate:2023-01-05T03:00同上復(fù)合索引提示raw_records和cleaned_records分離是關(guān)鍵設(shè)計(jì)。前者只做最小化入庫(kù)不做任何計(jì)算后者才承載清洗邏輯。這保證了“原始數(shù)據(jù)不可篡改”的審計(jì)底線——哪怕清洗腳本出錯(cuò)也能回滾到 raw 表重跑。2.3 初始化腳本一鍵創(chuàng)建帶索引的空庫(kù)執(zhí)行以下命令生成結(jié)構(gòu)完備的初始庫(kù)無任何數(shù)據(jù)# 創(chuàng)建空數(shù)據(jù)庫(kù)并執(zhí)行建表語句 sqlite3 air_quality.db schema.sql其中schema.sql內(nèi)容如下已通過PRAGMA journal_mode WAL;啟用寫優(yōu)化-- 啟用 WAL 模式提升并發(fā)寫入性能雖單機(jī)為主但批量插入時(shí)有效 PRAGMA journal_mode WAL; -- 站點(diǎn)表 CREATE TABLE stations ( id INTEGER PRIMARY KEY AUTOINCREMENT, code TEXT NOT NULL UNIQUE, name TEXT NOT NULL, lat REAL CHECK(lat BETWEEN -90 AND 90), lng REAL CHECK(lng BETWEEN -180 AND 180), type TEXT CHECK(type IN (urban, suburban, roadside, background)), status TEXT DEFAULT active ); CREATE INDEX idx_stations_code ON stations(code); -- 參數(shù)表 CREATE TABLE parameters ( id INTEGER PRIMARY KEY AUTOINCREMENT, code TEXT NOT NULL UNIQUE, name TEXT NOT NULL, unit TEXT, min_limit REAL, max_limit REAL ); INSERT INTO parameters (code, name, unit, min_limit, max_limit) VALUES (PM25, 細(xì)顆粒物, μg/m3, 0, 75), (PM10, 可吸入顆粒物, μg/m3, 0, 150), (SO2, 二氧化硫, μg/m3, 0, 150), (NO2, 二氧化氮, μg/m3, 0, 80), (CO, 一氧化碳, mg/m3, 0, 4), (O3_8H, 臭氧8小時(shí)滑動(dòng)平均, μg/m3, 0, 160); -- 原始記錄表關(guān)鍵dt 為 TEXT非 DATETIME CREATE TABLE raw_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, station_id INTEGER NOT NULL, param_id INTEGER NOT NULL, dt TEXT NOT NULL, -- 存儲(chǔ) ISO8601 格式字符串便于時(shí)區(qū)無感處理 value REAL, qc_flag INTEGER DEFAULT 0 CHECK(qc_flag IN (0,1,2)) ); CREATE INDEX idx_raw_station_param_dt ON raw_records(station_id, param_id, dt); -- 清洗后記錄表 CREATE TABLE cleaned_records ( id INTEGER PRIMARY KEY AUTOINCREMENT, station_id INTEGER NOT NULL, param_id INTEGER NOT NULL, dt TEXT NOT NULL, value REAL NOT NULL, reason TEXT ); CREATE INDEX idx_cleaned_station_param_dt ON cleaned_records(station_id, param_id, dt);邏輯說明dt字段堅(jiān)持用TEXT而非DATETIME是因?yàn)?SQLite 的DATETIME不支持毫秒級(jí)精度且時(shí)區(qū)處理黑盒化而空氣質(zhì)量數(shù)據(jù)常需跨時(shí)區(qū)比對(duì)如對(duì)比本地站與鄰市站同一 UTC 時(shí)間點(diǎn)。存為 ISO8601 字符串后所有時(shí)間運(yùn)算交由 Python 的datetime或pandas處理可控性拉滿。qc_flag用整數(shù)而非布爾為后續(xù)擴(kuò)展留余地如 3校準(zhǔn)中4待復(fù)核。3. 數(shù)據(jù)清洗流水線從野蠻 CSV 到可信 time-series 的七步法3.1 輸入文件預(yù)檢用file和head快速識(shí)別編碼與分隔符空氣質(zhì)量數(shù)據(jù) CSV 常見陷阱GB2312 編碼的中文表頭、制表符\t分隔、BOM 頭導(dǎo)致 pandas 讀取失敗。先執(zhí)行# 檢測(cè)文件編碼與分隔符Linux/macOS file -i station_202301.csv head -n 3 station_202301.csv | cat -A # 顯示不可見字符典型輸出station_202301.csv: text/plain; charsetiso-8859-1 ^Mstation_code^Mstation_name^MPM2.5(μg/m3)^M...看到^MWindows 換行和charsetiso-8859-1立刻轉(zhuǎn)碼iconv -f GB2312 -t UTF-8 station_202301.csv station_202301_utf8.csv3.2 字段標(biāo)準(zhǔn)化用正則統(tǒng)一污染因子列名不同來源 CSV 的 PM2.5 列名可能為PM25,PM2.5,PM2.5(μg/m3),細(xì)顆粒物(μg/m3)。清洗腳本normalize_columns.py定義映射字典# normalize_columns.py import re import pandas as pd COLUMN_MAPPING { r(?i)pm2[.\s]*5.*: PM25, r(?i)pm10.*: PM10, r(?i)so2.*: SO2, r(?i)no2.*: NO2, r(?i)co.*: CO, r(?i)o3.*8.*h.*: O3_8H, r(?i)temperature.*: TEMP, r(?i)humidity.*: HUMI } def standardize_column_names(df): new_cols [] for col in df.columns: standardized UNKNOWN for pattern, target in COLUMN_MAPPING.items(): if re.search(pattern, str(col)): standardized target break new_cols.append(standardized) df.columns new_cols return df # 使用示例 df pd.read_csv(station_202301_utf8.csv, encodingutf-8) df standardize_column_names(df) print(df.columns) # 輸出Index([PM25, PM10, SO2, NO2, CO, O3_8H], dtypeobject)參數(shù)說明(?i)啟用忽略大小寫匹配[.\s]*匹配點(diǎn)號(hào)或空格零次或多次.*匹配任意后續(xù)字符。此正則能覆蓋 95% 的列名變體未命中的列自動(dòng)標(biāo)為UNKNOWN便于人工復(fù)查。3.3 時(shí)間戳解析強(qiáng)制轉(zhuǎn)換為 ISO8601 并校驗(yàn)合理性空氣質(zhì)量數(shù)據(jù)要求時(shí)間戳必須滿足1精確到小時(shí)2不能早于 2013 年中國(guó)國(guó)控網(wǎng)全面運(yùn)行年份3不能晚于當(dāng)前日期。腳本parse_datetime.py實(shí)現(xiàn)魯棒解析# parse_datetime.py from datetime import datetime, timedelta import pandas as pd import numpy as np def safe_parse_datetime(dt_series): 嘗試多種格式解析時(shí)間戳返回 ISO8601 字符串或 NaT 支持格式%Y-%m-%d %H:%M:%S, %Y/%m/%d %H:%M, %Y%m%d%H, %Y-%m-%d %H:%M formats [ %Y-%m-%d %H:%M:%S, %Y/%m/%d %H:%M, %Y%m%d%H, %Y-%m-%d %H:%M ] parsed pd.Series([pd.NaT] * len(dt_series)) for fmt in formats: try: temp pd.to_datetime(dt_series, formatfmt, errorscoerce) # 僅當(dāng)該格式成功解析出 50% 非空值時(shí)采用 if temp.notna().sum() / len(temp) 0.5: parsed temp break except: continue # 強(qiáng)制轉(zhuǎn)為 ISO8601 字符串無毫秒時(shí)區(qū)為 UTC0 parsed_str parsed.dt.strftime(%Y-%m-%d %H:00:00).replace(NaT, ) # 二次校驗(yàn)剔除明顯錯(cuò)誤時(shí)間如 2000 年前或未來 30 天 now datetime.now() valid_mask ( (parsed datetime(2013, 1, 1)) (parsed now timedelta(days30)) (parsed.dt.minute 0) # 必須整點(diǎn) (parsed.dt.second 0) ) parsed_str[~valid_mask] # 置空后續(xù)清洗步驟處理 return parsed_str # 使用示例 df[dt_iso] safe_parse_datetime(df[time_column])邏輯說明errorscoerce將無法解析的值轉(zhuǎn)為NaTnotna().sum() 0.5防止誤匹配如%Y%m%d格式會(huì)把20230101當(dāng)作 2023 年 1 月 1 日但若數(shù)據(jù)含12345678這類亂碼該格式匹配率低自動(dòng)跳過dt.minute 0強(qiáng)制整點(diǎn)因國(guó)控站數(shù)據(jù)發(fā)布粒度為小時(shí)。3.4 異常值清洗基于物理約束與統(tǒng)計(jì)雙閾值的三級(jí)過濾清洗不是簡(jiǎn)單刪掉超限值而是分層決策層級(jí)規(guī)則動(dòng)作依據(jù)L1 物理層value 0或value 2000PM2.5標(biāo)記qc_flag2設(shè)備故障儀器檢測(cè)下限為 0上限受傳感器飽和限制L2 國(guó)標(biāo)層value max_limit查parameters表標(biāo)記qc_flag1人工修正待定如PM25超 75 μg/m3 為超標(biāo)但可能是真實(shí)重污染L3 統(tǒng)計(jì)層value Q3 3*IQR且value Q1 - 3*IQR按站點(diǎn)參數(shù)滾動(dòng) 7 天標(biāo)記qc_flag0原始值但寫入cleaned_records時(shí)插值排除短期脈沖干擾如揚(yáng)塵、設(shè)備瞬時(shí)抖動(dòng)核心代碼outlier_detection.pyimport numpy as np from scipy import stats def detect_outliers_by_iqr(group, window_days7): 對(duì)單個(gè)站點(diǎn)參數(shù)的時(shí)間序列用滾動(dòng) IQR 檢測(cè)異常 # 按時(shí)間排序并設(shè)置 dt 為索引 group group.sort_values(dt).set_index(dt) # 轉(zhuǎn)換為 pandas 時(shí)間序列便于滾動(dòng)窗口 ts group[value].astype(float) # 計(jì)算滾動(dòng) 7 天的 Q1/Q3/IQR q1 ts.rolling(f{window_days}D, min_periods12).quantile(0.25) q3 ts.rolling(f{window_days}D, min_periods12).quantile(0.75) iqr q3 - q1 # 異常閾值 lower_bound q1 - 3 * iqr upper_bound q3 3 * iqr # 標(biāo)記異常點(diǎn)True 為異常 is_outlier (ts lower_bound) | (ts upper_bound) # 為每個(gè)異常點(diǎn)生成清洗理由 reasons [] for idx, outlier in is_outlier.items(): if outlier: reasons.append(fiqr_outlier:{window_days}D) else: reasons.append() return is_outlier, reasons # 使用示例對(duì)每個(gè) (station_id, param_id) 組合調(diào)用 grouped df.groupby([station_id, param_id]) for (sid, pid), group in grouped: is_outlier, reasons detect_outliers_by_iqr(group) # 后續(xù)將 is_outlier 為 True 的點(diǎn)進(jìn)行線性插值參數(shù)說明min_periods12確保至少有 12 個(gè)有效點(diǎn)半日數(shù)據(jù)才計(jì)算 IQR避免冷啟動(dòng)偏差f{window_days}D使用日歷天數(shù)而非工作日因空氣質(zhì)量有周末效應(yīng)。4. 統(tǒng)計(jì)分析與可視化從“平均值”到“污染過程歸因”的進(jìn)階表達(dá)4.1 污染過程識(shí)別用滑動(dòng)窗口突變點(diǎn)檢測(cè)定位真實(shí)污染事件單純看月均值會(huì)掩蓋關(guān)鍵信息。例如某市 1 月 PM2.5 均值 65 μg/m3看似達(dá)標(biāo)但實(shí)際存在 3 次持續(xù) 48 小時(shí)的重度污染過程150 μg/m3。我們用ruptures庫(kù)的 Pelt 算法檢測(cè)突變點(diǎn)# pollution_event_detection.py import ruptures as rpt import numpy as np import pandas as pd def detect_pollution_events(series, min_size12, penalty5): 檢測(cè)時(shí)間序列中的突變點(diǎn)識(shí)別污染過程起止 series: pd.Series, index 為 datetime, value 為 PM2.5 濃度 min_size: 最小突變間隔小時(shí)避免檢測(cè)毛刺 penalty: 斷點(diǎn)懲罰系數(shù)越大越少斷點(diǎn)推薦 3-10 # 轉(zhuǎn)為 numpy 數(shù)組 signal series.values.astype(float) # 使用 Pelt 算法適合長(zhǎng)序列 algo rpt.Pelt(modelrbf).fit(signal) result algo.predict(penpenalty) # 過濾掉間隔過小的斷點(diǎn) filtered_result [result[0]] for pt in result[1:]: if pt - filtered_result[-1] min_size: filtered_result.append(pt) # 將斷點(diǎn)轉(zhuǎn)為時(shí)間區(qū)間 events [] for i in range(len(filtered_result)-1): start_idx filtered_result[i] end_idx filtered_result[i1] - 1 if end_idx len(series): end_idx len(series) - 1 start_time series.index[start_idx] end_time series.index[end_idx] # 僅保留持續(xù)超標(biāo)75的區(qū)間 segment series.iloc[start_idx:end_idx1] if segment.max() 75: events.append({ start: start_time, end: end_time, duration_h: int((end_time - start_time).total_seconds() / 3600), max_conc: float(segment.max()), avg_conc: float(segment.mean()) }) return pd.DataFrame(events) # 使用示例對(duì)某站點(diǎn) PM2.5 全年數(shù)據(jù)檢測(cè) pm25_series df.query(param_id 1).set_index(dt)[value].sort_index() events_df detect_pollution_events(pm25_series, min_size12, penalty8) print(events_df.head())輸出示例start end duration_h max_conc avg_conc 0 2023-01-05 00:00:00 2023-01-07 23:00:00 72 215.3 142.6 1 2023-02-12 00:00:00 2023-02-14 23:00:00 72 189.7 128.4邏輯說明min_size12確保識(shí)別的是持續(xù)半天以上的污染過程排除單點(diǎn)儀器故障penalty8經(jīng)實(shí)測(cè)在 1 年數(shù)據(jù)中平衡了過檢太多短過程和漏檢合并長(zhǎng)過程最終只保留max_conc 75的區(qū)間緊扣國(guó)標(biāo)二級(jí)濃度限值。4.2 空間熱力圖用 GeoPandas Folium 實(shí)現(xiàn)動(dòng)態(tài)站點(diǎn)濃度渲染靜態(tài)地圖無法體現(xiàn)時(shí)間維度。我們生成交互式 HTML 地圖支持時(shí)間軸拖拽# spatial_viz.py import geopandas as gpd import folium from folium.plugins import TimestampedGeoJson import pandas as pd def create_interactive_map(stations_gdf, time_series_df, param_codePM25): stations_gdf: GeoDataFrame, 含 geometry, code, name time_series_df: pd.DataFrame, 含 dt, station_code, value # 合并站點(diǎn)位置與時(shí)間序列 merged stations_gdf.merge( time_series_df, left_oncode, right_onstation_code, howinner ) # 按時(shí)間分組生成 GeoJSON 特征集合 features [] for dt, group in merged.groupby(dt): # 每個(gè)時(shí)間點(diǎn)生成一個(gè) FeatureCollection feature_collection { type: FeatureCollection, features: [] } for _, row in group.iterrows(): feature { type: Feature, geometry: { type: Point, coordinates: [float(row.geometry.x), float(row.geometry.y)] }, properties: { time: dt.isoformat(), popup: f{row.name}br{param_code}: {row.value:.1f} μg/m3, icon: circle, iconstyle: { fillColor: get_color_by_value(row.value), fillOpacity: 0.8, stroke: False, radius: 8 (row.value / 100) * 12 # 濃度越高圓圈越大 } } } feature_collection[features].append(feature) features.append(feature_collection) # 創(chuàng)建基礎(chǔ)地圖 m folium.Map(location[35, 105], zoom_start4, tilesCartoDB positron) # 添加時(shí)間序列圖層 TimestampedGeoJson( {type: FeatureCollection, features: features}, periodPT1H, # 每小時(shí)切換 add_last_pointTrue, auto_playFalse, loopFalse, max_speed1, loop_buttonTrue, date_optionsYYYY-MM-DD HH:mm, time_slider_drag_updateTrue ).add_to(m) return m def get_color_by_value(value): 根據(jù)濃度返回顏色綠-黃-橙-紅 if value 35: return #008000 # 優(yōu) elif value 75: return #FFD700 # 良 elif value 115: return #FF8C00 # 輕度 elif value 150: return #FF4500 # 中度 else: return #8B0000 # 重度 # 使用示例 # stations_gdf gpd.read_file(stations.geojson) # time_series_df pd.read_sql(SELECT dt, s.code as station_code, c.value FROM cleaned_records c JOIN stations s ON c.station_ids.id WHERE c.param_id1, conn) # m create_interactive_map(stations_gdf, time_series_df) # m.save(pm25_timeline.html)參數(shù)說明periodPT1H遵循 ISO8601 持續(xù)時(shí)間格式表示每小時(shí)更新radius動(dòng)態(tài)縮放確保低濃度站點(diǎn)不被淹沒get_color_by_value嚴(yán)格對(duì)應(yīng)《環(huán)境空氣質(zhì)量標(biāo)準(zhǔn)》GB 3095-2012六級(jí)劃分非主觀設(shè)定。4.3 污染源歸因初探用相關(guān)性矩陣鎖定協(xié)同變化污染物PM2.5 往往不是獨(dú)立出現(xiàn)而是與 SO2、NO2 等協(xié)同升高。我們計(jì)算各污染物兩兩間的 Spearman 相關(guān)系數(shù)抗異常值# source_apportionment.py import pandas as pd import numpy as np from scipy.stats import spearmanr def calculate_correlation_matrix(df, param_codes[PM25,SO2,NO2,CO,O3_8H]): 計(jì)算指定污染物在全站點(diǎn)的平均 Spearman 相關(guān)系數(shù)矩陣 df: cleaned_records 表導(dǎo)出的 DataFrame含 station_id, param_id, dt, value # pivot 為寬表indexdt, columnsparam_code, valuesvalue pivot_df df.pivot_table( indexdt, columnsparam_code, valuesvalue, aggfuncmean # 同一時(shí)刻多站點(diǎn)取均值 ).dropna() # 剔除任一污染物缺失的時(shí)刻 # 計(jì)算 Spearman 相關(guān)矩陣 corr_matrix pd.DataFrame( np.zeros((len(param_codes), len(param_codes))), indexparam_codes, columnsparam_codes ) for i, p1 in enumerate(param_codes): for j, p2 in enumerate(param_codes): if i j: corr_matrix.iloc[i, j] 1.0 else: # 計(jì)算非空配對(duì) valid_mask pivot_df[[p1, p2]].notna().all(axis1) if valid_mask.sum() 10: # 至少 10 個(gè)有效點(diǎn) corr, _ spearmanr( pivot_df.loc[valid_mask, p1], pivot_df.loc[valid_mask, p2] ) corr_matrix.iloc[i, j] round(corr, 3) else: corr_matrix.iloc[i, j] np.nan return corr_matrix # 使用示例 corr_df calculate_correlation_matrix(cleaned_df) print(corr_df)輸出示例對(duì)某城市PM25 SO2 NO2 CO O3_8H PM25 1.000 0.623 0.781 0.692 -0.412 SO2 0.623 1.000 0.512 0.587 -0.321 NO2 0.781 0.512 1.000 0.725 -0.503 CO 0.692 0.587 0.725 1.000 -0.445 O3_8H -0.412 -0.321 -0.503 -0.445 1.000解讀PM25與NO2相關(guān)系數(shù) 0.781顯著高于與SO20.623暗示機(jī)動(dòng)車排放NO2 主要來源對(duì)該市 PM2.5 貢獻(xiàn)可能大于燃煤SO2 主要來源O3_8H與所有顆粒物呈負(fù)相關(guān)符合光化學(xué)反應(yīng)規(guī)律高臭氧常伴隨晴朗天氣顆粒物沉降。這為后續(xù)受體模型如 PMF提供強(qiáng)假設(shè)入口。5. 避坑指南數(shù)據(jù)清洗與分析中五個(gè)血淚教訓(xùn)5.1 現(xiàn)象清洗后數(shù)據(jù)量比原始 CSV 少 30%排查發(fā)現(xiàn)raw_records表中大量dt字段為空原因CSV 中時(shí)間列名為監(jiān)測(cè)時(shí)間但normalize_columns.py的正則未覆蓋中文導(dǎo)致dt列未被識(shí)別safe_parse_datetime()輸入空 Series全部返回空字符串。解決在COLUMN_MAPPING中增加r(?i)監(jiān)測(cè)時(shí)間|date.*time.*: dt并添加預(yù)檢邏輯——若解析后dt_iso空值率 10%拋出ValueError并打印前 5 行原始時(shí)間列內(nèi)容強(qiáng)制人工介入。5.2 現(xiàn)象某站點(diǎn)PM25在 2023-03-15 全天恒為0.0清洗腳本未標(biāo)記導(dǎo)致月報(bào)中該站均值嚴(yán)重偏低原因L1 物理層規(guī)則value 0未覆蓋value 0的設(shè)備零點(diǎn)漂移場(chǎng)景且該站點(diǎn)當(dāng)日其他參數(shù)如TEMP,HUMI正常故未觸發(fā)設(shè)備故障標(biāo)記。解決增強(qiáng) L1 規(guī)則為value 0 and not (param_code in [TEMP, HUMI])并對(duì)PM25/PM10/SO2/NO2/CO/O3_8H這 6 項(xiàng)增加“連續(xù) N 小時(shí)恒為 0”的專項(xiàng)檢查N6標(biāo)記qc_flag2并記錄reasonzero_drift:6h。5.3 現(xiàn)象cleaned_records表中出現(xiàn)dt為2023-01-01 00:00:00和2023-01-01 00:00:00.000兩種格式導(dǎo)致后續(xù)GROUP BY dt統(tǒng)計(jì)結(jié)果重復(fù)原因部分?jǐn)?shù)據(jù)源時(shí)間戳含毫秒.000strftime(%Y-%m-%d %H:00:00)未截?cái)嗪撩攵鴖trftime(%Y-%m-%d %H:00:00)對(duì)含毫秒的datetime對(duì)象會(huì)保留.000。解決在safe_parse_datetime()末尾強(qiáng)制截?cái)鄍arsed_str parsed.dt.floor(H).dt.strftime(%Y-%m-%d %H:00:00)floor(H)將毫秒、分鐘歸零確保格式絕對(duì)統(tǒng)一。5.4 現(xiàn)象folium生成的地圖加載極慢Chrome 控制臺(tái)報(bào)RangeError: Maximum call stack size exceeded原因TimestampedGeoJson默認(rèn)將所有時(shí)間點(diǎn)的 GeoJSON 一次性注入 HTML1 年數(shù)據(jù)8760 小時(shí)生成的 JSON 體積超 200MB瀏覽器內(nèi)存溢出。解決啟用periodP1D按天切換并設(shè)置durationP1D同時(shí)用folium.GeoJson替代TimestampedGeoJson手動(dòng)實(shí)現(xiàn)時(shí)間軸控件每次只加載當(dāng)天數(shù)據(jù) 2MB。5.5 現(xiàn)象spearmanr計(jì)算PM25與O3_8H相關(guān)系數(shù)為-0.05但業(yè)務(wù)方反饋“臭氧高時(shí)顆粒物必然低”懷疑統(tǒng)計(jì)失效原因O3_8H是 8 小時(shí)滑動(dòng)平均而PM25是小時(shí)值直接對(duì)齊導(dǎo)致時(shí)間尺度錯(cuò)配。例如O3_8H在2023-01-01 12:00的值代表05:00-12:00平均但PM25取的是12:00單點(diǎn)。解決對(duì)O3_8H列先用pandas.Series.rolling(8).mean()重算小時(shí)級(jí)滑動(dòng)均值再與PM25對(duì)齊或改用O3小時(shí)值參與計(jì)算明確標(biāo)注“此處分析的是臭氧小時(shí)濃度與 PM2.5 的即時(shí)關(guān)聯(lián)”。6. 進(jìn)階技巧用 Airflow 編排全鏈路讓清洗分析真正“無人值守”6.1 DAG 設(shè)計(jì)四個(gè)原子任務(wù)串聯(lián)成可靠 pipelineAirflow 不是用來炫技的而是解決“誰來記得每月 1 號(hào)跑清洗腳本”這種管理問題。我們定義最簡(jiǎn) DAG僅含 4 個(gè)任務(wù)任務(wù)名依賴執(zhí)行動(dòng)作超時(shí)重試check_new_data無檢查data/incoming/目錄是否有新 CSV按文件名含202312判斷300s2ingest_to_rawcheck_new_data成功調(diào)用ingest_raw.py將 CSV 批量插入raw_records1800s1run_cleaningingest_to_raw成功調(diào)用clean_all.py執(zhí)行全量清洗寫入cleaned_records3600s1generate_reportrun_cleaning成功調(diào)用report_gen.py生成 PDF 月報(bào) HTML本文還有配套的精品資源點(diǎn)擊獲取