Python实战：CHS-DRG数据线性化处理与医疗数据工程实践

简介：本资源是一个基于Python开发的CHS-DRG分组辅助系统，面向医疗机构信息科人员、医保结算工程师及医疗大数据分析学习者，解决DRG分组规则解析难、MDC映射混乱、ADRG判定逻辑不透明等实际问题。项目共2000个文件，含886个核心Python模块（实现诊断编码解析、MCC/CC排除判断、线性化输出等功能）、624个pyc字节码文件、343个备份配置（zbak）、125个文本规则说明及18个dat数据集（如ZD_INFO、SS_VALID等结构化映射表），整体压缩包仅7.32MB，轻量易部署。已有118人学习下载，资源提供完整工程结构、标准化数据提取流程、可复用的规则解析脚本及配套文档，特别适合需快速理解CHS-DRG底层逻辑、开展本地化分组验证或教学演示的技术人员。

1. 项目缘起：当DRG遇上Python，一个数据工程师的实战选择

在医疗信息化领域，DRG（疾病诊断相关分组）是绕不开的核心话题。它不仅是医保支付改革的关键工具，更是医院精细化管理的“指挥棒”。我最初接触CHS-DRG（国家医疗保障疾病诊断相关分组）时，面对的是动辄几十兆、结构复杂、字段繁多的分组器数据包。这些数据通常以XML、CSV或特定数据库格式存在，内含成千上万条分组规则、诊断手术映射关系和权重参数。手动处理？效率低下且极易出错；用传统ETL工具？灵活性不足，难以应对分组逻辑的频繁迭代和复杂的数据清洗需求。

这时，Python进入了我的视野。它并非医疗领域的专属语言，但其强大的数据处理生态（Pandas, NumPy）、灵活的文本解析能力（lxml, json）以及高效的脚本化特性，使其成为构建一个轻量级、可定制化DRG数据提取与线性化处理系统的绝佳选择。这个系统的核心目标很明确：将官方发布的、结构嵌套的CHS-DRG分组器原始数据，通过自动化脚本，提取、清洗、转换并重组为结构扁平、字段清晰、便于后续分析和模型调用的线性化数据表。这不仅是数据格式的转换，更是将复杂的业务规则转化为机器可读、程序可处理的数据资产的关键一步。无论你是医院的数据分析师、医保系统的开发者，还是对医疗数据工程感兴趣的研究者，掌握这套方法，都能让你在面对海量DRG规则数据时，从“手工劳工”解放为“流程指挥官”。

2. 解构CHS-DRG数据包：原始数据的“矿藏”与“顽石”

在动手写代码之前，我们必须像地质学家一样，先弄清楚我们要挖掘的“矿藏”——CHS-DRG数据包——里面到底有什么，以及哪些是“顽石”需要剔除。官方发布的数据包通常是一个压缩文件，解压后你会发现一个结构化的目录。以某版本为例，其核心内容通常包括：

主要数据文件：

• 分组规则主表 (MDC*.xml或DRG_LIST.csv)：这是核心中的核心。它定义了从主要诊断类别（MDC）到具体DRG组的完整路径。每条记录包含了DRG代码、名称、权重、费率、诊断和手术操作入组逻辑等。其结构往往是树状或嵌套的，一个MDC下包含多个ADRG（基础DRG），一个ADRG下又可能细分出多个DRG。
• 诊断与手术操作映射表 (DIAG_MAP.csv,PROC_MAP.csv)：这两张表是“翻译官”。它们将成千上万的ICD-10诊断代码和ICD-9-CM-3手术操作代码，映射到DRG分组器能够识别的内部编码或直接关联到具体的分组条件。数据量巨大，且存在一对多、多对一的复杂映射关系。
• 辅助目录与参数表 (CC_LIST.csv,MCC_LIST.csv,AGE_SEX_INDEX.csv等)：这些表定义了并发症/合并症（CC）、严重并发症/合并症（MCC）列表，以及年龄、性别等对分组可能产生影响的修正因子。它们是分组逻辑中重要的判断条件。

数据特点与挑战（“顽石”）：

