3个坑!全国宜居城市排名源码解析实战 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标准化? 或者你有更好的权重分配方法? 评论区交流,看看大家怎么避坑。