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-quality"headers = {"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.csv"df.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}.csv"if 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标准化？ 或者你有更好的权重分配方法？ 评论区交流，看看大家怎么避坑。
