这次我们来看一个反直觉的趋势:人类医疗技术的进步,大幅提高了“活得更久”的概率,却没有同步提高“活得更健康”的预期。英文里一句话概括得很到位:We Got Better at Keeping You Alive, Not at Keeping You Healthy。
这句话放到技术语境里,不能只停留在感慨层面。它可以用公开健康数据集量化拆解:预期寿命(Lifespan)和健康预期寿命(Healthy Life Expectancy,HALE)之间的缺口,正在变大还是变小?哪些区域的缺口更明显?医疗资源的增加是否真的流向了“健康生存时间”的延长?
本文就带大家完成一次可复用的数据分析项目。我们从数据源选择、指标口径、脚本实现、批量拉取、可视化输出到问题排查,完整走一遍。这个项目不需要 GPU,不需要大数据平台,一台普通笔记本、一个 Python 环境就能跑。重点是帮你建立一套“用数据观察健康趋势”的分析框架,之后可以套用到其他公开健康数据源。
1. 核心能力速览
| 能力项 | 说明 |
|---|---|
| 项目主题 | 预期寿命与健康预期寿命缺口分析 |
| 数据来源 | 世卫组织(WHO)HALE 数据、世界银行预期寿命公开指标等 |
| 主要功能 | 数据下载、清洗、缺口计算、年份趋势对比、区域对比、图表输出 |
| 推荐硬件 | 普通 x86 电脑,4G 内存以上即可,不需要独立显卡 |
| 支持系统 | Windows / macOS / Linux |
| 启动方式 | Python 命令行脚本,或 Jupyter Notebook 分步执行 |
| 是否支持 API | 支持,可调用公开数据接口批量拉取 |
| 是否支持批量任务 | 支持,可按国家/年份批量抓取数据并缓存 |
| 输出物 | 清洗后的 CSV、缺口分析图表、汇总统计表 |
| 适用人群 | 数据分析师、公共卫生研究者、医疗政策评估人员、健康数据爱好者 |
这里要先明确一个概念:预期寿命是指“从出生开始平均可以活多少年”,健康预期寿命则是指“平均可以在完全健康状态下生活多少年”。两者的差值,可以理解为“带病生存时间”或“健康缺口”。这个缺口不是越小越好吗?不一定,实际分析中要结合医疗条件、慢性病患病率、老年人口结构等因素综合判断,不能只看一个差值就下结论。
2. 适用场景与使用边界
这个分析项目适合以下几类场景:
- 公共卫生领域研究者:快速查看某个国家或地区在健康寿命方面的长期变化趋势。
- 医疗政策评估人员:对比不同区域“寿命增长”和“健康寿命增长”是否同步,评估医疗投入是否过度集中在“延长生命”而忽视了“提高生命质量”。
- 数据分析教学案例:数据量不大、指标含义清晰、可视化成图直观,适合作为 Pandas 和 Matplotlib 的实操练习。
- 媒体和健康科普内容创作者:在写“寿命更长不等于更健康”这类文章时,用统一口径的数据支撑观点。
也有明确的边界:
- 该分析只面向宏观群体数据,不能用于个体医疗诊断或健康预测。
- 不使用任何未授权的患者数据、医院内部数据,只使用公开的、可合法再分析的统计数据。
- 不同数据源的年份覆盖、指标口径可能存在差异,不能把不同来源的数据直接混在一起做趋势图而不做口径说明。
- 健康预期寿命估算依赖调查问卷、疾病登记、生命表等间接数据,不同国家的统计质量不同,分析结果可能存在系统偏差。
涉及健康数据时必须强调合规:如果之后把分析范围扩展到地区级、医院级或个体级数据,一定要先确认数据授权范围,完成脱敏和合规审查,再进入建模和发布流程。
3. 环境准备与前置条件
3.1 基础运行环境
本次分析只需要 Python 3.9 以上版本,建议使用虚拟环境隔离依赖。
核心依赖:
| 库 | 用途 |
|---|---|
| pandas | 数据读取、清洗、透视、合并 |
| matplotlib | 绘制折线图、柱状图、差值面积图 |
| requests | 调用公开 API 拉取数据 |
| openpyxl | 可选,导出 Excel 汇总表 |
| tabulate | 可选,在终端打印 Markdown 表格 |
如果你正在使用 Anaconda 环境,大部分依赖已经具备,只需要额外安装缺失的部分。
3.2 创建虚拟环境
在项目根目录执行:
python -m venv venvWindows 系统激活虚拟环境:
venv\Scripts\activatemacOS / Linux 系统激活虚拟环境:
source venv/bin/activate接着创建依赖文件requirements.txt:
pandas>=1.5.0 matplotlib>=3.6.0 requests>=2.31.0 openpyxl>=3.1.0 tabulate>=0.9.0安装依赖:
pip install -r requirements.txt如果网络环境有限,可以使用国内 PyPI 镜像源加速安装:
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple安装完成后,建议先确认核心库可正常导入:
python -c "import pandas, matplotlib, requests; print('ok')"这一步能提前排除依赖缺失问题。
4. 安装部署与启动方式
这个项目不是一个大型 Web 服务,而是一组可复用的 Python 脚本。为了方便后续维护,建议目录结构如下:
health_gap_analysis/ ├── data/ │ ├── raw/ # 原始下载数据 │ ├── processed/ # 清洗后数据 │ └── cache/ # API 请求缓存 ├── output/ │ ├── charts/ # 图表输出 │ └── tables/ # 汇总表输出 ├── scripts/ │ ├── download_data.py │ └── analyze_health_gap.py ├── requirements.txt └── README.md4.1 数据下载脚本模板
以下脚本是一个通用模板,实际运行前需要把API_URL替换为你要使用的公开数据接口,并把INDICATOR_CODE替换为对应指标代码。不同数据源的返回格式不同,需要根据实际响应结构调整解析字段。
import json import time import requests from pathlib import Path RAW_DIR = Path("data/raw") RAW_DIR.mkdir(parents=True, exist_ok=True) # 以世界银行公开 API 为例,具体接口路径请以官方文档为准 API_URL = "https://api.worldbank.org/v2/country/{country}/indicator/{indicator}" INDICATOR_CODE = "SP.DYN.LE00.IN" # 预期寿命指标示例,请替换为实际指标代码 COUNTRY_LIST = ["CN", "US", "DE", "JP", "BR"] # 示例国家列表,可按需扩展 def fetch_country_data(country: str, indicator: str = INDICATOR_CODE) -> list: params = { "format": "json", "per_page": 200, } all_records = [] for page in range(1, 6): params["page"] = page resp = requests.get(API_URL.format(country=country, indicator=indicator), params=params, timeout=30) resp.raise_for_status() data = resp.json() if len(data) < 2 or not data[1]: break all_records.extend(data[1]) if len(data[1]) < int(params["per_page"]): break time.sleep(0.5) return all_records def main(): for country in COUNTRY_LIST: records = fetch_country_data(country) if not records: continue output_path = RAW_DIR / f"le_{country}.json" with open(output_path, "w", encoding="utf-8") as f: json.dump(records, f, ensure_ascii=False, indent=2) print(f"[OK] {country}: {len(records)} records") if __name__ == "__main__": main()注意:这个示例里的指标代码和接口结构不是固定不变的。World Bank 的指标代码体系已经比较成熟,但健康预期寿命相关指标可能在不同数据源里名称不同,建议以你实际使用的数据源文档为准。
4.2 分析与可视化脚本模板
下面是一个独立的分析脚本,读取包含“国家/年份/预期寿命/健康预期寿命”的 CSV 数据,计算健康缺口,并输出图表和汇总表。
import argparse import pandas as pd import matplotlib.pyplot as plt from pathlib import Path def load_and_clean(filepath: str) -> pd.DataFrame: df = pd.read_csv(filepath) required_cols = {"country", "year", "life_expectancy", "healthy_life_expectancy"} if not required_cols.issubset(df.columns): raise ValueError("CSV 缺少必要列,请检查数据格式") df = df.dropna(subset=["life_expectancy", "healthy_life_expectancy"]) df["health_gap"] = df["life_expectancy"] - df["healthy_life_expectancy"] return df def plot_health_gap(df: pd.DataFrame, output_path: Path) -> None: plt.figure(figsize=(10, 6)) country_list = df["country"].unique()[:5] for country in country_list: subset = df[df["country"] == country].sort_values("year") plt.plot(subset["year"], subset["health_gap"], marker="o", label=country) plt.title("Health Gap over Years") plt.xlabel("Year") plt.ylabel("Health Gap (years)") plt.legend() plt.grid(True, linestyle="--", alpha=0.5) output_path.parent.mkdir(parents=True, exist_ok=True) plt.savefig(output_path, dpi=150, bbox_inches="tight") plt.close() print(f"[OK] Chart saved to {output_path}") def main(): parser = argparse.ArgumentParser(description="Analyze health gap") parser.add_argument("--data", default="data/processed/health_gap_data.csv") parser.add_argument("--output", default="output/charts/health_gap_trend.png") args = parser.parse_args() df = load_and_clean(args.data) summary = df.groupby(["country", "year"])["health_gap"].mean().reset_index() plot_health_gap(summary, Path(args.output)) summary_table = df.groupby("year")["health_gap"].agg(["mean", "median", "max"]) print(summary_table.head(20)) if __name__ == "__main__": main()4.3 一键运行
在项目根目录执行:
python scripts/analyze_health_gap.py \ --data data/processed/health_gap_data.csv \ --output output/charts/health_gap_trend.png执行成功后,终端会打印各年份的缺口统计,并在output/charts/下生成趋势图。如果数据列名不同,需要先修改load_and_clean函数里的必要列名。
5. 功能测试与效果验证
5.1 数据加载测试
测试目的:确认 CSV 能正确加载,列名完整,无全空列。
import pandas as pd df = pd.read_csv("data/processed/health_gap_data.csv") print(df.shape) print(df.dtypes) print(df.isna().sum())关键判断标准:
- shape 大于 0。
- 必要列
country、year、life_expectancy、healthy_life_expectancy都存在。 - 数值列被正确识别为 float 或 int,而不是 object。
如果年份列变成了字符串,需要转换:
df["year"] = pd.to_numeric(df["year"], errors="coerce")5.2 指标计算测试
测试目的:验证“健康缺口”计算逻辑正确。
df["health_gap"] = df["life_expectancy"] - df["healthy_life_expectancy"] print(df[["country", "year", "life_expectancy", "healthy_life_expectancy", "health_gap"]].head())预期结果:health_gap等于前两列数值差。如果出现负值,说明数据源中健康预期寿命高于预期寿命,需要检查指标口径是否一致。
5.3 可视化输出测试
测试目的:确认图表能正常生成,坐标轴正确。
运行分析脚本后,检查输出图片是否存在,并打开图片确认:
- 横轴是年份,纵轴是健康缺口。
- 图例中包含所选国家。
- 图线没有明显断层,折线之间能反映横截面差异。
如果图片中文显示为方块,是因为系统缺少中文字体。此时可以在代码中指定系统中文字体:
import matplotlib.pyplot as plt plt.rcParams["font.sans-serif"] = ["SimHei", "Microsoft YaHei", "PingFang SC"] plt.rcParams["axes.unicode_minus"] = False5.4 验证成功的标准
一次完整的验证应该满足:
- 原始数据下载成功,且缓存文件可再次读取。
- 清洗后数据无全空列,缺失值占比可控。
- 健康缺口计算结果符合业务含义。
- 趋势图能清晰表达逐年变化。
- 汇总统计表可以导出为 CSV 或 Markdown 格式。
6. 接口 API 与批量任务
6.1 公开接口调用模板
很多公开健康数据源都提供 REST API。这里给出一个通用的数据拉取模板,重点展示超时、重试和缓存策略,具体字段需要根据目标接口调整。
import json import time import requests from pathlib import Path CACHE_DIR = Path("data/cache") CACHE_DIR.mkdir(parents=True, exist_ok=True) def fetch_with_retry(url: str, params: dict, max_retries: int = 3) -> dict: for attempt in range(max_retries): try: resp = requests.get(url, params=params, timeout=30) resp.raise_for_status() return resp.json() except requests.Timeout: print(f"[Timeout] attempt={attempt + 1}") except requests.RequestException as e: print(f"[Error] attempt={attempt + 1}: {e}") time.sleep(2 * (attempt + 1)) raise RuntimeError("Request failed after multiple retries") def cached_request(url: str, params: dict, cache_key: str) -> dict: cache_path = CACHE_DIR / f"{cache_key}.json" if cache_path.exists(): with open(cache_path, "r", encoding="utf-8") as f: return json.load(f) data = fetch_with_retry(url, params) with open(cache_path, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) return data这样做的好处是:首次请求成功后,后续分析直接走本地缓存,既减少对公共数据源的压力,也避免重复下载导致时间浪费。
6.2 批量国家与年份
批量任务通常有两种设计方式:
- 按国家循环:一次请求一个国家,得到全部年份。
- 按国家 + 年份循环:请求更细粒度,但请求次数更多,需要控制频率。
推荐按国家循环,并在每次请求之间加入time.sleep(0.5)。如果数据源要求更高,可以升级为指数退避策略。
country_list = ["CN", "US", "DE", "JP", "BR"] for country in country_list: key = f"country_{country}" data = cached_request(API_URL.format(country=country, indicator=INDICATOR_CODE), params={"format": "json", "per_page": 200}, cache_key=key) print(country, len(data) if isinstance(data, list) else "empty") time.sleep(0.5)批量任务一定要写日志。至少记录:
- 当前处理到哪个国家。
- 成功还是失败。
- 失败重试了几次。
- 是否使用了缓存。
6.3 幂等与增量更新
原始数据更新频率通常不高,没必要每次分析都全量拉取。
可以按年份判断增量:
existing_years = set() cache_file = CACHE_DIR / "le_summary.json" if cache_file.exists(): existing_years = set(json.loads(cache_file.read_text(encoding="utf-8")).keys()) # 只拉取缺失的年份或超过更新门槛的年份这样可以在长期项目中保持高效更新。
7. 资源占用与性能观察
健康数据分析项目的计算量通常不会太高,但需要注意以下几点。
7.1 内存占用
如果你只分析几十个国家的年度数据,数据量通常在数万行以内,Pandas 完全能轻松处理。4G 内存的笔记本执行数据清洗和绘图基本不会卡顿。
如果扩展到“所有国家 + 多个指标 + 长年份区间”,原始数据可能达到几十万行,此时要避免一次性把多个超大 CSV 读入内存再拼接。更稳妥的做法是:
- 按国家分块读取。
- 只保留需要的列。
- 用 CSV 分块参数
chunksize分批处理。
chunks = pd.read_csv("large_file.csv", chunksize=20000) partial_results = [] for chunk in chunks: partial = chunk.groupby("year")["health_gap"].mean() partial_results.append(partial) full_result = pd.concat(partial_results).groupby(level=0).mean()7.2 CPU 与 GPU
该分析任务不涉及深度学习,不需要 GPU。绝大多数处理依赖 CPU 单核或双核性能。如果之后引入疾病分类模型、医学影像分析或个体健康预测,才需要考虑 GPU 资源。
7.3 网络请求与带宽
批量拉取公开 API 时,真正的瓶颈在网络延迟和请求频率限制。不要并发发大量请求,否则容易被限流。推荐使用requests.Session复用连接:
session = requests.Session() adapter = requests.adapters.HTTPAdapter(pool_connections=10, pool_maxsize=10) session.mount("https://", adapter)同时保留本地缓存,避免重复下载。
8. 常见问题与排查方法
| 问题现象 | 可能原因 | 排查方式 | 解决方案 |
|---|---|---|---|
| 安装依赖失败 | 网络源不可达 | 查看 pip 报错信息 | 使用国内镜像源或更换 Python 版本 |
| 数据文件下载超时 | 公共 API 响应慢或限流 | 查看 requests 异常类型 | 增加超时时间,加入重试和缓存 |
| CSV 加载后列为 object 类型 | 数据中存在非数值字符 | 使用df.dtypes检查 | 强制类型转换,不能转换的置空 |
| 健康缺口出现负值 | 两个指标口径不一致 | 对比两个数据源的指标定义 | 统一数据源,或重新选择口径一致的字段 |
| 图表中文乱码 | 系统缺少中文字体 | 观察图片中的方框字符 | 在 matplotlib 中指定中文字体 |
| 某些年份缺失严重 | 经济或历史原因导致统计断档 | 统计每列的缺失率 | 对缺失年份做插值或直接排除,并在结论中说明 |
| 批量任务中途失败 | 单次请求异常 | 查看日志中记录的国家位置 | 断点续跑,已成功的数据通过缓存跳过 |
| 内存占用过高 | 一次性读入过多数据 | 使用系统监视器观察 | 改用分块读取,只保留必要列 |
| 结果与官方报告不一致 | 指标口径或年份范围不同 | 核对官方报告的数据来源方法 | 在分析报告中明确说明时间范围和指标定义 |
排查时要先看日志,再复现最小样例,不要在大数据集上反复试错。对于数据质量问题,优先保留原始文件,在加工副本上修改清洗逻辑。
9. 最佳实践与使用建议
9.1 第一次跑通再扩展参数
刚拿到数据时,不要急着做全量分析。先选一个数据质量较好的国家,跑通完整流程:下载、清洗、计算、绘图。整个流程确认没问题后,再扩展到多国家和多年份。
9.2 保留原始数据与加工脚本分离
原始数据放在data/raw/,清洗后的数据放在data/processed/。所有清洗操作必须写进脚本,不要手工在 Excel 里改。这样数据更新后可以一键重新生成。
9.3 批量任务必须加日志和缓存
批量任务至少记录三个信息:当前处理单元、是否成功、失败原因。缓存文件命名要包含国家和指标代码,避免不同指标互相覆盖。
9.4 可视化图表要带完整图注
一张图至少包含:标题、横纵轴含义、数据来源、统计时间范围。如果是国家对比,要标注是“健康预期寿命”还是“预期寿命”,不能笼统写“寿命”。
9.5 涉及个体数据要合规
本项目只使用公开宏观统计数据。在扩展为区域级或个体级健康分析时,必须确认数据使用授权、脱敏处理和隐私保护措施。健康数据属于敏感个人信息,不能仅凭“看起来可以公开”就使用。
9.6 结论要克制
出现“健康缺口扩大”不等于“医疗体系失败”,可能是人口老龄化、慢性病管理改善、筛查率提高导致的诊断数量增加。分析报告里要写清楚相关关系不等于因果关系。
10. 总结与下一步
这个项目最值得尝试的点,是让你用最快的方式把“活得更久”和“活得更健康”拆成两个可量化、可对比的指标,并且用几行 Pandas 代码算出它们之间的缺口。整个流程跑通后,后续扩展方向也很清晰:
- 加入疾病负担指标,比如 DALY(伤残调整生命年),进一步分析缺口主要来自哪些疾病。
- 加入医疗支出数据,判断医疗投入与健康水平之间的关系。
- 接入人口年龄结构数据,清洗年龄结构对健康缺口的干扰。
- 使用统计模型预测未来健康缺口的趋势,而不仅仅是画历史折线。
最容易踩的坑有两个:一是不同数据源的指标口径不一致,二是批量下载时没有做缓存导致重复请求被限流。建议从单一国家、单一指标开始,先画出一张趋势图,再逐步扩大到多国家对比。把数据源、清洗规则、代码版本都固定下来,这个分析项目就能长期复用。
建议收藏备用,等你拿到第一份健康数据后,按本文的脚本模板跑一遍,很快就能验证“寿命越来越长,但健康寿命是否同步增长”这个问题。