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 实现：可配置的清洗 Pipeline

import 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(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class CleaningConfig: """清洗配置""" date_formats: List[str] = field(default_factory=lambda: [ "%Y-%m-%d", "%Y/%m/%d", "%Y年%m月%d日" ]) # 症状同义词映射 symptom_synonyms: Dict[str, str] = field(default_factory=lambda: { "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, encoding='utf-8') except UnicodeDecodeError: df = pd.read_csv(filepath, encoding='gbk') 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, index=df.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'], errors='coerce' ) > datetime.now() scores -= future.astype(float) * 25 scores = scores.clip(lower=0, upper=100) 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 要设计为可重跑的，支持增量清洗和全量重洗两种模式。