简介本资源是一个基于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列表以及年龄、性别等对分组可能产生影响的修正因子。它们是分组逻辑中重要的判断条件。数据特点与挑战“顽石”结构嵌套与层级关系原始数据为了表达“MDC-ADRG-DRG”的树状逻辑常采用XML格式或通过父子ID关联的多个CSV表。这种结构适合人类阅读规则文档但不适合直接进行关联分析和批量计算。编码不一致与冗余不同表格间使用的编码体系可能略有差异例如诊断映射表中可能同时存在“诊断编码”和“诊断内部码”需要理清主键。此外可能存在大量为了表达“排除”规则而设置的标志位或特殊字符。非标准化与缺失值文本描述字段可能存在前后空格、换行符、不统一的缩写。某些条件字段可能为空或存在默认值需要明确其业务含义。业务逻辑隐含在数据结构中分组的优先级、互斥关系有时并不直接体现在字段里而是通过记录的顺序或特定字段的组合来暗示这需要结合分组器技术规范进行解读。理解这些我们的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, encodinggbk, sep,): 加载CSV文件自动尝试常见编码 file_path self.data_dir / file_name try: df pd.read_csv(file_path, encodingencoding, sepsep, dtypestr) # 初始全按字符串加载避免类型误判 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, encodingutf-8, sepsep, dtypestr) 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是处理中文医疗数据的第一道坎。dtypestr的初始加载策略很关键它能防止数字编码如‘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, inplaceTrue) # 使用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(., , regexFalse).str.upper() return df staticmethod def handle_duplicates(df, key_columns, keepfirst): 基于关键字段去重并记录日志 initial_count len(df) df_deduped df.drop_duplicates(subsetkey_columns, keepkeep).reset_index(dropTrue) 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_CODEMDC_NAMEADRG_CODEADRG_NAMEDRG_CODEDRG_NAMEWEIGHTDIAG_CONDITIONPROC_CONDITIONMDCA神经系统疾病ADRG001颅脑创伤DRG001A颅脑创伤伴手术2.5S06.*01.21MDCA神经系统疾病ADRG001颅脑创伤DRG001B颅脑创伤不伴手术1.8S06.*为空或特定值MDCA神经系统疾病ADRG002脑血管病DRG002A脑梗死急性期2.1I63.*为空这看起来已经是表格但它的“线性化”程度不够。一条完整的、机器友好的分组逻辑记录应该包含从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(parentsTrue, exist_okTrue) for name, df in df_dict.items(): output_path Path(output_dir) / f{name}.parquet df.to_parquet(output_path, indexFalse) print(f已导出: {output_path}) staticmethod def export_to_sqlite(df_dict, db_path, if_existsreplace): 导出到SQLite数据库便于查询和关联 import sqlite3 conn sqlite3.connect(db_path) for name, df in df_dict.items(): df.to_sql(name, conn, if_existsif_exists, indexFalse) print(f已写入表: {name}) conn.close() print(f数据库已保存至: {db_path}) staticmethod def create_data_dictionary(df_dict, output_path): 生成数据字典记录每个表的字段含义这对业务交接至关重要 with open(output_path, w, encodingutf-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的sepNone自动检测分隔符和enginepython兼容性更好参数。对于关键文件手动用文本编辑器打开查看一小部分最可靠。坑2内存杀手——超大映射表问题诊断映射表可能有数十万行直接explode操作或不当的合并操作如merge可能导致内存溢出。对策分块处理使用pandas.read_csv的chunksize参数分批读入和处理。使用更高效的数据类型将字符串类型的编码字段转换为category类型可以大幅减少内存占用。避免笛卡尔积关联查询时务必先过滤再合并并使用pd.merge并关注on参数。考虑Dask或Modin如果数据量真的巨大可以考虑使用这些兼容Pandas API的并行计算库。坑3业务逻辑的“灰色地带”问题分组规则中常有“主要诊断满足A且手术操作不包括B或伴有并发症C”的复杂逻辑。单纯的数据扁平化无法完全表达这些逻辑。对策线性化处理的是“数据”不是“规则引擎”。我们的系统产出的是干净、结构化的数据层。复杂的布尔逻辑需要在上层的应用层如用SQL的CASE WHEN或用Python写一个规则评估函数来实现。在数据层我们确保每个原子条件如一个诊断码、一个手术码、一个年龄标志都被清晰地提取和标记出来。坑4版本管理的噩梦问题CHS-DRG分组器每年都可能更新。处理不同年份的数据包时字段含义、编码规则可能发生变化。对策配置文件驱动将字段映射关系、清洗规则、编码转换表写在JSON或YAML配置文件中而非硬编码在Python脚本里。不同年份的数据包使用不同的配置文件。数据版本化在输出文件命名或数据库表中加入数据版本号如drg_rules_v1_2_2023。单元测试为关键的数据转换函数编写单元测试用已知的输入输出对来验证。当分组器版本升级时运行测试集能快速发现不兼容的变更。性能优化点向量化操作尽量使用Pandas的向量化函数如str.replace()、applymap代替循环。提前过滤在数据加载后尽早过滤掉不需要的行或列减少后续处理的数据量。使用inplaceTrue参数对于大型DataFrame原地修改比重新赋值更节省内存。输出格式选择对于主要用于Python生态的中间数据Parquet格式在读写速度和压缩比上通常优于CSV。Feather格式的读写速度更快但压缩率较低。构建这样一个系统其价值远不止于完成一次数据转换。它建立了一套可重复、可追溯、可扩展的数据处理流程将杂乱无章的官方数据包变成了随时可以喂养给分析模型、分组模拟程序或可视化报表的“标准粮草”。当你下次再拿到新版CHS-DRG数据包时只需要修改配置文件然后轻松地运行这个流水线一切尽在掌握。这种从被动应对到主动掌控的转变正是数据工程师工作的魅力所在。本文还有配套的精品资源点击获取