3个坑!全国宜居城市排名源码解析实战
面试被问原理答不上来？别慌。
刚入职做数据项目，领导甩来个全国宜居城市排名需求，我愣住。
以为只是查个数据排序，结果发现坑多到怀疑人生。
这文章不讲虚的，直接上源码解析。
带你从零搭建这个实战项目，避开我踩过的所有雷。
全程Python，代码可复现，看完就能上手。
项目目标与需求拆解
先说清楚我们要做什么。
不是简单拉个Excel排个序，而是构建一个可复用的评估系统。
核心指标必须量化，不能靠拍脑袋。
空气指数、医疗资源、交通便利度、绿化覆盖率、生活成本。
这五项权重不一样，得用专家打分法确定。
很多新手第一步就错。
直接找网上现成排名数据，拿来就分析。
结果发现数据是2019年的，早就过时了。
更坑的是，不同城市统计口径根本不一样。
真实场景是这样的：
你给甲方交付报告，他问“成都和杭州谁更宜居？”
你拿2018年的数据回答，直接被怼。
这时候你才意识到，数据清洗和实时性才是核心竞争力。
项目目标很明确：搭建数据获取管道，支持多源数据整合
实现权重计算引擎，支持动态调整
生成可视化报告，一键导出PDF
代码模块化，方便后续扩展新指标别小看这四个目标。
我最初只做了前两个，结果甲方要可视化，又加了三天班。
前期规划不清晰，后期返工是常态。
目录结构设计
工程化思维，从目录结构开始。
别把所有代码堆在一个文件里，那是实习生干的事。
city_ranking/
├── config/
│ └── weights.yaml # 权重配置文件
├── data/
│ ├── raw/ # 原始数据
│ └── processed/ # 清洗后数据
├── src/
│ ├── __init__.py
│ ├── data_fetcher.py # 数据获取
│ ├── data_cleaner.py # 数据清洗
│ ├── 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版本要锁定，不然同事环境跑不起来。
我吃过亏，没锁版本，本地跑得好好的，服务器报错。
生产环境，版本控制是底线。
核心代码实现
进入正题，看关键代码。
这部分是源码解析的核心，逐行讲解。
数据获取模块
# 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:获取空气指数数据# 模拟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]]# 保存原始数据output_file = self.raw_dir / air_quality.csvdf.to_csv(output_file, index=False)return dfexcept requests.exceptions.RequestException as e:print(f获取空气数据失败: {e})return pd.DataFrame()逐行解析：Path.mkdir(parents=True, exist_ok=True)：确保目录存在，不报错
timeout=10：必须设置超时，不然网络卡住整个程序就挂了
raise_for_status()：HTTP错误抛异常，别静默失败
只保留必要字段：减少存储，加快后续处理这里有个坑，我最初没处理异常。
某次API挂了，程序直接崩溃，数据全丢。
网络请求，异常处理是标配。
评分引擎
这是核心中的核心，源码解析重点看这里。
# 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项目，权重分配经过专家论证。
我直接借鉴，省得自己研究。
数据清洗
# src/data_cleaner.py
import pandas as pdclass DataCleaner:def clean_city_data(self, df: pd.DataFrame) - pd.DataFrame:清洗城市数据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'保留第一条，实际项目要确认哪条更准确
缺失值填充：用中位数，比均值更抗异常值
异常值检测：3倍标准差法，简单有效
名称标准化：'北京市'和'北京'要统一，不然排序错乱我最初没做名称标准化，结果北京市排在第50名，
北京排在第10名，同一个城市两个排名，被领导骂惨。
数据一致性，细节决定成败。
运行与测试
代码写完，跑起来。
别直接上生产，先本地测试。
主入口
# 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. 获取数据fetcher = DataFetcher()air_data = fetcher.fetch_air_quality()if air_data.empty:logger.error(获取数据失败，程序退出)return# 2. 数据清洗cleaner = DataCleaner()clean_data = cleaner.clean_city_data(air_data)logger.info(f清洗后数据量: {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. 创建虚拟环境
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):测试归一化函数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，才能放心提交代码。
没测试的代码，都是耍流氓。
优化扩展
项目跑通了，怎么让它更强大？
性能优化
数据量小的时候，性能不是问题。
但要是扩展到全国300+城市，数据量上百万，就得优化了。
几个实用技巧：并行处理：用joblib并行清洗数据
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)缓存机制：数据获取结果缓存，避免重复请求
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增量更新：只更新变化的数据，不全量刷新功能扩展
新指标接入：
比如加个教育资源指标，步骤：在data_fetcher.py加fetch_education()方法
在weights.yaml加education: 0.15
调整其他指标权重，总和保持1.0
重新运行，自动生成新排名动态权重调整：
前端加个滑块，用户拖动调整权重，实时看排名变化。
这个功能我做了，甲方很满意，项目验收顺利通过。
多场景对比：
支持老人宜居、年轻人宜居、家庭宜居等不同场景。
不同场景权重不同，一键切换。
小结
这个项目做完，我最大的感受：别高估简单需求，别低估细节处理。
全国宜居城市排名，听起来就是排个序。
实际做下来，数据清洗占了40%时间，测试占了20%。
真正的排名逻辑，反而最简单。
几个核心经验：数据质量 算法复杂度：脏数据喂进模型，出来的是垃圾
配置驱动设计：权重、参数放配置文件，别硬编码
异常处理是底线：网络请求、文件读写，都要try-except
测试不是可选：评分逻辑出错，整个项目就废了这个项目代码在GitHub上开源了，地址在文末。
你可以fork下来，跑一跑，改一改。
加个新指标，调个权重，看看排名怎么变。
动手比看一百遍教程都有用。
你更常用哪种写法？Min-Max归一化还是Z-Score标准化？
或者你有更好的权重分配方法？
评论区交流，看看大家怎么避坑。
