【医药AI实战系列⑥】真实世界数据RWD怎么喂给机器学习模型

从一次被FDA打回来的申请说起2022年，某药企尝试用真实世界数据（RWD）支持一个适应症扩展申请。数据来自美国三家大型医疗系统，覆盖超过180万名患者，时间跨度8年。数据量足够大，团队信心十足。FDA的回复用了整整14页，核心意见浓缩成一句话：“提交方未能充分证明研究人群与目标人群之间的可比性，且未对已知和未知混杂因素进行充分控制。”项目重做，又花了18个月。这个故事不是个例。FDA在2021年发布的《真实世界证据项目框架》里明确指出：RWD的质量问题，是目前RWE申请最主要的失败原因，没有之一。今天我们就把RWD的三类核心数据质量问题拆开，逐一给出工程解法。RWD和RCT：为什么RWD天生更脏先理解为什么RWD的数据质量问题是结构性的，不是偶然的。随机对照试验（RCT）是在严格控制条件下生产数据：入排标准精确、访视时间固定、数据采集标准化、有专职数据管理员。数据的目的就是为了分析。RWD完全相反。电子病历、理赔数据、患者登记数据，这些数据的生产目的是临床诊疗和医院运营，不是科学研究。数据里的每一个缺失、每一个不一致，都有它的临床或行政原因。RCT数据生产逻辑： 设计分析目标 → 设计数据采集 → 采集数据 → 分析 RWD数据生产逻辑： 临床诊疗需要 → 顺手记录 → 数据堆积 → 事后挖掘 ↑ 这一步的随意性，决定了后面所有的麻烦FDA RWE指南把RWD质量问题归为五类：相关性、可靠性、完整性、准确性、可及性。我们今天重点拆其中工程难度最高的三类：ICD编码漂移、混杂偏倚、缺失值。第一类：ICD编码漂移——同一个病，八年前和八年后是两套语言问题本质ICD（国际疾病分类）编码是电子病历里疾病诊断的标准语言。美国在2015年10月从ICD-9切换到ICD-10，编码数量从约14,000个暴增到约70,000个。中国在2019年起推行ICD-11。这意味着：如果你的RWD时间跨度超过这些切换节点，同一个疾病在不同时间段的编码完全不同，而且不是简单的一对一映射。以2型糖尿病为例：ICD-9: 250.00 2型糖尿病，无并发症 250.02 2型糖尿病，无并发症，未控制 250.40 2型糖尿病伴肾脏并发症 （共约15个相关编码） ICD-10: E11.9 2型糖尿病，无并发症 E11.65 2型糖尿病伴高血糖 E11.21 2型糖尿病伴糖尿病肾病 E11.311 2型糖尿病伴未指明的糖尿病视网膜病变，伴黄斑水肿 （共约70个相关编码，粒度细化了约5倍）如果直接把ICD-9和ICD-10的编码拼在一起喂给模型，会发生什么？模型会学到：2015年10月之前，糖尿病肾病很少见；2015年10月之后，糖尿病肾病突然大量出现。这个"趋势"是编码系统切换造成的伪影，不是真实的疾病流行病学变化。工程解法：编码映射 + 时间感知特征importpandasaspdimportnumpyasnpfromtypingimportDict,List,Set,Optionalfromdatetimeimportdatetime# ICD-9 到 ICD-10 的映射表（使用CMS官方GEM文件）# 下载地址：https://www.cms.gov/Medicare/Coding/ICD10/2018-ICD-10-CM-and-GEMs# 文件名：2018_I9gem.txt（ICD-9到ICD-10的通用等价映射）defload_gem_mapping(gem_file_path:str)-Dict[str,List[str]]:""" 加载CMS官方GEM（General Equivalence Mappings）文件 返回 icd9_code - [icd10_codes] 的映射字典 GEM文件格式（空格分隔）： ICD9CODE ICD10CODE FLAG 25000 E119 10111 """mapping={}withopen(gem_file_path,'r')asf:forlineinf:parts=line.strip().split()iflen(parts)2:continueicd9=parts[0].strip()icd10=parts[1].strip()# 格式化ICD-9编码（加小数点）iflen(icd9)3and'.'notinicd9:icd9=icd9[:3]+'.'+icd9[3:]ificd9notinmapping:mapping[icd9]=[]mapping[icd9].append(icd10)returnmappingdefnormalize_icd_codes(df:pd.DataFrame,code_col:str,date_col:str,gem_mapping:Dict[str,List[str]],icd10_cutoff_date:str="2015-10-01",strategy:str="map_to_icd10")-pd.DataFrame:""" 统一ICD编码版本，解决跨版本漂移问题 Args: df: 包含ICD编码和日期的DataFrame code_col: ICD编码列名 date_col: 诊断日期列名 gem_mapping: ICD-9到ICD-10的GEM映射 icd10_cutoff_date: ICD-10切换日期 strategy: "map_to_icd10" - 将所有ICD-9编码映射到ICD-10（推荐） "use_icd9_parent" - 统一使用ICD-9的父级编码（粒度粗但一致） "flag_version" - 保留原编码但添加版本标记（用于模型特征） Returns: 添加了标准化编码列的DataFrame """cutoff=pd.Timestamp(icd10_cutoff_date)df=df.copy()df[date_col]=pd.to_datetime(df[date_col])# 判断每条记录使用的ICD版本df['icd_version']=np.where(df[date_col]cutoff,'ICD9','ICD10')ifstrategy=="map_to_icd10":defmap_code(row):ifrow['icd_version']=='ICD10':return[row[code_col]]else:# ICD-9映射到ICD-10icd9_code=row[code_col]mapped=gem_mapping.get(icd9_code,[])ifnotmapped:# 尝试父级编码映射parent=icd9_code[:3]if'.'notinicd9_codeelseicd9_code.split('.')[0]mapped=gem_mapping.get(parent,[f"UNMAPPED_{icd9_code}"])returnmapped df['icd10_codes']=df.apply(map_code,axis=1)df['icd10_primary']=df['icd10_codes'].apply(lambdax:x[0]ifxelseNone)df['mapping_confidence']=df.apply(lambdarow:'exact'ifrow['icd_version']=='ICD10'else('mapped'ifnotrow['icd10_primary'].startswith('UNMAPPED')else'failed'),axis=1)elifstrategy=="flag_version":# 保留原编码，添加版本和时间特征df['icd_normalized']=df[code_col]df['icd_version_flag']=(df['icd_version']=='ICD10').astype(int)df['months_from_cutoff']=((df[date_col]-cutoff).dt.days/30).round(1)returndfdefcreate_diagnosis_phenotype(df:pd.DataFrame,phenotype_name:str,icd10_codes:Set[str],icd9_codes:Optional[Set[str]]=None,re