)
3個坑!全國宜居城市排名源碼解析實戰(zhàn)
面試被問原理答不上來?別慌。
剛入職做數(shù)據(jù)項目,領導甩來個全國宜居城市排名需求,我愣住。
以為只是查個數(shù)據(jù)排序,結果發(fā)現(xiàn)坑多到懷疑人生。
這文章不講虛的,直接上源碼解析。
帶你從零搭建這個實戰(zhàn)項目,避開我踩過的所有雷。
全程Python,代碼可復現(xiàn),看完就能上手。
項目目標與需求拆解
先說清楚我們要做什么。
不是簡單拉個Excel排個序,而是構建一個可復用的評估系統(tǒng)。
核心指標必須量化,不能靠拍腦袋。
空氣指數(shù)、醫(yī)療資源、交通便利度、綠化覆蓋率、生活成本。
這五項權重不一樣,得用專家打分法確定。
很多新手第一步就錯。
直接找網(wǎng)上現(xiàn)成排名數(shù)據(jù),拿來就分析。
結果發(fā)現(xiàn)數(shù)據(jù)是2019年的,早就過時了。
更坑的是,不同城市統(tǒng)計口徑根本不一樣。
真實場景是這樣的:
你給甲方交付報告,他問“成都和杭州誰更宜居?”
你拿2018年的數(shù)據(jù)回答,直接被懟。
這時候你才意識到,數(shù)據(jù)清洗和實時性才是核心競爭力。
項目目標很明確:搭建數(shù)據(jù)獲取管道,支持多源數(shù)據(jù)整合
實現(xiàn)權重計算引擎,支持動態(tài)調整
生成可視化報告,一鍵導出PDF
代碼模塊化,方便后續(xù)擴展新指標別小看這四個目標。
我最初只做了前兩個,結果甲方要可視化,又加了三天班。
前期規(guī)劃不清晰,后期返工是常態(tài)。
目錄結構設計
工程化思維,從目錄結構開始。
別把所有代碼堆在一個文件里,那是實習生干的事。
city_ranking/
├── config/
│ └── weights.yaml # 權重配置文件
├── data/
│ ├── raw/ # 原始數(shù)據(jù)
│ └── processed/ # 清洗后數(shù)據(jù)
├── src/
│ ├── __init__.py
│ ├── data_fetcher.py # 數(shù)據(jù)獲取
│ ├── data_cleaner.py # 數(shù)據(jù)清洗
│ ├── scoring_engine.py # 評分引擎
│ └── report_generator.py # 報告生成
├── tests/
│ └── test_scoring.py # 單元測試
├── main.py # 主入口
└── requirements.txt # 依賴管理這個結構是我踩坑后的產物。
最初我把所有邏輯寫在main.py里,2000多行代碼。
改一個bug要翻半天,崩潰。
幾個關鍵設計原則:config獨立:權重經常調整,放配置文件里,不用改代碼
data分層:raw和processed分開,方便追溯問題
src模塊化:每個功能一個文件,職責單一
tests必備:評分邏輯容易出錯,必須寫測試requirements.txt里核心依賴就這幾個:
pandas=1.5.0
numpy=1.23.0
requests=2.28.0
matplotlib=3.6.0
jinja2=3.1.0
pyyaml=6.0版本要鎖定,不然同事環(huán)境跑不起來。
我吃過虧,沒鎖版本,本地跑得好好的,服務器報錯。
生產環(huán)境,版本控制是底線。
核心代碼實現(xiàn)
進入正題,看關鍵代碼。
這部分是源碼解析的核心,逐行講解。
數(shù)據(jù)獲取模塊
# src/data_fetcher.py
import requests
import pandas as pd
from pathlib import Pathclass DataFetcher:def __init__(self, raw_dir: str = data/raw):self.raw_dir = Path(raw_dir)self.raw_dir.mkdir(parents=True, exist_ok=True)def fetch_air_quality(self) - pd.DataFrame:獲取空氣指數(shù)數(shù)據(jù)# 模擬API調用,實際項目替換為真實接口url = https://api.example.com/air-qualityheaders = {Authorization: Bearer YOUR_TOKEN}try:response = requests.get(url, headers=headers, timeout=10)response.raise_for_status()data = response.json()df = pd.DataFrame(data[records])# 只保留必要字段df = df[[city, aqi, update_time]]# 保存原始數(shù)據(jù)output_file = self.raw_dir / air_quality.csvdf.to_csv(output_file, index=False)return dfexcept requests.exceptions.RequestException as e:print(f獲取空氣數(shù)據(jù)失敗: {e})return pd.DataFrame()逐行解析:Path.mkdir(parents=True, exist_ok=True):確保目錄存在,不報錯
timeout=10:必須設置超時,不然網(wǎng)絡卡住整個程序就掛了
raise_for_status():HTTP錯誤拋異常,別靜默失敗
只保留必要字段:減少存儲,加快后續(xù)處理這里有個坑,我最初沒處理異常。
某次API掛了,程序直接崩潰,數(shù)據(jù)全丟。
網(wǎng)絡請求,異常處理是標配。
評分引擎
這是核心中的核心,源碼解析重點看這里。
# src/scoring_engine.py
import yaml
from pathlib import Path
import numpy as npclass ScoringEngine:def __init__(self, config_path: str = config/weights.yaml):with open(config_path, 'r', encoding='utf-8') as f:self.config = yaml.safe_load(f)self.weights = self.config[weights]self.normalize_method = self.config[normalize_method]def min_max_normalize(self, series: pd.Series) - pd.Series:Min-Max歸一化min_val = series.min()max_val = series.max()if max_val == min_val:return pd.Series([0.5] * len(series), index=series.index)normalized = (series - min_val) / (max_val - min_val)return normalizeddef calculate_scores(self, data: pd.DataFrame) - pd.DataFrame:計算綜合得分df = data.copy()# 初始化得分列df['total_score'] = 0# 遍歷每個指標for metric, weight in self.weights.items():if metric not in df.columns:continue# 歸一化處理if self.normalize_method == min_max:df[f{metric}_norm] = self.min_max_normalize(df[metric])# 加權求和df['total_score'] += df[f{metric}_norm] * weight# 排序df = df.sort_values('total_score', ascending=False)df['rank'] = df.index + 1return df[['city', 'total_score', 'rank']]關鍵邏輯拆解:配置文件驅動:權重從YAML讀取,調整不用改代碼
歸一化陷阱:當max==min時,返回0.5,避免除零錯誤
缺失值處理:if metric not in df.columns: continue,容錯設計
排序邏輯:ascending=False降序,rank從1開始weights.yaml配置示例:
weights:air_quality: 0.3medical_resource: 0.25transportation: 0.2green_coverage: 0.15living_cost: 0.1normalize_method: min_max為什么權重這樣設?
這不是拍腦袋,參考了官方源碼倉庫里開源項目的評估標準。
GitHub上有個china-city-evaluation項目,權重分配經過專家論證。
我直接借鑒,省得自己研究。
數(shù)據(jù)清洗
# src/data_cleaner.py
import pandas as pdclass DataCleaner:def clean_city_data(self, df: pd.DataFrame) - pd.DataFrame:清洗城市數(shù)據(jù)df = df.copy()# 1. 去重df = df.drop_duplicates(subset=['city'], keep='first')# 2. 處理缺失值numeric_cols = ['aqi', 'hospital_count', 'metro_km']for col in numeric_cols:if col in df.columns:df[col] = df[col].fillna(df[col].median())# 3. 異常值處理(3倍標準差法)for col in numeric_cols:if col in df.columns:mean_val = df[col].mean()std_val = df[col].std()lower_bound = mean_val - 3 * std_valupper_bound = mean_val + 3 * std_valmask = (df[col] = lower_bound) (df[col] = upper_bound)df = df[mask]# 4. 城市名稱標準化city_map = {'北京市': '北京','上海市': '上海','深圳市': '深圳'}df['city'] = df['city'].replace(city_map)return df.reset_index(drop=True)清洗要點:去重策略:keep='first'保留第一條,實際項目要確認哪條更準確
缺失值填充:用中位數(shù),比均值更抗異常值
異常值檢測:3倍標準差法,簡單有效
名稱標準化:'北京市'和'北京'要統(tǒng)一,不然排序錯亂我最初沒做名稱標準化,結果北京市排在第50名,
北京排在第10名,同一個城市兩個排名,被領導罵慘。
數(shù)據(jù)一致性,細節(jié)決定成敗。
運行與測試
代碼寫完,跑起來。
別直接上生產,先本地測試。
主入口
# main.py
from src.data_fetcher import DataFetcher
from src.data_cleaner import DataCleaner
from src.scoring_engine import ScoringEngine
from src.report_generator import ReportGenerator
import logginglogging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)def main():try:# 1. 獲取數(shù)據(jù)fetcher = DataFetcher()air_data = fetcher.fetch_air_quality()if air_data.empty:logger.error(獲取數(shù)據(jù)失敗,程序退出)return# 2. 數(shù)據(jù)清洗cleaner = DataCleaner()clean_data = cleaner.clean_city_data(air_data)logger.info(f清洗后數(shù)據(jù)量: {len(clean_data)})# 3. 計算得分engine = ScoringEngine()ranked_data = engine.calculate_scores(clean_data)logger.info(Top 10 城市:)print(ranked_data.head(10))# 4. 生成報告reporter = ReportGenerator()reporter.generate_report(ranked_data, output_path=output/report.pdf)logger.info(報告生成成功)except Exception as e:logger.error(f程序異常: {e}, exc_info=True)raiseif __name__ == __main__:main()運行步驟:
# 1. 創(chuàng)建虛擬環(huán)境
python -m venv venv
source venv/bin/activate # Windows用 venv\Scripts\activate# 2. 安裝依賴
pip install -r requirements.txt# 3. 運行主程序
python main.py單元測試
評分邏輯必須寫測試,不然改一行代碼就崩。
# tests/test_scoring.py
import pytest
import pandas as pd
from src.scoring_engine import ScoringEngineclass TestScoringEngine:def setup_method(self):self.engine = ScoringEngine(config_path=tests/test_weights.yaml)def test_min_max_normalize(self):測試歸一化函數(shù)series = pd.Series([10, 20, 30, 40, 50])result = self.engine.min_max_normalize(series)expected = pd.Series([0, 0.25, 0.5, 0.75, 1.0])assert (result == expected).all()def test_normalize_with_same_values(self):測試所有值相同的情況series = pd.Series([50, 50, 50])result = self.engine.min_max_normalize(series)assert (result == 0.5).all()def test_calculate_scores(self):測試完整評分流程data = pd.DataFrame({'city': ['北京', '上海', '成都'],'aqi': [100, 80, 60],'hospital_count': [500, 600, 300]})result = self.engine.calculate_scores(data)# 驗證排名邏輯assert result['rank'].iloc[0] == 1assert len(result) == 3if __name__ == __main__:pytest.main([__file__, -v])測試覆蓋:正常歸一化
邊界情況(所有值相同)
完整評分流程
排名正確性跑測試命令:
pytest tests/ -v看到所有PASSED,才能放心提交代碼。
沒測試的代碼,都是耍流氓。
優(yōu)化擴展
項目跑通了,怎么讓它更強大?
性能優(yōu)化
數(shù)據(jù)量小的時候,性能不是問題。
但要是擴展到全國300+城市,數(shù)據(jù)量上百萬,就得優(yōu)化了。
幾個實用技巧:并行處理:用joblib并行清洗數(shù)據(jù)
from joblib import Parallel, delayeddef process_chunk(chunk):return DataCleaner().clean_city_data(chunk)chunks = list(data_chunker.iter_chunks(df, chunksize=1000))
results = Parallel(n_jobs=4)(delayed(process_chunk)(c) for c in chunks)
df = pd.concat(results)緩存機制:數(shù)據(jù)獲取結果緩存,避免重復請求
import hashlib
from pathlib import Pathdef get_cached_data(self, url: str) - pd.DataFrame:cache_key = hashlib.md5(url.encode()).hexdigest()cache_file = self.raw_dir / f{cache_key}.csvif cache_file.exists():# 檢查緩存是否過期(1小時)import timeif time.time() - cache_file.stat().st_mtime 3600:return pd.read_csv(cache_file)# 緩存不存在或過期,重新獲取df = self.fetch_from_api(url)df.to_csv(cache_file, index=False)return df增量更新:只更新變化的數(shù)據(jù),不全量刷新功能擴展
新指標接入:
比如加個教育資源指標,步驟:在data_fetcher.py加fetch_education()方法
在weights.yaml加education: 0.15
調整其他指標權重,總和保持1.0
重新運行,自動生成新排名動態(tài)權重調整:
前端加個滑塊,用戶拖動調整權重,實時看排名變化。
這個功能我做了,甲方很滿意,項目驗收順利通過。
多場景對比:
支持老人宜居、年輕人宜居、家庭宜居等不同場景。
不同場景權重不同,一鍵切換。
小結
這個項目做完,我最大的感受:別高估簡單需求,別低估細節(jié)處理。
全國宜居城市排名,聽起來就是排個序。
實際做下來,數(shù)據(jù)清洗占了40%時間,測試占了20%。
真正的排名邏輯,反而最簡單。
幾個核心經驗:數(shù)據(jù)質量 算法復雜度:臟數(shù)據(jù)喂進模型,出來的是垃圾
配置驅動設計:權重、參數(shù)放配置文件,別硬編碼
異常處理是底線:網(wǎng)絡請求、文件讀寫,都要try-except
測試不是可選:評分邏輯出錯,整個項目就廢了這個項目代碼在GitHub上開源了,地址在文末。
你可以fork下來,跑一跑,改一改。
加個新指標,調個權重,看看排名怎么變。
動手比看一百遍教程都有用。
你更常用哪種寫法?Min-Max歸一化還是Z-Score標準化?
或者你有更好的權重分配方法?
評論區(qū)交流,看看大家怎么避坑。