1. 结构嵌套与层级关系：原始数据为了表达“MDC->ADRG->DRG”的树状逻辑，常采用XML格式或通过父子ID关联的多个CSV表。这种结构适合人类阅读规则文档，但不适合直接进行关联分析和批量计算。
2. 编码不一致与冗余：不同表格间使用的编码体系可能略有差异，例如诊断映射表中可能同时存在“诊断编码”和“诊断内部码”，需要理清主键。此外，可能存在大量为了表达“排除”规则而设置的标志位或特殊字符。
3. 非标准化与缺失值：文本描述字段可能存在前后空格、换行符、不统一的缩写。某些条件字段可能为空或存在默认值，需要明确其业务含义。
4. 业务逻辑隐含在数据结构中：分组的优先级、互斥关系有时并不直接体现在字段里，而是通过记录的顺序或特定字段的组合来暗示，这需要结合分组器技术规范进行解读。

理解这些，我们的Python处理系统目标就清晰了：将这些分散、嵌套、隐含业务逻辑的“矿藏”，提炼、熔合成一张张字段明确、记录独立、关系清晰的“数据锭”，即线性化处理。

3. 系统核心架构设计：模块化与流水线思维

一个健壮的处理系统不能是“一锅烩”的脚本，而应该像工厂流水线，每个环节职责清晰。我将系统设计为四个核心模块，形成处理流水线。

3.1 数据加载与探查模块

这是流水线的起点，负责将原始数据“搬上”流水线。我们使用Pandas作为核心。

import pandas as pd import os from pathlib import Path class DRGDataLoader: def __init__(self, data_dir): self.data_dir = Path(data_dir) self.raw_data = {} # 用于存储加载后的原始DataFrame def load_csv(self, file_name, encoding='gbk', sep=','): """加载CSV文件，自动尝试常见编码""" file_path = self.data_dir / file_name try: df = pd.read_csv(file_path, encoding=encoding, sep=sep, dtype=str) # 初始全按字符串加载，避免类型误判 self.raw_data[file_name] = df print(f"成功加载 {file_name}, 形状: {df.shape}") return df except UnicodeDecodeError: # 尝试UTF-8 try: df = pd.read_csv(file_path, encoding='utf-8', sep=sep, dtype=str) self.raw_data[file_name] = df print(f"成功加载 {file_name} (UTF-8), 形状: {df.shape}") return df except Exception as e: print(f"加载 {file_name} 失败: {e}") return None def load_xml(self, file_name): """加载XML文件，使用lxml解析为DataFrame""" from lxml import etree file_path = self.data_dir / file_name try: tree = etree.parse(str(file_path)) root = tree.getroot() # 这里需要根据具体的XML结构编写解析逻辑 # 例如，假设每个DRG规则是一个<item>元素 data = [] for item in root.xpath('//item'): record = {} for child in item: record[child.tag] = child.text data.append(record) df = pd.DataFrame(data) self.raw_data[file_name] = df print(f"成功解析XML {file_name}, 形状: {df.shape}") return df except Exception as e: print(f"解析XML {file_name} 失败: {e}") return None def summary(self): """快速探查所有加载数据的概览""" for name, df in self.raw_data.items(): print(f"\n--- {name} ---") print(f"行数: {len(df)}, 列数: {len(df.columns)}") print("前5行:") print(df.head()) print("列名:", df.columns.tolist())

注意：编码问题（GBK vs UTF-8）是处理中文医疗数据的第一道坎。dtype=str的初始加载策略很关键，它能防止数字编码（如‘001’）被误转为整数1，导致后续匹配失败。

3.2 数据清洗与标准化模块

原始数据上流水线后，需要进行“除锈”和“校准”。

class DRGDataCleaner: @staticmethod def strip_whitespace(df): """去除所有字符串字段的首尾空格""" str_cols = df.select_dtypes(include=['object']).columns df[str_cols] = df[str_cols].applymap(lambda x: x.strip() if isinstance(x, str) else x) return df @staticmethod def normalize_null(df, null_values=['NULL', 'null', '', 'NaN', 'NaT']): """将多种形式的空值统一为Python的None或NaN""" df.replace(null_values, pd.NA, inplace=True) # 使用pandas的NA表示空值 return df @staticmethod def standardize_codes(df, code_columns): """标准化编码字段：去除点号、统一为大写等""" for col in code_columns: if col in df.columns: # 例如，将‘A01.1’转为‘A011’，并大写 df[col] = df[col].astype(str).str.replace('.', '', regex=False).str.upper() return df @staticmethod def handle_duplicates(df, key_columns, keep='first'): """基于关键字段去重，并记录日志""" initial_count = len(df) df_deduped = df.drop_duplicates(subset=key_columns, keep=keep).reset_index(drop=True) removed = initial_count - len(df_deduped) if removed > 0: print(f"警告: 在字段 {key_columns} 上发现 {removed} 条重复记录，已移除。") return df_deduped

