证据链接式特征工程：让心衰预测建模可追溯可复现

这次我们来看一个不做模型训练、却直接影响模型上限的项目Tracing the Heart。直译过来是“追踪心脏”但它不是影像分割也不是心电识别而是一条围绕心衰Heart Failure数据集的证据链接式特征工程 Pipeline。它的目标很集中把原始临床记录整理成可建模、可追溯、可复现的特征矩阵并且让每一个特征都能对应到临床证据或医学指南上的依据。最值得关注的不是它又封装了多少个模型而是它的设计逻辑——证据链接Evidence-Linked。传统特征工程做完一版特征往往只剩下一个data.csv和一个“效果还行”的结果这个 Pipeline 强调在特征衍生的同时保存证据字典包括特征名、衍生公式、临床依据、处理层级让后续的模型迭代、论文复现、临床解释都有据可查。对做医疗数据挖掘和预测建模的人来说这比单纯堆特征重要得多。从门槛看它不依赖 GPU不需要大显存普通 CPU 机器和 Python 数据科学环境就能跑。它关注的核心不是推理速度而是数据质量、特征可解释性和批处理稳定性。如果你正在做医疗数据挖掘、心衰预测建模、Kaggle 类比赛或者想把特征工程按工程化标准固化下来这篇文章可以直接收藏。下面会完整拆解这条 Pipeline 的设计思路、环境准备、模块划分、代码实现、批处理与接口化部署以及容易踩的坑。1. 项目核心能力速览先从整体上把这条 Pipeline 的定位讲清楚。它不是某个可以直接pip install的固定库而是一个可以按项目仓库落地的特征工程框架通常由数据质检、证据字典、特征衍生、特征筛选、产物导出五个部分组成。能力项说明项目类型医疗数据特征工程 Pipeline / 数据预处理框架核心目标将心衰临床记录转化为可建模特征矩阵并提供证据追溯输入数据结构化临床表CSV/Parquet常见字段如年龄、射血分数、血清肌酐、血清钠、随访天数等输出产物特征矩阵、证据字典、特征报告、数据集切分结果运行环境CPU 即可Python 3.8无 GPU 依赖启动方式命令行或 Python 调用可封装为 API 服务是否支持批量支持批量数据集和多折特征生成接口能力可暴露为 REST API主要依赖pandas、numpy、scikit-learn、pyarrow可选 FastAPI/uvicorn适合人群医疗数据研究者、特征工程开发、预测建模工程师这条 Pipeline 的核心能力可以归纳成四句话特征有出处、衍生有公式、切分可复现、结果可审计。所谓“特征有出处”指的是每个特征都带一个evidence_id指向指南条目、公开研究或专家规则“衍生有公式”要求所有特征处理逻辑写成显式函数而不是散落在 notebook 里“切分可复现”指每次生成训练集和测试集都使用固定随机种子“结果可审计”指特征报告会记录每个特征的缺失率、分布、处理方式方便回溯。2. 适用场景与使用边界这条 Pipeline 适合谁首先是在公开数据集或院内科研数据上做心衰预测建模的研究人员特征工程阶段需要一个统一框架来管理几十上百个候选特征。其次是算法团队的工程同学想摆脱手工维护特征脚本、希望每次建模都能复现同一套特征版本。再就是做论文复现的读者证据链接设计可以让审稿人看到每个特征为什么被构造出来。它不适合什么场景不适合直接做临床诊断。特征工程输出的特征矩阵只能用于算法研究和模型验证不能替代医生判断。它也不适合完全没有结构化数据的场景比如只有心电图原始波形或影像 Dicom 文件那需要的是信号处理和影像特征提取不是这张表级特征 Pipeline。此外如果团队只想要“一个黑盒模型赶紧出结果”不愿意维护证据字典和特征报告那这条 Pipeline 反而会显得重。医疗数据的合规边界必须单独强调。心衰记录里有大量受保护的健康信息任何处理流程都要遵守所在地区的数据保护法规例如中国网络安全法、数据安全法、个人信息保护法以及国际上的 HIPAA、GDPR 等要求。公开数据集如果有授权协议也要确认是否允许二次加工和发布。凡是涉及患者隐私的数据建议在进入 Pipeline 之前先完成去标识化保留字段名和脱敏 ID不要在特征报告里输出任何可定位到个人的信息。3. 数据集与前置环境准备3.1 数据格式假设这条 Pipeline 的输入是结构化临床表常见来源包括 UCI Heart Failure Clinical Records 这类公开数据集以及医院科研数据库导出的 CSV 或 Parquet。这类数据通常包含以下典型字段我这里以常见公开字段为例人口学字段年龄、性别生命体征与心功能射血分数ejection fraction、血压实验室指标血清肌酐、血清钠、肌酸激酶、血小板合并症标志贫血、糖尿病、高血压、吸烟史随访信息随访天数、死亡结局在写 Pipeline 之前先确认字段类型是连续值还是分类型并检查缺失值比例。医疗数据最忌讳的是对缺失含义不做区分有些缺失是“未检测”有些是“记录丢失”二者处理方式完全不同。建议在接入阶段就为每个字段维护一个元数据描述文件包含字段名、类型、单位、缺失含义、取值范围。3.2 环境清单不需要 GPU也不需要配置 CUDA一台普通的 4 核 CPU、16G 内存的机器就足够。Python 环境建议用 conda 或 venv 隔离避免污染系统环境。# 创建独立虚拟环境 python3 -m venv .venv source .venv/bin/activate # 安装核心依赖 pip install pandas numpy scikit-learn pyarrow pydantic # 如果要把 Pipeline 封装成接口服务 pip install fastapi uvicorn依赖锁定建议用 requirements.txt 或 pyproject.toml特征工程的结果对依赖版本很敏感pandas 和 scikit-learn 的版本升级可能导致行为变化不锁定版本复现就是空话。3.3 目录结构推荐按分层方式组织目录把原始数据、中间产物、最终特征分开heart_pipeline/ ├── config/ │ └── pipeline.yaml ├── data/ │ ├── raw/ # 原始数据只读 │ ├── interim/ # 中间产物 │ └── features/ # 最终特征矩阵 ├── src/ │ ├── ingest.py │ ├── quality.py │ ├── evidence.py │ ├── derivation.py │ └── selection.py ├── output/ │ ├── reports/ # 特征报告 │ └── models/ # 特征筛选结果 └── tests/ └── test_features.py这个结构的关键是原始终只读中间产物可重建最终特征必须能从代码重新生成。任何手工修改特征文件的习惯都要改掉一旦手工改数据整个可复现性就断了。4. Pipeline 总体架构与证据链接设计4.1 分层架构这条 Pipeline 可以拆成六个阶段每个阶段输入输出都是确定的表结构阶段之间不共享可变状态阶段输入输出核心职责1. 数据接入原始 CSV/Parquet标准化 DataFrame字段映射、类型转换、去标识化2. 数据质检标准化 DataFrame质检报告缺失率、取值异常、时间序检查3. 证据目录外部指南/文献/专家规则Evidence Catalog为每个特征定义证据依据4. 特征衍生标准化 DataFrame Evidence Catalog候选特征矩阵执行显式的衍生函数5. 特征筛选候选特征矩阵筛选后特征集相关性、稳定性、重要性筛选6. 产物导出筛选后特征集特征文件 报告落盘并生成审计记录每个阶段都可以单独运行也可以由总入口按顺序调度。这样设计的好处是当新增一个特征时不需要重跑前面所有阶段只需要在证据目录里追加一条再执行特征衍生和后续阶段。4.2 Evidence Catalog 的设计Evidence Catalog 是本项目的核心概念。它本质上是一个不可变的注册表每条记录包含特征名、特征类型、衍生公式、临床证据来源、证据类型、创建时间。from dataclasses import dataclass, asdict from datetime import datetime dataclass(frozenTrue) class EvidenceEntry: feature_name: str feature_type: str # raw / derived / flag formula: str # 特征计算公式或逻辑描述 evidence_source: str # 指南条目、文献 DOI 或专家规则编号 evidence_type: str # guideline / publication / expert version: str # 证据版本 created_at: str datetime.now().isoformat() def to_dict(self): return asdict(self)注意这里使用了frozenTrue目的是让证据记录不可变。特征工程最怕的是同一个特征名在不同版本里有不同含义用不可变对象可以从数据结构层面减少这种风险。如果确实要修改证据就生成一条新版本而不是覆盖旧记录。5. 各功能模块实现与验证5.1 数据接入与质检数据接入阶段要做三件事字段映射、类型转换、去标识化。字段映射解决的是同一字段在不同来源里叫法不同的问题例如“射血分数”可能叫ejection_fraction也可能叫ef统一映射表是必须的。import pandas as pd COLUMN_MAP { ejection_fraction: ef, serum_creatinine: creatinine, serum_sodium: sodium, age: age, DEATH_EVENT: death_event, } def ingest(path: str) - pd.DataFrame: df pd.read_csv(path) df df.rename(columnsCOLUMN_MAP) # 只保留映射后字段防止意外字段进入下游 keep_cols list(COLUMN_MAP.values()) [patient_id] df df[[c for c in keep_cols if c in df.columns]] # 去标识化patient_id 作为唯一标识不保留姓名/身份证等字段 return df数据质检报告要覆盖三个维度缺失率、取值范围、逻辑一致性。逻辑一致性是医疗数据最容易出问题的地方例如“随访天数为 0 但结局事件已发生”这类组合虽然不一定是错误但需要标记出来人工判断。def quality_report(df: pd.DataFrame) - pd.DataFrame: report pd.DataFrame({ col: df.columns, missing_rate: df.isna().mean().round(4), dtype: df.dtypes.astype(str), min: df.min(numeric_onlyTrue), max: df.max(numeric_onlyTrue), }) report[status] report[missing_rate].apply( lambda x: ok if x 0.1 else check ) return report这个阶段判断成功的标准是字段全部完成映射、缺失率超过 10% 的字段被标记、没有任何未授权字段进入下游。如果发现原始表里出现姓名、身份证号、精确出生日期等敏感字段直接丢弃或脱敏不要带进特征矩阵。5.2 证据驱动特征衍生特征衍生是整条 Pipeline 的主体。这里的写法要求每个特征一个函数函数名就是特征名函数内部先引用 Evidence Catalog 中的条目再执行计算。下面以三个常见临床特征为例说明怎么写更稳妥。import numpy as np # 特征1估算肾小球滤过率相关指标基于肌酐和年龄的简单衍生 def renal_function_index(row) - float: 基于血清肌酐与年龄的粗算衍生指标。 证据来源慢性肾脏病相关评估的常用组合指标。 注意仅用于算法研究不能替代临床 eGFR 计算。 creatinine row.get(creatinine, np.nan) age row.get(age, np.nan) if pd.isna(creatinine) or pd.isna(age) or creatinine 0: return np.nan return age / creatinine # 特征2低钠风险标志 def low_sodium_flag(row) - int: 血清钠低于常见参考下限时置 1。 依据低钠血症与心衰预后的关联在多篇研究中被讨论。 sodium row.get(sodium, np.nan) if pd.isna(sodium): return -1 # 缺失用 -1 标记 return 1 if sodium 135 else 0 # 特征3随访时间与结局的交互观察字段 def followup_bucket(row) - int: 将随访天数分桶便于后续做时间相关建模。 t row.get(time, np.nan) if pd.isna(t): return -1 if t 30: return 0 elif t 90: return 1 elif t 180: return 2 return 3批量衍生的主函数负责遍历特征函数列表并同步生成证据报告DERIVED_FEATURES [ renal_function_index, low_sodium_flag, followup_bucket, ] def derive(df: pd.DataFrame) - tuple[pd.DataFrame, pd.DataFrame]: out df.copy() for fn in DERIVED_FEATURES: out[fn.__name__] out.apply(fn, axis1) evidence [ { feature_name: fn.__name__, feature_type: derived, formula: fn.__doc__, evidence_source: expert_rule, evidence_type: expert, } for fn in DERIVED_FEATURES ] return out, pd.DataFrame(evidence)这里有个容易被忽视的坑apply(axis1)在数据量大的时候会很慢几万行以内问题不大几十万行就可能要等很久。如果数据量大建议把特征函数改成向量化写法传pd.Series而不是逐行row。在验证阶段先用小样本跑通逻辑再用完整数据跑性能。5.3 特征筛选与数据集切分特征衍生之后候选特征可能很多。筛选的主要目的是去掉高相关、低波动的冗余特征同时保留证据重要但统计上不显著的特征。因为心衰数据样本量通常不大特征太多容易过拟合。from sklearn.model_selection import train_test_split from sklearn.ensemble import RandomForestClassifier import pandas as pd def select_features( feature_matrix: pd.DataFrame, target_col: str death_event, random_state: int 42, ) - list[str]: cols [c for c in feature_matrix.columns if c ! target_col] X feature_matrix[cols] y feature_matrix[target_col] # 去掉方差极低的特征 low_var [c for c in cols if X[c].nunique() 1] keep [c for c in cols if c not in low_var] # 用简单模型评估重要性这里只用于筛选不用于最终模型 X_clean X[keep].fillna(-1) clf RandomForestClassifier( n_estimators100, random_staterandom_state, class_weightbalanced, ) clf.fit(X_clean, y) imp pd.Series(clf.feature_importances_, indexkeep) # 保留重要性超过阈值的特征阈值按实际数据调整 selected imp[imp 0.02].index.tolist() return selected def split_features( feature_matrix: pd.DataFrame, target_col: str death_event, random_state: int 42, ): X feature_matrix.drop(columns[target_col]) y feature_matrix[target_col] return train_test_split( X, y, test_size0.2, stratifyy, random_staterandom_state )筛选完成之后切分必须固定random_state并且对分类目标使用stratify分层抽样。心衰数据集的结局事件一般不是均匀分布的直接随机切分很容易造成测试集里某类样本过少导致评估结果波动很大。固定种子之后每次跑出来的训练集和测试集完全一致模型对比才有意义。5.4 功能验证方法每个模块的验证标准要明确。数据质检的验证标准是“原始数据进去质检报告出来字段缺失和异常全部被标记”。特征衍生的验证标准是“每个新特征在证据目录里都能查到记录且空值比例不超过预设阈值”。特征筛选的验证标准是“输出特征集合的英文名、中文含义、证据来源都能对应”。最后跑一次全流程确认在相同输入下两次执行结果完全一致这一步可以通过对最终特征文件计算哈希值来验证。6. 批处理、接口 API 与工程化部署6.1 CLI 批处理把 Pipeline 封装成命令行工具批量跑多个数据集文件。推荐用argparse或者click这类标准库级方案避免引入不必要的依赖。# 启动全流程特征生成 python src/run_pipeline.py \ --config config/pipeline.yaml \ --input data/raw/heart_failure.csv \ --output data/features/heart_features.parquet \ --report-dir output/reports批量场景下最稳妥的做法是保持统一目录结构输入和输出目录分离{ input_dir: data/raw, output_dir: data/features, report_dir: output/reports, feature_engine: { random_state: 42, test_size: 0.2, importance_threshold: 0.02 }, evidence_version: 2025.01 }批处理要加失败重试和日志。医疗数据集经常遇到某个文件字段缺失、某个值类型异常导致整个批次中断的情况。建议把每个文件的处理包在try/except里失败时写error.log并继续处理下一个文件而不是整体崩掉。6.2 FastAPI 接口服务化如果要把 Pipeline 开放给团队其他成员使用可以封装成 REST API。下面是一个最小可用的 FastAPI 示例实际字段以项目为准from fastapi import FastAPI, UploadFile, File from fastapi.responses import JSONResponse from io import BytesIO import pandas as pd from derivation import derive app FastAPI(titleHeart Failure Feature Pipeline API) app.post(/api/features) async def create_features(file: UploadFile File(...)): content await file.read() df pd.read_csv(BytesIO(content)) # 数据接入与质检 df df.rename(columnsCOLUMN_MAP) keep_cols [c for c in COLUMN_MAP.values() if c in df.columns] df df[keep_cols] # 特征衍生 out, evidence derive(df) return JSONResponse({ status: ok, rows: len(out), features: list(out.columns), evidence: evidence.to_dict(orientrecords), })启动服务uvicorn api:app --host 127.0.0.1 --port 8000注意三点第一接口默认只绑定127.0.0.1不要直接暴露到公网特征矩阵是脱敏后的数据但字段组合依然可能构成隐私风险第二接口层只接收结构化表格不要接收包含患者自由文本的文件否则要做额外的文本脱敏第三上传文件大小要加限制特征处理是 CPU 密集任务过大文件会拖垮服务。6.3 请求与返回示例用 Python 的requests调用接口import requests url http://127.0.0.1:8000/api/features files {file: open(data/raw/heart_failure_sample.csv, rb)} resp requests.post(url, filesfiles, timeout120) print(resp.json()[status]) print(resp.json()[features])接口跑通之后后续就能接到自动化任务里例如每天早上定时处理前一日新增的脱敏临床表输出特征文件到指定目录再由下游建模任务消费。7. 资源占用与性能观察这条 Pipeline 没有显存占用问题重点观察的是 CPU、内存和运行时间。数据接入和apply(axis1)特征衍生是主要耗时点。以几万行的公开心衰数据集为例全流程通常几十秒内能跑完但如果数据量到百万行级别就必须做三件事。第一用 Parquet 替代 CSV 作为中间格式读写速度快且自带列式压缩。第二特征衍生从逐行apply改成向量化计算pandas的向量化操作比逐行循环快一个数量级以上。第三使用polars或pandas 2.x的copy-on-write模式处理大表减少内存复制。memory_profiler可以用来定位内存峰值pip install memory_profiler mprof run python src/run_pipeline.py --config config/pipeline.yaml --input data/raw/heart_failure.csv --output data/features/out.parquet mprof plot性能优化有一条铁律先跑通再优化。第一次先用 1000 行的子集跑通全流程确认逻辑正确后再逐步放大到全量数据同时记录每个阶段的耗时和内存。不要一开始就把几十万行数据丢进一个没测试过的特征函数里出了问题排查成本极高。8. 常见问题与排查方法问题现象可能原因排查方式解决方案运行时报 KeyError找不到某个字段字段映射表与原始数据列名不一致打印原始表columns对比映射表补齐COLUMN_MAP或运行前做字段存在性检查特征衍生出现大量空值原始字段缺失率高或衍生函数对缺失值未处理查看质检报告的missing_rate在衍生函数中对空值显式标记不静默传递两次运行结果不一致没有固定随机种子或有随机采样逻辑检查代码中的random、np.random、train_test_split统一设置random_state并核对版本切分后训练集和测试集类别分布差异大没有做分层抽样打印y_train.value_counts()与y_test.value_counts()使用stratifyy若样本量过小改用重复分层切分数据量增大后运行很慢逐行apply或重复读取大文件用mprof和代码计时定位耗时环节改成向量化计算中间结果用 Parquet 落盘API 返回 413 或 504上传文件过大或处理超时检查服务端日志和上传限制限制上传大小、拆分文件、增加超时时间特征报告里出现敏感字段原始数据未经去标识化就进入下游检查接入阶段keep_cols白名单强制字段白名单任何未授权字段直接丢弃证据目录和特征名对不上新增特征时没有同步更新 Evidence Catalog跑一次证据一致性检查脚本在 CI 或启动入口做证据覆盖校验最典型的问题出在字段映射。医疗数据集的列名经常变同一个字段在导出时可能是EF也可能是EjectionFraction手工拼出来的COLUMN_MAP一旦漏掉一个下游就是一片 KeyError。建议在接入阶段加一个字段存在性检查函数启动时强制校验必需字段缺了就立即报错而不是运行到一半才暴露。另一个常见问题是缺失值处理的隐蔽错误。很多人习惯用fillna(0)一次性填掉所有空值但在医疗数据里这非常危险。fillna(0)会把“缺失的肌酐”和“肌酐为 0”混为一谈模型可能会学到错误规律。更稳妥的做法是保留缺失标记例如用 -1 表示缺失或者在衍生函数里单独处理。9. 最佳实践与合规建议9.1 工程实践第一第一次跑全流程前先用小样本数据集做“冒烟测试”。准备一个 500 行左右的脱敏样本覆盖正常值、边界值和缺失值确保每个函数都不会报错、不会产生意外类型。小样本跑通之后再上完整数据。第二保留一套最小可运行配置。把config/pipeline.yaml做成精简版只包含默认字段映射、默认随机种子、默认输出路径任何新环境部署都先用这套配置验证成功后再调整参数。这样可以快速区分“环境问题”和“业务问题”。第三模型文件、输入素材、输出结果分目录管理。原始数据放到data/raw只读不写特征矩阵放到data/features报告和日志放到output。目录之间不交叉避免误覆盖。第四批处理必须加日志和失败重试。日志要记录每个文件的处理状态、耗时、失败原因。失败重试建议采用指数退避不要无脑循环重试否则磁盘或网络问题会放大负载。第五接口服务必须限制访问范围。默认绑定127.0.0.1如果团队内部使用放在内网并通过网关鉴权不要直接在服务器上暴露 8000 端口到公网。9.2 数据合规医疗特征工程的合规红线这里再强调一遍。数据去标识化优先。在数据进入 Pipeline 之前移除姓名、身份证号、精确出生日期、详细住址等可直接识别字段。patient_id使用脱敏后的随机 ID不能是真实病历号。公开数据集遵循授权协议。很多公开数据集虽然可以下载但授权协议会限制使用范围尤其是“禁止重新发布”“仅用于非商业研究”等条款发布特征样例或论文材料前要逐条核对。涉及人脸、声音、版权素材的规则同样适用。虽然本 Pipeline 处理的是表格字段但只要是医疗数据就要坚持最小必要原则只用完成研究目标所必需的字段不采集无关字段。特征报告不输出小样本统计。当某个分组的样本量过小时输出统计值可能间接暴露个体信息报告层面可以做单元格抑制处理对过小分组的数值打码。商用部署前必须做合规审查。算法研究阶段和产品商用阶段的数据合规标准不同如果特征矩阵最终要进入医疗产品需要专门的数据保护和伦理审查流程。10. 总结与下一步方向这条 Pipeline 最值得尝试的点是把特征工程从“写一个处理脚本”升级成了“带证据追溯的工程框架”。它不要求多高端的硬件也不依赖任何商业工具一个 Python 环境加 pandas 就能搭起来。它最核心的价值是复现性和可解释性每个特征都能从 Evidence Catalog 追溯到来源每次结果都能用固定种子复现每个版本都有审计记录。建议最先验证的功能是数据质检和证据目录。先准备一份真实的心衰结构化表跑通字段映射和质检报告确认缺失率和异常值能正确暴露这比直接优化模型更重要。最容易踩的坑是字段映射不完整和缺失值被统一fillna(0)这两类问题在医疗数据里会直接污染下游特征而且极难排查。后续可以继续扩展的方向有三条。一是把证据目录从静态注册表升级为版本化数据库每个证据条目支持生效时间、失效时间方便长期维护。二是接入更多临床特征源例如用药记录、多次随访序列把特征工程从“静态截面”扩展成“时间序列特征”。三是增加自动化测试覆盖为每个衍生特征写单元测试保证新增特征时旧特征不复现这条 Pipeline 就能从“能跑通”变成“敢进生产”。如果你正在做心衰预测或者医疗结构化数据的特征工程建议直接把这条 Pipeline 的思路落地到你现有的实验流程里。先把目录结构、Evidence Catalog、固定随机种子这三件事做起来再逐步补全后续模块。这套框架不挑数据集心衰能用其他疾病队列一样能套用越早固化下来后面的建模迭代越省心。