
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 Path
class 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 df
except 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 np
class 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 normalized
def 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 + 1
return 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.3
medical_resource: 0.25
transportation: 0.2
green_coverage: 0.15
living_cost: 0.1
normalize_method: min_max
为什么权重这样设?
这不是拍脑袋,参考了官方源码仓库里开源项目的评估标准。
GitHub上有个china-city-evaluation项目,权重分配经过专家论证。
我直接借鉴,省得自己研究。
数据清洗
# src/data_cleaner.py
import pandas as pd
class 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_val
upper_bound = mean_val + 3 * std_val
mask = (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 logging
logging.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)
raise
if __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 ScoringEngine
class 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] == 1
assert len(result) == 3
if __name__ == __main__:
pytest.main([__file__, -v])
测试覆盖:
正常归一化
边界情况(所有值相同)
完整评分流程
排名正确性
跑测试命令:
pytest tests/ -v
看到所有PASSED,才能放心提交代码。
没测试的代码,都是耍流氓。
优化扩展
项目跑通了,怎么让它更强大?
性能优化
数据量小的时候,性能不是问题。
但要是扩展到全国300+城市,数据量上百万,就得优化了。
几个实用技巧:
并行处理:用joblib并行清洗数据
from joblib import Parallel, delayed
def 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 Path
def 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 time
if 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标准化?
或者你有更好的权重分配方法?
评论区交流,看看大家怎么避坑。