清洗顺序很重要，通常先做strip_whitespace和normalize_null，再做标准化和去重，避免因空格或空值格式不一致导致去重逻辑失效。

3.3 核心转换：从树状到线性的“降维打击”

这是系统的灵魂所在。我们以处理分组规则主表为例，演示如何将嵌套结构“拍平”。

假设原始分组规则表drg_rules_raw结构如下（简化）：

MDC_CODE	MDC_NAME	ADRG_CODE	ADRG_NAME	DRG_CODE	DRG_NAME	WEIGHT	DIAG_CONDITION	PROC_CONDITION
MDCA	神经系统疾病	ADRG001	颅脑创伤	DRG001A	颅脑创伤伴手术	2.5	S06.*	01.21
MDCA	神经系统疾病	ADRG001	颅脑创伤	DRG001B	颅脑创伤不伴手术	1.8	S06.*	为空或特定值
MDCA	神经系统疾病	ADRG002	脑血管病	DRG002A	脑梗死急性期	2.1	I63.*	为空

这看起来已经是表格，但它的“线性化”程度不够。一条完整的、机器友好的分组逻辑记录，应该包含从MDC到DRG的所有必要信息，且每个字段都是原子性的。此外，DIAG_CONDITION和PROC_CONDITION字段可能包含复杂的模式匹配表达式（如I10-I15表示范围，J18.0|J18.1表示“或”关系）。

class DRGLinearizer: def flatten_hierarchy(self, df_rule): """ 扁平化处理。如果原始数据中MDC/ADRG信息在父记录中， 而DRG在子记录，需要通过关联展开。 本例假设数据已初步关联，主要处理字段原子化。 """ # 1. 确保关键代码字段不为空 df_rule = df_rule.dropna(subset=['DRG_CODE', 'MDC_CODE']).copy() # 2. 构建全局唯一分组键和完整路径描述（可选，便于理解） df_rule['GROUP_PATH'] = df_rule['MDC_CODE'] + '>' + df_rule['ADRG_CODE'] + '>' + df_rule['DRG_CODE'] df_rule['GROUP_FULL_NAME'] = df_rule['MDC_NAME'] + '-' + df_rule['ADRG_NAME'] + '-' + df_rule['DRG_NAME'] # 3. 拆解复杂的条件字段（这是线性化的关键） # 假设DIAG_CONDITION字段包含用‘|’分隔的多个诊断模式 def split_conditions(condition_str): if pd.isna(condition_str): return [] # 按‘|’分割，并清理空格 return [c.strip() for c in str(condition_str).split('|') if c.strip()] df_rule['DIAG_CONDITION_LIST'] = df_rule['DIAG_CONDITION'].apply(split_conditions) df_rule['PROC_CONDITION_LIST'] = df_rule['PROC_CONDITION'].apply(split_conditions) # 4. 将列表展开为多行（如果需要为每个条件生成独立记录） # 这里我们选择保留列表，但也可以使用explode方法展开 # df_exploded = df_rule.explode('DIAG_CONDITION_LIST') print(f"扁平化完成。生成唯一DRG记录 {len(df_rule)} 条。") return df_rule def parse_condition_pattern(self, pattern): """ 解析单个条件模式，如 'I10-I15', 'E11.9', 'S06*'。 返回一个结构化的字典，便于后续匹配。 """ result = {'raw': pattern, 'type': 'exact', 'value': pattern} if '-' in pattern and '*' not in pattern: # 范围模式，如 I10-I15 start, end = pattern.split('-') result['type'] = 'range' result['start'] = start result['end'] = end elif '*' in pattern: # 前缀匹配模式，如 S06* result['type'] = 'prefix' result['value'] = pattern.rstrip('*') elif '|' in pattern: # 或关系，已在上一级拆分，这里按精确处理 pass # 更复杂的正则表达式可以在这里扩展 return result

线性化的核心思想是：让每一条记录都独立表达一个完整的最小分组逻辑单元，并且将复合字段拆解为原子字段或结构化的字段，便于后续的精确匹配和计算。我们选择将条件拆分为列表而非直接展开成多行，是为了在保持DRG主体信息不冗余的前提下，保留完整的条件集。具体选择取决于下游应用是更关注“一个DRG有哪些条件”还是“一个条件对应哪些DRG”。

