Python 医学数据处理结构化病历的标准化清洗 Pipeline一、医生写的病历机器读起来像天书电子病历EMR的数据质量是医疗 AI 落地最大的拦路虎。一份看似完整的病历经过程序解析后往往会暴露各种问题主诉字段填了 500 字的流水账、诊断名称同时存在 ICD-10 和 ICD-11 两种编码、高血压有时写作高血压病有时缩写为HBP。如果不做标准化清洗下游模型训练出来就是垃圾进垃圾出。更糟糕的是病历数据中的错误类型五花八门数值型字段填了文字体温正常而不是36.5、日期格式不统一2024/3/15和2024-03-15混用、必填字段留空但填了一个无字。传统 ETL 工具面对这种半结构化数据基本失灵需要一个专门针对医学文本的清洗 Pipeline。二、清洗 Pipeline 架构化零为整逐层提纯标准化的病历清洗不是一步到位而是分阶段提纯的过程每个阶段产出带标注的中间数据这样出问题时可以快速定位是哪个环节的锅。关键设计前一个阶段的输出不会覆盖原始数据而是产生新列如diagnosis_raw→diagnosis_cleaned→diagnosis_normalized方便追溯。三、Python 实现可配置的清洗 Pipelineimport re import pandas as pd from datetime import datetime from typing import Dict, List, Optional, Tuple from dataclasses import dataclass, field import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) dataclass class CleaningConfig: 清洗配置 date_formats: List[str] field(default_factorylambda: [ %Y-%m-%d, %Y/%m/%d, %Y年%m月%d日 ]) # 症状同义词映射 symptom_synonyms: Dict[str, str] field(default_factorylambda: { HBP: 高血压, 高血压病: 高血压, 血压偏高: 高血压, DM: 糖尿病, 糖尿病Ⅱ型: 2型糖尿病, }) # 数值字段的正则 temperature_pattern: re.Pattern re.compile( r(\d{2}\.?\d?)\s*[°℃度]? ) class EMRCleaner: 电子病历清洗器 def __init__(self, config: Optional[CleaningConfig] None): self.config config or CleaningConfig() self.cleaning_log: List[Dict] [] def load_raw_data(self, filepath: str) - pd.DataFrame: 加载原始病历数据 try: df pd.read_csv(filepath, encodingutf-8) except UnicodeDecodeError: df pd.read_csv(filepath, encodinggbk) logger.info(f加载 {len(df)} 条病历记录) return df def standardize_dates(self, df: pd.DataFrame, date_cols: List[str]) - pd.DataFrame: 阶段1日期格式标准化 for col in date_cols: if col not in df.columns: continue df[f{col}_std] df[col].apply(self._parse_date) return df def _parse_date(self, value) - Optional[str]: 尝试多种格式解析日期 if pd.isna(value) or str(value).strip() in (, 无, 不详): return None value_str str(value).strip() for fmt in self.config.date_formats: try: dt datetime.strptime(value_str, fmt) return dt.strftime(%Y-%m-%d) except ValueError: continue # 尝试只提取数字 digits re.findall(r\d, value_str) if len(digits) 3: return f{digits[0]}-{digits[1].zfill(2)}-{digits[2].zfill(2)} logger.warning(f无法解析日期: {value_str}) return None def clean_temperature(self, df: pd.DataFrame, col: str temperature) - pd.DataFrame: 清洗体温字段文字描述转为数值 def extract_temp(val): if pd.isna(val): return None val_str str(val).strip() # 处理正常无发热等描述 if val_str in (正常, 无发热, 体温正常): return 36.5 match self.config.temperature_pattern.search(val_str) if match: temp float(match.group(1)) # 异常值检测 if 34 temp 43: return temp else: logger.warning(f体温异常值: {temp}原文: {val_str}) return None return None df[f{col}_numeric] df[col].apply(extract_temp) return df def normalize_symptoms(self, df: pd.DataFrame, col: str chief_complaint) - pd.DataFrame: 阶段3症状术语归一化 synonyms self.config.symptom_synonyms def normalize(text: str) - str: if pd.isna(text): return text result str(text) for old, new in synonyms.items(): result result.replace(old, new) return result df[f{col}_normalized] df[col].apply(normalize) return df def quality_score(self, df: pd.DataFrame) - pd.DataFrame: 阶段4数据质量评分0-100 scores pd.Series(100.0, indexdf.index) # 必填字段缺失扣分每缺失1个字段扣15分 required_fields [patient_id, visit_date_std, diagnosis] for field in required_fields: if field in df.columns: scores - df[field].isna().astype(float) * 15 # 异常体温扣分 if temperature_numeric in df.columns: temp df[temperature_numeric] abnormal (temp 35) | (temp 42) scores - abnormal.astype(float) * 20 # 日期未来时间扣分 if visit_date_std in df.columns: future pd.to_datetime( df[visit_date_std], errorscoerce ) datetime.now() scores - future.astype(float) * 25 scores scores.clip(lower0, upper100) df[quality_score] scores return df def pipeline(self, filepath: str) - Tuple[pd.DataFrame, Dict]: 执行完整清洗 Pipeline df self.load_raw_data(filepath) date_cols [c for c in df.columns if any(kw in c.lower() for kw in [date, time, 日期, 时间])] df self.standardize_dates(df, date_cols) df self.clean_temperature(df) df self.normalize_symptoms(df) df self.quality_score(df) # 生成清洗报告 report { total_records: len(df), high_quality: int((df[quality_score] 80).sum()), need_review: int(((df[quality_score] 60) (df[quality_score] 80)).sum()), excluded: int((df[quality_score] 60).sum()), date_parse_failures: df[[c for c in df.columns if c.endswith(_std)]] .isna().sum().to_dict(), } logger.info(f清洗完成: {report}) return df, report四、边界分析与 Trade-offs清洗粒度自动 vs 人工全自动清洗的准确率大约在 85%-90%剩下的 10% 需要人工核实。经验法则是质量分 80 的记录直接入库60-79 分的提交人工审核队列 60 分的标记后暂存等数据源修复后重新跑。不要在代码里硬编码太多业务规则——规则变化比代码变化快得多。同义词映射的维护症状和诊断的同义词是最容易过时的。建把映射表放在配置中心如 Consul 或本地 YAML 文件支持热更新不要让运维为了改一个高血压的同义词而重新部署服务。映射表应该由医学知识团队维护而不是开发团队。大文件的内存策略pandas 默认把整个 DataFrame 加载到内存处理 10GB 的病历文件会直接 OOM。解决方案是分块读取chunksize参数或使用 Dask/Polars 这样的增量计算框架。分块时需要注意某些清洗操作如日期范围校验需要全局上下文这种情况可以先采样估算全局参数再分块处理。五、总结病历数据清洗的核心思路是先标准化格式再归一化语义最后量化质量。分阶段处理的优势在于问题可追溯——数据质量报告会精确告诉你哪一列、哪种错误类型最严重。DataFrame 的apply操作虽然方便但在百万级数据上会变慢生产环境建议迁移到 Polars 或 PySpark。最重要的一点清洗不是一次性的数据源的格式会随时间变化医院升级 HIS 系统等所以 Pipeline 要设计为可重跑的支持增量清洗和全量重洗两种模式。