3.4 输出与持久化模块

处理好的数据需要妥善保存，供下游系统使用。

class DRGDataExporter: @staticmethod def export_to_parquet(df_dict, output_dir): """导出为Parquet格式，兼顾压缩率和读取速度""" Path(output_dir).mkdir(parents=True, exist_ok=True) for name, df in df_dict.items(): output_path = Path(output_dir) / f"{name}.parquet" df.to_parquet(output_path, index=False) print(f"已导出: {output_path}") @staticmethod def export_to_sqlite(df_dict, db_path, if_exists='replace'): """导出到SQLite数据库，便于查询和关联""" import sqlite3 conn = sqlite3.connect(db_path) for name, df in df_dict.items(): df.to_sql(name, conn, if_exists=if_exists, index=False) print(f"已写入表: {name}") conn.close() print(f"数据库已保存至: {db_path}") @staticmethod def create_data_dictionary(df_dict, output_path): """生成数据字典，记录每个表的字段含义，这对业务交接至关重要""" with open(output_path, 'w', encoding='utf-8') as f: f.write("# CHS-DRG线性化数据字典\n\n") for table_name, df in df_dict.items(): f.write(f"## 表名: {table_name}\n") f.write(f"- 记录数: {len(df)}\n") f.write(f"- 字段列表:\n") for col in df.columns: # 这里可以加入从配置文件读取的字段描述 f.write(f" - `{col}`: (类型: {df[col].dtype}) - [请填写业务描述]\n") f.write("\n")

提示：Parquet格式特别适合存储处理后的中型数据，它列式存储、支持压缩，且被Pandas、Spark等工具广泛支持，是数据交换的优秀载体。同时，生成一份数据字典是专业性的体现，能极大降低后续维护和团队协作的成本。

4. 实战串联：构建端到端处理流水线

现在，我们将各个模块像乐高一样拼接起来，形成一个完整的处理流程。我们假设原始数据是CSV格式。

def main_pipeline(data_dir, output_dir): """主处理流水线""" print("=== CHS-DRG数据提取与线性化处理系统启动 ===") # 步骤1: 加载 loader = DRGDataLoader(data_dir) drg_rules_raw = loader.load_csv('DRG_LIST.csv') diag_map_raw = loader.load_csv('DIAG_MAP.csv') proc_map_raw = loader.load_csv('PROC_MAP.csv') # ... 加载其他表 # 步骤2: 清洗 cleaner = DRGDataCleaner() drg_rules_clean = cleaner.strip_whitespace(drg_rules_raw) drg_rules_clean = cleaner.normalize_null(drg_rules_clean) drg_rules_clean = cleaner.standardize_codes(drg_rules_clean, ['MDC_CODE', 'ADRG_CODE', 'DRG_CODE']) drg_rules_clean = cleaner.handle_duplicates(drg_rules_clean, ['DRG_CODE']) # 假设DRG_CODE应唯一 diag_map_clean = cleaner.strip_whitespace(diag_map_raw) diag_map_clean = cleaner.standardize_codes(diag_map_clean, ['DIAG_CODE']) # ... 清洗其他表 # 步骤3: 核心转换与线性化 linearizer = DRGLinearizer() drg_rules_linearized = linearizer.flatten_hierarchy(drg_rules_clean) # 对映射表进行类似处理，确保诊断/手术码标准化 # 例如，将映射关系展开，确保每条记录都是“一个原始编码 -> 一个目标编码/条件” def expand_mapping(df_map, code_col, target_col): """展开可能包含多个目标的映射""" # 假设target_col可能用‘;’分隔多个值 df_expanded = df_map.copy() df_expanded[target_col] = df_expanded[target_col].astype(str).str.split(';') df_expanded = df_expanded.explode(target_col) df_expanded[target_col] = df_expanded[target_col].str.strip() return df_expanded diag_map_expanded = expand_mapping(diag_map_clean, 'DIAG_CODE', 'TARGET_CONDITION') proc_map_expanded = expand_mapping(proc_map_clean, 'PROC_CODE', 'TARGET_CONDITION') # 步骤4: 整合与关联（可选） # 可以将线性化的DRG规则表与展开的映射表通过条件字段进行关联预计算， # 生成一张“诊断/手术码 -> 潜在DRG组”的快速查找表，这能极大加速后续分组模拟。 # 这部分逻辑较复杂，需根据具体业务规则实现。 # 步骤5: 输出 exporter = DRGDataExporter() data_to_export = { 'drg_rules_linear': drg_rules_linearized, 'diag_map_expanded': diag_map_expanded, 'proc_map_expanded': proc_map_expanded, } exporter.export_to_parquet(data_to_export, output_dir) exporter.export_to_sqlite(data_to_export, Path(output_dir)/'drg_data.db') exporter.create_data_dictionary(data_to_export, Path(output_dir)/'数据字典.md') print("=== 处理流程全部完成 ===") # 运行 if __name__ == '__main__': main_pipeline('./raw_data', './processed_data')

5. 避坑指南与性能优化：来自实战的经验

在多次运行这类处理脚本后，我积累了一些宝贵的教训，这些是文档里不会写的“坑”。

坑1：编码与分隔符的“幽灵”

• 问题：CSV文件可能是GB2312、GBK、UTF-8 with BOM、UTF-8 without BOM等多种编码。分隔符可能是逗号、分号或制表符。
• 对策：实现一个“智能探测”加载函数。先用chardet库检测文件编码概率，再尝试用pd.read_csv的sep=None（自动检测分隔符）和engine='python'（兼容性更好）参数。对于关键文件，手动用文本编辑器打开查看一小部分最可靠。

坑2：内存杀手——超大映射表

• 问题：诊断映射表可能有数十万行，直接explode操作或不当的合并操作（如merge）可能导致内存溢出。
• 对策：
  1. 分块处理：使用pandas.read_csv的chunksize参数分批读入和处理。
  2. 使用更高效的数据类型：将字符串类型的编码字段转换为category类型，可以大幅减少内存占用。
  3. 避免笛卡尔积：关联查询时，务必先过滤再合并，并使用pd.merge并关注on参数。
  4. 考虑Dask或Modin：如果数据量真的巨大，可以考虑使用这些兼容Pandas API的并行计算库。

坑3：业务逻辑的“灰色地带”

• 问题：分组规则中常有“主要诊断满足A，且手术操作不包括B，或伴有并发症C”的复杂逻辑。单纯的数据扁平化无法完全表达这些逻辑。
• 对策：线性化处理的是“数据”，不是“规则引擎”。我们的系统产出的是干净、结构化的数据层。复杂的布尔逻辑需要在上层的应用层（如用SQL的CASE WHEN，或用Python写一个规则评估函数）来实现。在数据层，我们确保每个原子条件（如一个诊断码、一个手术码、一个年龄标志）都被清晰地提取和标记出来。

坑4：版本管理的噩梦

• 问题：CHS-DRG分组器每年都可能更新。处理不同年份的数据包时，字段含义、编码规则可能发生变化。
• 对策：
  1. 配置文件驱动：将字段映射关系、清洗规则、编码转换表写在JSON或YAML配置文件中，而非硬编码在Python脚本里。不同年份的数据包使用不同的配置文件。
  2. 数据版本化：在输出文件命名或数据库表中加入数据版本号（如drg_rules_v1_2_2023）。
  3. 单元测试：为关键的数据转换函数编写单元测试，用已知的输入输出对来验证。当分组器版本升级时，运行测试集能快速发现不兼容的变更。

性能优化点：

• 向量化操作：尽量使用Pandas的向量化函数（如str.replace()、applymap）代替循环。
• 提前过滤：在数据加载后尽早过滤掉不需要的行或列，减少后续处理的数据量。
• 使用inplace=True参数：对于大型DataFrame，原地修改比重新赋值更节省内存。
• 输出格式选择：对于主要用于Python生态的中间数据，Parquet格式在读写速度和压缩比上通常优于CSV。Feather格式的读写速度更快，但压缩率较低。

构建这样一个系统，其价值远不止于完成一次数据转换。它建立了一套可重复、可追溯、可扩展的数据处理流程，将杂乱无章的官方数据包，变成了随时可以喂养给分析模型、分组模拟程序或可视化报表的“标准粮草”。当你下次再拿到新版CHS-DRG数据包时，只需要修改配置文件，然后轻松地运行这个流水线，一切尽在掌握。这种从被动应对到主动掌控的转变，正是数据工程师工作的魅力所在。

本文还有配套的精品资源，点击获取