尧图网站开发 
简介这是一套面向医疗信息管理人员、医院DRG专员及Python数据处理学习者的CHS-DRG分组工具实现方案聚焦于解决临床诊断数据向标准化分组映射过程中的规则提取与线性化转换难题。资源以Python为核心技术栈完整封装MDC分类逻辑、ADRG判定规则及MCC/CC排除条件的结构化解析能力可直接支撑医疗机构开展分组运算与医疗质量分析。压缩包共2000个文件含886个可读可调的Python源模块实现核心算法与数据流、624个pyc字节码保障运行效率、125个txt配置说明与映射文档、18个dat结构化数据集如ZD_INFO、SS_VALID等关键规则表整体体积仅7.32MB轻量易部署。已有118人下载学习用户可获得从原始CHS-DRG标准到可执行分组逻辑的全链路代码实现、清晰分层的工程目录结构含授权协议与版本配置以及面向真实医疗数据字段的预处理与规则加载范式。1. 项目概述从DRG到代码的桥梁在医疗信息化和医保支付改革的大背景下DRG疾病诊断相关分组已经从一个专业术语变成了医院管理者、医保经办人员乃至临床医生都绕不开的核心工具。它本质上是一套“病例组合”方案将临床过程相近、资源消耗相似的病例归入同一组并据此进行打包付费。而CHS-DRG即国家医疗保障疾病诊断相关分组则是我们国内统一执行的官方标准版本。这个项目的核心就是要用Python这把“瑞士军刀”去构建一个能够自动处理CHS-DRG分组逻辑的系统。听起来可能有点抽象我打个比方。CHS-DRG分组手册就像一本极其复杂的“烹饪食谱”里面规定了成千上万种“菜品”病例的“烹饪规则”分组逻辑。传统的分组方式是人工拿着这本厚厚的食谱对照着病例的“食材清单”主要诊断、手术操作、并发症等一页一页去翻判断这道菜该归到哪一类。这个过程不仅效率低下而且极易出错。我们这个Python项目就是要写一个“智能厨娘”她能自动读取食谱CHS-DRG分组器逻辑然后快速、准确地处理海量的“食材清单”医院病案首页数据最终输出每一道“菜”的标准化分类结果。这个系统的价值不言而喻。对于医院它可以用于住院病例的预分组和费用模拟帮助科室进行成本控制和绩效管理对于医保部门它可以作为审核和结算的辅助工具提升效率和公平性。而用Python来实现则是因为其强大的数据处理生态如Pandas, NumPy、清晰的语法以及丰富的规则引擎库非常适合处理这种逻辑复杂但规则明确的任务。接下来我就带你一步步拆解如何从零开始搭建这个“智能厨娘”系统。2. 核心需求与系统设计解析在动手写代码之前我们必须把CHS-DRG分组器的“脾气”摸透。一个完整的分组过程远不止简单的“if-else”判断它是一条严谨的逻辑流水线。我们的系统设计必须紧扣这条流水线的每一个环节。2.1 DRG分组核心流程拆解CHS-DRG的分组路径可以概括为“三步走”主要诊断大类MDC判断这是第一道关卡。根据病案首页上的“主要诊断”编码通常是ICD-10将病例划分到26个MDC中的一个比如“MDCA 先期分组疾病及相关操作”或“MDCF 循环系统疾病”。这一步错了后面全错。外科部分ADRG判断进入某个MDC后再看病例是否有手术操作。如果有则根据主要手术操作编码通常是ICD-9-CM-3进入对应的外科ADRG如果没有则进入内科ADRG。这一步开始引入“主要操作”这个关键变量。最终DRGDRG判断在ADRG的基础上再考虑患者的年龄、体重新生儿、并发症与合并症CC、严重并发症与合并症MCC等因素进行细分得到最终的DRG代码。这一步是逻辑最复杂、规则最琐碎的部分涉及大量的条件组合。我们的系统就是要用代码精准地模拟这三步。输入是结构化的病案首页数据一个包含诊断、操作、患者基本信息等的字典或DataFrame行输出是MDC、ADRG、DRG的编码以及完整的逻辑路径。2.2 系统架构设计思路基于上述流程一个稳健的DRG分组器系统应该包含以下核心模块数据接口层负责从各种数据源如数据库、Excel、CSV文件读取原始病案数据并进行初步的清洗和校验比如诊断/操作编码格式是否正确、必填字段是否缺失。规则配置与加载模块这是系统的“大脑”。CHS-DRG的分组规则诊断目录、操作目录、分组器逻辑表通常以Excel或数据库表的形式发布。我们需要设计一个规则加载器将这些静态规则表如MDC表、内科ADRG表、外科ADRG表、CC/MCC排除表等读入内存并组织成便于快速查询的数据结构比如字典嵌套字典、或使用Pandas的DataFrame建立索引。核心分组引擎这是系统的“心脏”。它接收清洗后的单条病例数据按照“MDC - ADRG - DRG”的流程调用规则配置进行一步步的逻辑判断。这里会大量使用到查找Mapping和条件判断。线性化处理与输出模块分组完成后原始的输出可能是一个复杂的对象。线性化处理旨在将这个结果“压平”转换成一条标准的、字段明确的记录方便写入数据库或导出为报表。例如将分组路径“MDCA - ADRG1 - DRG1”以及用到的关键诊断、操作编码都作为字段输出。批处理与性能模块医院的数据量动辄数十万、百万级。系统必须支持高效批处理利用Python的多进程multiprocessing或异步IOasyncio来提升吞吐量。注意在规则加载的设计上一个常见的“坑”是规则表的更新。CHS-DRG版本可能每年更新。我们的系统必须做到规则与代码解耦即分组逻辑的变化只需要更新规则配置文件而不需要或极少需要修改核心引擎代码。这通常通过将规则设计为可配置的“决策表”来实现。3. 关键技术实现细节理论说完了我们进入实战环节。我会用代码片段和具体例子展示几个最关键部分的实现。3.1 规则数据的结构化加载假设我们有一个mdc_mapping.xlsx文件其中一列是ICD10_CODE诊断编码一列是MDC_CODEMDC编码。编码可能是范围如A00-A09或单个编码。import pandas as pd import re class RuleLoader: def __init__(self, rule_path): self.rule_path rule_path self.mdc_map {} # 用于存储诊断码到MDC的映射 self.cc_list [] # 存储CC列表 self.mcc_list [] # 存储MCC列表 def load_mdc_mapping(self): 加载MDC映射表处理编码范围 df pd.read_excel(f{self.rule_path}/mdc_mapping.xlsx) for _, row in df.iterrows(): code_range row[ICD10_CODE] mdc row[MDC_CODE] # 处理编码范围如 A00-A09 if - in code_range: start, end code_range.split(-) # 将编码范围展开为具体编码列表简化处理实际需考虑编码规则 # 这里示例将范围转换为前缀匹配逻辑更严谨的做法是使用编码树 self.mdc_map[code_range] mdc # 先存储范围规则 else: self.mdc_map[code_range] mdc # 存储具体编码 # 实际项目中这里会构建更高效的查找结构如前缀树(Trie) def load_cc_mcc(self): 加载CC/MCC排除表 df_cc pd.read_excel(f{self.rule_path}/cc_list.xlsx) self.cc_list df_cc[ICD10_CODE].tolist() df_mcc pd.read_excel(f{self.rule_path}/mcc_list.xlsx) self.mcc_list df_mcc[ICD10_CODE].tolist() def get_mdc_for_diagnosis(self, diag_code): 根据诊断编码获取MDC # 1. 先尝试精确匹配 if diag_code in self.mdc_map: return self.mdc_map[diag_code] # 2. 尝试匹配范围规则 (简化逻辑实际需解析范围) for code_range, mdc in self.mdc_map.items(): if - in code_range: start, end code_range.split(-) if start diag_code end: # 字符串比较适用于像‘A00’这样的编码 return mdc # 3. 未找到返回默认或抛出异常 return MDCZZ # 或返回None根据业务定义这个RuleLoader类负责将所有规则加载到内存中。注意对于编码范围的匹配上述简化代码使用字符串比较这在ICD-10编码如A00.0上可能不准确。生产环境中需要将编码解析为可比较的格式或使用专门的医学编码库。3.2 核心分组引擎的实现分组引擎是业务逻辑最集中的地方。我们将其实现为一个类它持有RuleLoader实例并对单条病例数据进行处理。class DRGGrouper: def __init__(self, rule_loader): self.rule_loader rule_loader self.patient_data None def group(self, patient_data): 对单条病例数据进行分组 self.patient_data patient_data result { patient_id: patient_data[patient_id], main_diag: patient_data[main_diagnosis], operations: patient_data.get(operations, []), age: patient_data[age], weight: patient_data.get(weight), # 新生儿体重 secondary_diags: patient_data.get(secondary_diagnoses, []) } # 步骤1: 确定MDC mdc self._determine_mdc(result[main_diag]) result[mdc] mdc if not mdc: result[drg] 000 # 无法入组 return result # 步骤2: 确定ADRG先判断外科部分 adrg self._determine_adrg(mdc, result[main_diag], result[operations]) result[adrg] adrg if not adrg: result[drg] 000 # 无法入组 return result # 步骤3: 判断CC/MCC并确定最终DRG has_mcc, has_cc self._check_cc_mcc(result[secondary_diags]) drg self._determine_drg(adrg, result[age], result[weight], has_mcc, has_cc) result[has_mcc] has_mcc result[has_cc] has_cc result[drg] drg return result def _determine_mdc(self, main_diag): return self.rule_loader.get_mdc_for_diagnosis(main_diag) def _determine_adrg(self, mdc, main_diag, operations): # 1. 检查是否有主要手术操作 if operations: primary_op operations[0] # 假设第一个是主要操作 # 根据mdc和primary_op去外科ADRG映射表中查找 adrg self._lookup_surgical_adrg(mdc, primary_op) if adrg: return adrg # 2. 无手术或未找到外科ADRG查找内科ADRG return self._lookup_medical_adrg(mdc, main_diag) def _check_cc_mcc(self, secondary_diags): has_mcc False has_cc False for diag in secondary_diags: if diag in self.rule_loader.mcc_list: has_mcc True # MCC优先级最高找到一个即可中断需看规则有时是看是否有任何一个MCC # break if diag in self.rule_loader.cc_list: has_cc True return has_mcc, has_cc def _determine_drg(self, adrg, age, weight, has_mcc, has_cc): # 这是一个复杂的决策过程通常需要查表 # 例如根据adrg找到其下所有的DRG子组然后根据年龄、体重、CC/MCC状态筛选 # 这里用伪代码表示 drg_candidates self.rule_loader.get_drg_candidates(adrg) for candidate in drg_candidates: if self._meets_drg_criteria(candidate, age, weight, has_mcc, has_cc): return candidate[drg_code] return adrg Z # 返回ADRG对应的无CC/MCC组或默认组 # 以下为查找函数示例实际需要从rule_loader中获取对应的映射表 def _lookup_surgical_adrg(self, mdc, operation): # 伪代码 return surgical_map.get((mdc, operation)) pass def _lookup_medical_adrg(self, mdc, diagnosis): # 伪代码 return medical_map.get((mdc, diagnosis)) pass这个DRGGrouper类清晰地勾勒出了分组的主干逻辑。_determine_adrg方法体现了“先外科后内科”的核心原则。_check_cc_mcc方法展示了如何利用预加载的CC/MCC列表进行快速判断。3.3 数据线性化处理分组引擎输出的result字典可能嵌套较深比如operations是个列表。为了便于存储和交换我们需要将其“线性化”。def linearize_grouping_result(result_dict): 将分组结果字典扁平化。 输入: DRGGrouper.group() 返回的字典 输出: 一个扁平字典所有值为基本类型str, int, float, bool linear {} # 直接复制简单字段 for key in [patient_id, main_diag, mdc, adrg, drg, has_mcc, has_cc, age]: if key in result_dict: linear[key] result_dict[key] # 处理列表字段拼接成字符串 if operations in result_dict and result_dict[operations]: linear[operations] ;.join(result_dict[operations]) else: linear[operations] if secondary_diags in result_dict and result_dict[secondary_diags]: linear[secondary_diags] ;.join(result_dict[secondary_diags]) else: linear[secondary_diags] # 可以添加分组路径 linear[grouping_path] f{linear.get(mdc, )}-{linear.get(adrg, )}-{linear.get(drg, )} # 添加时间戳和处理状态 from datetime import datetime linear[process_time] datetime.now().isoformat() linear[status] SUCCESS if result_dict.get(drg) ! 000 else FAILED return linear # 使用示例 grouped_result grouper.group(sample_patient) flat_record linearize_grouping_result(grouped_result) # 此时 flat_record 可以直接用 pandas.DataFrame.append 或 to_sql 写入数据库线性化后每条病例就变成了一条拥有固定字段的平坦记录非常适合存入关系型数据库的表格中或者导出为CSV文件进行后续分析。4. 性能优化与批量处理实战当面对数万甚至百万级的病案数据时单条处理的速度是无法接受的。我们必须引入批处理和并发技术。4.1 利用Pandas进行向量化批处理对于规则匹配这类操作如果能将数据全部放入Pandas DataFrame并利用其向量化操作速度会有质的飞跃。例如为大量诊断编码匹配MDCimport pandas as pd import numpy as np def batch_match_mdc(df_patients, rule_loader): df_patients: DataFrame包含‘main_diagnosis’列 rule_loader: 已加载规则的RuleLoader实例 返回添加了‘mdc’列的DataFrame # 方法一使用apply相对灵活但非完全向量化 # df_patients[mdc] df_patients[main_diagnosis].apply(rule_loader.get_mdc_for_diagnosis) # 方法二构建映射字典然后使用map更高效 # 假设我们已经将规则加载成了一个全面的映射字典 all_mdc_map (包含具体编码和范围) # 这里需要先将范围规则预处理掉生成一个所有可能编码到MDC的完整映射如果范围不大可以这么做 # 以下为简化示例假设我们只有精确匹配 all_mdc_map rule_loader.mdc_map # 这里假设mdc_map已是完整映射 df_patients[mdc] df_patients[main_diagnosis].map(all_mdc_map) # 处理未匹配到的map返回NaN df_patients[mdc] df_patients[mdc].fillna(MDCZZ) # 默认组 return df_patients实操心得在数据量极大时map函数比apply快得多。但前提是映射字典要尽可能完整。对于CHS-DRG这种有范围规则的映射一个折中的办法是先将所有范围规则展开成具体编码列表如果范围可枚举且数量可接受生成一个巨大的映射字典。如果范围不可枚举则需要写一个自定义函数并在apply中结合一些缓存机制来优化。4.2 使用多进程并行处理如果单机内存足够但CPU有多核我们可以将数据分块利用multiprocessing库进行并行分组。from multiprocessing import Pool, cpu_count import pandas as pd def parallel_grouping(data_chunk, rule_loader): 处理一个数据块的函数 grouper DRGGrouper(rule_loader) # 每个进程有自己的grouper实例 results [] for _, row in data_chunk.iterrows(): patient_dict row.to_dict() result grouper.group(patient_dict) results.append(linearize_grouping_result(result)) return pd.DataFrame(results) def batch_group_parallel(df_master, rule_loader, n_jobsNone): 主函数并行批处理分组 if n_jobs is None: n_jobs max(1, cpu_count() - 1) # 留一个核心给系统 # 将大数据框拆分成多个块 chunk_size max(1, len(df_master) // n_jobs) chunks [df_master.iloc[i:i chunk_size] for i in range(0, len(df_master), chunk_size)] # 创建进程池 with Pool(processesn_jobs) as pool: # 使用starmap传递多个参数这里需要将rule_loader也传递进去。 # 注意rule_loader对象需要是可序列化的(picklable)如果包含复杂的自定义对象可能有问题。 # 更稳妥的方式是在每个子进程中重新初始化规则加载器从文件读取。 args [(chunk, rule_loader) for chunk in chunks] result_dfs pool.starmap(parallel_grouping, args) # 合并所有结果 final_df pd.concat(result_dfs, ignore_indexTrue) return final_df # 使用前需要确保rule_loader是可pickle的或者修改parallel_grouping函数在内部重新加载规则。重要提示多进程编程中每个进程都有独立的内存空间。传递rule_loader对象时如果它内部有大量数据如加载的规则表会被序列化并复制到每个进程可能导致内存消耗剧增。一种优化方案是在每个子进程内部重新从文件或共享内存中加载规则。另一种更高级的方案是使用multiprocessing.Manager创建共享字典但会牺牲一些速度。5. 部署、调试与常见问题排查系统开发完成后如何让它稳定可靠地运行起来才是真正的挑战。5.1 系统部署与配置管理一个完整的项目应该包含清晰的配置文件将一切可变的参数外部化。# config.yaml data: input_path: ./data/input/ output_path: ./data/output/ rule_path: ./rules/chs_drg_v1.2/ grouping: default_mdc_for_unmatched: MDCZZ default_drg_for_unmatched: 000 enable_cc_mcc_check: true processing: batch_size: 1000 use_multiprocessing: true n_jobs: 4 logging: level: INFO file: ./logs/drg_grouper.log在代码中使用PyYAML库读取配置。这样当规则路径、默认值或处理参数需要变更时只需修改配置文件无需触动代码。5.2 详尽的日志记录日志是调试和监控的生命线。务必在关键步骤和异常处添加日志。import logging import sys def setup_logging(config): log_level getattr(logging, config[logging][level]) log_file config[logging][file] logging.basicConfig( levellog_level, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(log_file, encodingutf-8), logging.StreamHandler(sys.stdout) # 同时输出到控制台 ] ) return logging.getLogger(__name__) # 在分组引擎中使用 logger logging.getLogger(__name__) class DRGGrouper: def group(self, patient_data): patient_id patient_data.get(patient_id, UNKNOWN) logger.info(f开始处理病例: {patient_id}) try: # ... 分组逻辑 ... if not mdc: logger.warning(f病例 {patient_id} 主要诊断 {main_diag} 未找到MDC映射入组失败。) result[drg] 000 result[status] FAILED_MDC return result # ... 更多逻辑 ... logger.info(f病例 {patient_id} 分组成功: DRG{drg}) except Exception as e: logger.error(f处理病例 {patient_id} 时发生未知错误: {e}, exc_infoTrue) result[drg] 000 result[status] ERROR return result5.3 常见问题排查手册在实际运行中你一定会遇到各种各样的问题。下面这个表格整理了我踩过的一些“坑”和解决方法问题现象可能原因排查步骤与解决方案大量病例无法入组DRG为‘000’1. 诊断/操作编码格式与规则表不匹配。2. 规则表未加载或加载错误。3. 主要诊断字段为空或无效。1.检查编码格式对比一个失败病例的诊断编码和规则表中的编码格式是否带点字母大小写。使用df[main_diag].str.strip().unique()查看所有编码变体。2.验证规则加载打印rule_loader.mdc_map的前几项检查是否成功加载。确认规则文件路径正确。3.数据质量检查运行df[main_diagnosis].isnull().sum()统计空值。分组结果与官方工具不一致1. CC/MCC判断逻辑有误。2. 手术操作优先级或主要操作判断错误。3. 年龄、新生儿体重等细分规则未生效。1.CC/MCC复核提取一个结果不一致的病例手动检查其次要诊断是否在CC/MCC列表中。确认你的CC/MCC列表是否完整、版本是否正确。2.手术逻辑检查确认你的“主要手术操作”判定逻辑是否与CHS-DRG规则一致通常是资源消耗最大、技术难度最高的。3.细分规则调试在_determine_drg函数中增加详细日志打印出所有候选DRG和判断条件逐步比对。程序运行速度慢内存占用高1. 单条处理未使用批处理。2. 多进程时规则数据被重复复制。3. Pandas操作未优化产生中间副本。1.向量化尽可能使用Pandas的map、apply和向量化运算避免Python层级的for循环。2.共享内存考虑使用multiprocessing.shared_memory或第三方库如ray、dask来共享只读的规则数据。3.内存管理处理完一个数据块后及时使用del释放变量或使用pandas.read_csv的chunksize参数流式处理超大文件。更新DRG版本后分组结果大面积变化1. 规则文件未正确更新。2. 程序缓存了旧的规则数据。1.版本控制在配置文件和日志中明确记录使用的规则版本号如CHS-DRG v1.2。2.清除缓存如果规则数据被缓存如使用joblib或自定义缓存更新文件后务必清除缓存或重启服务。3.做版本对比测试用同一份测试数据集分别用新旧版本规则运行对比结果差异确认符合预期。5.4 单元测试与集成测试为了保证分组逻辑的准确性必须编写严格的测试。import unittest from your_module import DRGGrouper, RuleLoader class TestDRGGrouper(unittest.TestCase): classmethod def setUpClass(cls): 所有测试开始前执行一次加载规则 cls.rule_loader RuleLoader(./test_rules/) cls.rule_loader.load_all_rules() cls.grouper DRGGrouper(cls.rule_loader) def test_mdc_assignment(self): 测试MDC判断是否正确 test_cases [ ({main_diagnosis: A01.0}, MDCA), # 假设A01.0属于MDCA ({main_diagnosis: I10}, MDCF), # 假设I10属于MDCF ({main_diagnosis: INVALID}, MDCZZ), # 无效编码 ] for input_data, expected_mdc in test_cases: with self.subTest(input_datainput_data): result self.grouper.group(input_data) self.assertEqual(result[mdc], expected_mdc) def test_surgical_adrg(self): 测试有手术操作的病例是否能正确进入外科ADRG patient_with_op { main_diagnosis: K35.0, # 急性阑尾炎 operations: [47.01], # 阑尾切除术 age: 30, secondary_diagnoses: [] } result self.grouper.group(patient_with_op) # 断言结果中的ADRG应该是外科相关的例如‘GD1’ self.assertTrue(result[adrg].startswith(G)) # 假设外科ADRG以G开头 self.assertNotEqual(result[drg], 000) def test_cc_mcc_impact(self): 测试CC/MCC对最终DRG的影响 base_patient {main_diagnosis: I21.0, operations: [], age: 65} patient_without_cc {**base_patient, secondary_diagnoses: []} patient_with_cc {**base_patient, secondary_diagnoses: [E11.9]} # 糖尿病 result_no_cc self.grouper.group(patient_without_cc) result_with_cc self.grouper.group(patient_with_cc) # 同一个ADRG下有CC的DRG应该与无CC的DRG不同 self.assertEqual(result_no_cc[adrg], result_with_cc[adrg]) self.assertNotEqual(result_no_cc[drg], result_with_cc[drg]) if __name__ __main__: unittest.main()编写覆盖主要分组路径、边界条件如新生儿、极高龄患者和异常情况如编码缺失的测试用例是确保代码长期稳定运行的最有效手段。每次规则更新后都应跑一遍完整的测试集。这个Python实现的CHS-DRG分组器从设计到实现再到优化和排错是一个典型的将复杂业务规则转化为自动化系统的过程。它的核心价值在于将一致性、效率与可维护性结合了起来。人工分组难免会有疏漏和疲劳导致的错误而系统只要规则配置正确每一次计算都是严格一致的。处理速度从“天”级别提升到“分钟”甚至“秒”级别解放了人力。最后通过模块化设计和配置化管理系统能够相对平滑地适应CHS-DRG规则的年度更新。当然真实世界的挑战远不止于此比如诊断编码的归一化、手术操作链的复杂判断等但有了这个坚实的基础框架后续的扩展和深化就有了明确的路径。本文还有配套的精品资源点击获取 
简介这是一套面向医疗信息管理人员、医院DRG专员及Python数据处理学习者的CHS-DRG分组工具实现方案聚焦于解决临床诊断数据向标准化分组映射过程中的规则提取与线性化转换难题。资源以Python为核心技术栈完整封装MDC分类逻辑、ADRG判定规则及MCC/CC排除条件的结构化解析能力可直接支撑医疗机构开展分组运算与医疗质量分析。压缩包共2000个文件含886个可读可调的Python源模块实现核心算法与数据流、624个pyc字节码保障运行效率、125个txt配置说明与映射文档、18个dat结构化数据集如ZD_INFO、SS_VALID等关键规则表整体体积仅7.32MB轻量易部署。已有118人下载学习用户可获得从原始CHS-DRG标准到可执行分组逻辑的全链路代码实现、清晰分层的工程目录结构含授权协议与版本配置以及面向真实医疗数据字段的预处理与规则加载范式。1. 项目概述从DRG到代码的桥梁在医疗信息化和医保支付改革的大背景下DRG疾病诊断相关分组已经从一个专业术语变成了医院管理者、医保经办人员乃至临床医生都绕不开的核心工具。它本质上是一套“病例组合”方案将临床过程相近、资源消耗相似的病例归入同一组并据此进行打包付费。而CHS-DRG即国家医疗保障疾病诊断相关分组则是我们国内统一执行的官方标准版本。这个项目的核心就是要用Python这把“瑞士军刀”去构建一个能够自动处理CHS-DRG分组逻辑的系统。听起来可能有点抽象我打个比方。CHS-DRG分组手册就像一本极其复杂的“烹饪食谱”里面规定了成千上万种“菜品”病例的“烹饪规则”分组逻辑。传统的分组方式是人工拿着这本厚厚的食谱对照着病例的“食材清单”主要诊断、手术操作、并发症等一页一页去翻判断这道菜该归到哪一类。这个过程不仅效率低下而且极易出错。我们这个Python项目就是要写一个“智能厨娘”她能自动读取食谱CHS-DRG分组器逻辑然后快速、准确地处理海量的“食材清单”医院病案首页数据最终输出每一道“菜”的标准化分类结果。这个系统的价值不言而喻。对于医院它可以用于住院病例的预分组和费用模拟帮助科室进行成本控制和绩效管理对于医保部门它可以作为审核和结算的辅助工具提升效率和公平性。而用Python来实现则是因为其强大的数据处理生态如Pandas, NumPy、清晰的语法以及丰富的规则引擎库非常适合处理这种逻辑复杂但规则明确的任务。接下来我就带你一步步拆解如何从零开始搭建这个“智能厨娘”系统。2. 核心需求与系统设计解析在动手写代码之前我们必须把CHS-DRG分组器的“脾气”摸透。一个完整的分组过程远不止简单的“if-else”判断它是一条严谨的逻辑流水线。我们的系统设计必须紧扣这条流水线的每一个环节。2.1 DRG分组核心流程拆解CHS-DRG的分组路径可以概括为“三步走”主要诊断大类MDC判断这是第一道关卡。根据病案首页上的“主要诊断”编码通常是ICD-10将病例划分到26个MDC中的一个比如“MDCA 先期分组疾病及相关操作”或“MDCF 循环系统疾病”。这一步错了后面全错。外科部分ADRG判断进入某个MDC后再看病例是否有手术操作。如果有则根据主要手术操作编码通常是ICD-9-CM-3进入对应的外科ADRG如果没有则进入内科ADRG。这一步开始引入“主要操作”这个关键变量。最终DRGDRG判断在ADRG的基础上再考虑患者的年龄、体重新生儿、并发症与合并症CC、严重并发症与合并症MCC等因素进行细分得到最终的DRG代码。这一步是逻辑最复杂、规则最琐碎的部分涉及大量的条件组合。我们的系统就是要用代码精准地模拟这三步。输入是结构化的病案首页数据一个包含诊断、操作、患者基本信息等的字典或DataFrame行输出是MDC、ADRG、DRG的编码以及完整的逻辑路径。2.2 系统架构设计思路基于上述流程一个稳健的DRG分组器系统应该包含以下核心模块数据接口层负责从各种数据源如数据库、Excel、CSV文件读取原始病案数据并进行初步的清洗和校验比如诊断/操作编码格式是否正确、必填字段是否缺失。规则配置与加载模块这是系统的“大脑”。CHS-DRG的分组规则诊断目录、操作目录、分组器逻辑表通常以Excel或数据库表的形式发布。我们需要设计一个规则加载器将这些静态规则表如MDC表、内科ADRG表、外科ADRG表、CC/MCC排除表等读入内存并组织成便于快速查询的数据结构比如字典嵌套字典、或使用Pandas的DataFrame建立索引。核心分组引擎这是系统的“心脏”。它接收清洗后的单条病例数据按照“MDC - ADRG - DRG”的流程调用规则配置进行一步步的逻辑判断。这里会大量使用到查找Mapping和条件判断。线性化处理与输出模块分组完成后原始的输出可能是一个复杂的对象。线性化处理旨在将这个结果“压平”转换成一条标准的、字段明确的记录方便写入数据库或导出为报表。例如将分组路径“MDCA - ADRG1 - DRG1”以及用到的关键诊断、操作编码都作为字段输出。批处理与性能模块医院的数据量动辄数十万、百万级。系统必须支持高效批处理利用Python的多进程multiprocessing或异步IOasyncio来提升吞吐量。注意在规则加载的设计上一个常见的“坑”是规则表的更新。CHS-DRG版本可能每年更新。我们的系统必须做到规则与代码解耦即分组逻辑的变化只需要更新规则配置文件而不需要或极少需要修改核心引擎代码。这通常通过将规则设计为可配置的“决策表”来实现。3. 关键技术实现细节理论说完了我们进入实战环节。我会用代码片段和具体例子展示几个最关键部分的实现。3.1 规则数据的结构化加载假设我们有一个mdc_mapping.xlsx文件其中一列是ICD10_CODE诊断编码一列是MDC_CODEMDC编码。编码可能是范围如A00-A09或单个编码。import pandas as pd import re class RuleLoader: def __init__(self, rule_path): self.rule_path rule_path self.mdc_map {} # 用于存储诊断码到MDC的映射 self.cc_list [] # 存储CC列表 self.mcc_list [] # 存储MCC列表 def load_mdc_mapping(self): 加载MDC映射表处理编码范围 df pd.read_excel(f{self.rule_path}/mdc_mapping.xlsx) for _, row in df.iterrows(): code_range row[ICD10_CODE] mdc row[MDC_CODE] # 处理编码范围如 A00-A09 if - in code_range: start, end code_range.split(-) # 将编码范围展开为具体编码列表简化处理实际需考虑编码规则 # 这里示例将范围转换为前缀匹配逻辑更严谨的做法是使用编码树 self.mdc_map[code_range] mdc # 先存储范围规则 else: self.mdc_map[code_range] mdc # 存储具体编码 # 实际项目中这里会构建更高效的查找结构如前缀树(Trie) def load_cc_mcc(self): 加载CC/MCC排除表 df_cc pd.read_excel(f{self.rule_path}/cc_list.xlsx) self.cc_list df_cc[ICD10_CODE].tolist() df_mcc pd.read_excel(f{self.rule_path}/mcc_list.xlsx) self.mcc_list df_mcc[ICD10_CODE].tolist() def get_mdc_for_diagnosis(self, diag_code): 根据诊断编码获取MDC # 1. 先尝试精确匹配 if diag_code in self.mdc_map: return self.mdc_map[diag_code] # 2. 尝试匹配范围规则 (简化逻辑实际需解析范围) for code_range, mdc in self.mdc_map.items(): if - in code_range: start, end code_range.split(-) if start diag_code end: # 字符串比较适用于像‘A00’这样的编码 return mdc # 3. 未找到返回默认或抛出异常 return MDCZZ # 或返回None根据业务定义这个RuleLoader类负责将所有规则加载到内存中。注意对于编码范围的匹配上述简化代码使用字符串比较这在ICD-10编码如A00.0上可能不准确。生产环境中需要将编码解析为可比较的格式或使用专门的医学编码库。3.2 核心分组引擎的实现分组引擎是业务逻辑最集中的地方。我们将其实现为一个类它持有RuleLoader实例并对单条病例数据进行处理。class DRGGrouper: def __init__(self, rule_loader): self.rule_loader rule_loader self.patient_data None def group(self, patient_data): 对单条病例数据进行分组 self.patient_data patient_data result { patient_id: patient_data[patient_id], main_diag: patient_data[main_diagnosis], operations: patient_data.get(operations, []), age: patient_data[age], weight: patient_data.get(weight), # 新生儿体重 secondary_diags: patient_data.get(secondary_diagnoses, []) } # 步骤1: 确定MDC mdc self._determine_mdc(result[main_diag]) result[mdc] mdc if not mdc: result[drg] 000 # 无法入组 return result # 步骤2: 确定ADRG先判断外科部分 adrg self._determine_adrg(mdc, result[main_diag], result[operations]) result[adrg] adrg if not adrg: result[drg] 000 # 无法入组 return result # 步骤3: 判断CC/MCC并确定最终DRG has_mcc, has_cc self._check_cc_mcc(result[secondary_diags]) drg self._determine_drg(adrg, result[age], result[weight], has_mcc, has_cc) result[has_mcc] has_mcc result[has_cc] has_cc result[drg] drg return result def _determine_mdc(self, main_diag): return self.rule_loader.get_mdc_for_diagnosis(main_diag) def _determine_adrg(self, mdc, main_diag, operations): # 1. 检查是否有主要手术操作 if operations: primary_op operations[0] # 假设第一个是主要操作 # 根据mdc和primary_op去外科ADRG映射表中查找 adrg self._lookup_surgical_adrg(mdc, primary_op) if adrg: return adrg # 2. 无手术或未找到外科ADRG查找内科ADRG return self._lookup_medical_adrg(mdc, main_diag) def _check_cc_mcc(self, secondary_diags): has_mcc False has_cc False for diag in secondary_diags: if diag in self.rule_loader.mcc_list: has_mcc True # MCC优先级最高找到一个即可中断需看规则有时是看是否有任何一个MCC # break if diag in self.rule_loader.cc_list: has_cc True return has_mcc, has_cc def _determine_drg(self, adrg, age, weight, has_mcc, has_cc): # 这是一个复杂的决策过程通常需要查表 # 例如根据adrg找到其下所有的DRG子组然后根据年龄、体重、CC/MCC状态筛选 # 这里用伪代码表示 drg_candidates self.rule_loader.get_drg_candidates(adrg) for candidate in drg_candidates: if self._meets_drg_criteria(candidate, age, weight, has_mcc, has_cc): return candidate[drg_code] return adrg Z # 返回ADRG对应的无CC/MCC组或默认组 # 以下为查找函数示例实际需要从rule_loader中获取对应的映射表 def _lookup_surgical_adrg(self, mdc, operation): # 伪代码 return surgical_map.get((mdc, operation)) pass def _lookup_medical_adrg(self, mdc, diagnosis): # 伪代码 return medical_map.get((mdc, diagnosis)) pass这个DRGGrouper类清晰地勾勒出了分组的主干逻辑。_determine_adrg方法体现了“先外科后内科”的核心原则。_check_cc_mcc方法展示了如何利用预加载的CC/MCC列表进行快速判断。3.3 数据线性化处理分组引擎输出的result字典可能嵌套较深比如operations是个列表。为了便于存储和交换我们需要将其“线性化”。def linearize_grouping_result(result_dict): 将分组结果字典扁平化。 输入: DRGGrouper.group() 返回的字典 输出: 一个扁平字典所有值为基本类型str, int, float, bool linear {} # 直接复制简单字段 for key in [patient_id, main_diag, mdc, adrg, drg, has_mcc, has_cc, age]: if key in result_dict: linear[key] result_dict[key] # 处理列表字段拼接成字符串 if operations in result_dict and result_dict[operations]: linear[operations] ;.join(result_dict[operations]) else: linear[operations] if secondary_diags in result_dict and result_dict[secondary_diags]: linear[secondary_diags] ;.join(result_dict[secondary_diags]) else: linear[secondary_diags] # 可以添加分组路径 linear[grouping_path] f{linear.get(mdc, )}-{linear.get(adrg, )}-{linear.get(drg, )} # 添加时间戳和处理状态 from datetime import datetime linear[process_time] datetime.now().isoformat() linear[status] SUCCESS if result_dict.get(drg) ! 000 else FAILED return linear # 使用示例 grouped_result grouper.group(sample_patient) flat_record linearize_grouping_result(grouped_result) # 此时 flat_record 可以直接用 pandas.DataFrame.append 或 to_sql 写入数据库线性化后每条病例就变成了一条拥有固定字段的平坦记录非常适合存入关系型数据库的表格中或者导出为CSV文件进行后续分析。4. 性能优化与批量处理实战当面对数万甚至百万级的病案数据时单条处理的速度是无法接受的。我们必须引入批处理和并发技术。4.1 利用Pandas进行向量化批处理对于规则匹配这类操作如果能将数据全部放入Pandas DataFrame并利用其向量化操作速度会有质的飞跃。例如为大量诊断编码匹配MDCimport pandas as pd import numpy as np def batch_match_mdc(df_patients, rule_loader): df_patients: DataFrame包含‘main_diagnosis’列 rule_loader: 已加载规则的RuleLoader实例 返回添加了‘mdc’列的DataFrame # 方法一使用apply相对灵活但非完全向量化 # df_patients[mdc] df_patients[main_diagnosis].apply(rule_loader.get_mdc_for_diagnosis) # 方法二构建映射字典然后使用map更高效 # 假设我们已经将规则加载成了一个全面的映射字典 all_mdc_map (包含具体编码和范围) # 这里需要先将范围规则预处理掉生成一个所有可能编码到MDC的完整映射如果范围不大可以这么做 # 以下为简化示例假设我们只有精确匹配 all_mdc_map rule_loader.mdc_map # 这里假设mdc_map已是完整映射 df_patients[mdc] df_patients[main_diagnosis].map(all_mdc_map) # 处理未匹配到的map返回NaN df_patients[mdc] df_patients[mdc].fillna(MDCZZ) # 默认组 return df_patients实操心得在数据量极大时map函数比apply快得多。但前提是映射字典要尽可能完整。对于CHS-DRG这种有范围规则的映射一个折中的办法是先将所有范围规则展开成具体编码列表如果范围可枚举且数量可接受生成一个巨大的映射字典。如果范围不可枚举则需要写一个自定义函数并在apply中结合一些缓存机制来优化。4.2 使用多进程并行处理如果单机内存足够但CPU有多核我们可以将数据分块利用multiprocessing库进行并行分组。from multiprocessing import Pool, cpu_count import pandas as pd def parallel_grouping(data_chunk, rule_loader): 处理一个数据块的函数 grouper DRGGrouper(rule_loader) # 每个进程有自己的grouper实例 results [] for _, row in data_chunk.iterrows(): patient_dict row.to_dict() result grouper.group(patient_dict) results.append(linearize_grouping_result(result)) return pd.DataFrame(results) def batch_group_parallel(df_master, rule_loader, n_jobsNone): 主函数并行批处理分组 if n_jobs is None: n_jobs max(1, cpu_count() - 1) # 留一个核心给系统 # 将大数据框拆分成多个块 chunk_size max(1, len(df_master) // n_jobs) chunks [df_master.iloc[i:i chunk_size] for i in range(0, len(df_master), chunk_size)] # 创建进程池 with Pool(processesn_jobs) as pool: # 使用starmap传递多个参数这里需要将rule_loader也传递进去。 # 注意rule_loader对象需要是可序列化的(picklable)如果包含复杂的自定义对象可能有问题。 # 更稳妥的方式是在每个子进程中重新初始化规则加载器从文件读取。 args [(chunk, rule_loader) for chunk in chunks] result_dfs pool.starmap(parallel_grouping, args) # 合并所有结果 final_df pd.concat(result_dfs, ignore_indexTrue) return final_df # 使用前需要确保rule_loader是可pickle的或者修改parallel_grouping函数在内部重新加载规则。重要提示多进程编程中每个进程都有独立的内存空间。传递rule_loader对象时如果它内部有大量数据如加载的规则表会被序列化并复制到每个进程可能导致内存消耗剧增。一种优化方案是在每个子进程内部重新从文件或共享内存中加载规则。另一种更高级的方案是使用multiprocessing.Manager创建共享字典但会牺牲一些速度。5. 部署、调试与常见问题排查系统开发完成后如何让它稳定可靠地运行起来才是真正的挑战。5.1 系统部署与配置管理一个完整的项目应该包含清晰的配置文件将一切可变的参数外部化。# config.yaml data: input_path: ./data/input/ output_path: ./data/output/ rule_path: ./rules/chs_drg_v1.2/ grouping: default_mdc_for_unmatched: MDCZZ default_drg_for_unmatched: 000 enable_cc_mcc_check: true processing: batch_size: 1000 use_multiprocessing: true n_jobs: 4 logging: level: INFO file: ./logs/drg_grouper.log在代码中使用PyYAML库读取配置。这样当规则路径、默认值或处理参数需要变更时只需修改配置文件无需触动代码。5.2 详尽的日志记录日志是调试和监控的生命线。务必在关键步骤和异常处添加日志。import logging import sys def setup_logging(config): log_level getattr(logging, config[logging][level]) log_file config[logging][file] logging.basicConfig( levellog_level, format%(asctime)s - %(name)s - %(levelname)s - %(message)s, handlers[ logging.FileHandler(log_file, encodingutf-8), logging.StreamHandler(sys.stdout) # 同时输出到控制台 ] ) return logging.getLogger(__name__) # 在分组引擎中使用 logger logging.getLogger(__name__) class DRGGrouper: def group(self, patient_data): patient_id patient_data.get(patient_id, UNKNOWN) logger.info(f开始处理病例: {patient_id}) try: # ... 分组逻辑 ... if not mdc: logger.warning(f病例 {patient_id} 主要诊断 {main_diag} 未找到MDC映射入组失败。) result[drg] 000 result[status] FAILED_MDC return result # ... 更多逻辑 ... logger.info(f病例 {patient_id} 分组成功: DRG{drg}) except Exception as e: logger.error(f处理病例 {patient_id} 时发生未知错误: {e}, exc_infoTrue) result[drg] 000 result[status] ERROR return result5.3 常见问题排查手册在实际运行中你一定会遇到各种各样的问题。下面这个表格整理了我踩过的一些“坑”和解决方法问题现象可能原因排查步骤与解决方案大量病例无法入组DRG为‘000’1. 诊断/操作编码格式与规则表不匹配。2. 规则表未加载或加载错误。3. 主要诊断字段为空或无效。1.检查编码格式对比一个失败病例的诊断编码和规则表中的编码格式是否带点字母大小写。使用df[main_diag].str.strip().unique()查看所有编码变体。2.验证规则加载打印rule_loader.mdc_map的前几项检查是否成功加载。确认规则文件路径正确。3.数据质量检查运行df[main_diagnosis].isnull().sum()统计空值。分组结果与官方工具不一致1. CC/MCC判断逻辑有误。2. 手术操作优先级或主要操作判断错误。3. 年龄、新生儿体重等细分规则未生效。1.CC/MCC复核提取一个结果不一致的病例手动检查其次要诊断是否在CC/MCC列表中。确认你的CC/MCC列表是否完整、版本是否正确。2.手术逻辑检查确认你的“主要手术操作”判定逻辑是否与CHS-DRG规则一致通常是资源消耗最大、技术难度最高的。3.细分规则调试在_determine_drg函数中增加详细日志打印出所有候选DRG和判断条件逐步比对。程序运行速度慢内存占用高1. 单条处理未使用批处理。2. 多进程时规则数据被重复复制。3. Pandas操作未优化产生中间副本。1.向量化尽可能使用Pandas的map、apply和向量化运算避免Python层级的for循环。2.共享内存考虑使用multiprocessing.shared_memory或第三方库如ray、dask来共享只读的规则数据。3.内存管理处理完一个数据块后及时使用del释放变量或使用pandas.read_csv的chunksize参数流式处理超大文件。更新DRG版本后分组结果大面积变化1. 规则文件未正确更新。2. 程序缓存了旧的规则数据。1.版本控制在配置文件和日志中明确记录使用的规则版本号如CHS-DRG v1.2。2.清除缓存如果规则数据被缓存如使用joblib或自定义缓存更新文件后务必清除缓存或重启服务。3.做版本对比测试用同一份测试数据集分别用新旧版本规则运行对比结果差异确认符合预期。5.4 单元测试与集成测试为了保证分组逻辑的准确性必须编写严格的测试。import unittest from your_module import DRGGrouper, RuleLoader class TestDRGGrouper(unittest.TestCase): classmethod def setUpClass(cls): 所有测试开始前执行一次加载规则 cls.rule_loader RuleLoader(./test_rules/) cls.rule_loader.load_all_rules() cls.grouper DRGGrouper(cls.rule_loader) def test_mdc_assignment(self): 测试MDC判断是否正确 test_cases [ ({main_diagnosis: A01.0}, MDCA), # 假设A01.0属于MDCA ({main_diagnosis: I10}, MDCF), # 假设I10属于MDCF ({main_diagnosis: INVALID}, MDCZZ), # 无效编码 ] for input_data, expected_mdc in test_cases: with self.subTest(input_datainput_data): result self.grouper.group(input_data) self.assertEqual(result[mdc], expected_mdc) def test_surgical_adrg(self): 测试有手术操作的病例是否能正确进入外科ADRG patient_with_op { main_diagnosis: K35.0, # 急性阑尾炎 operations: [47.01], # 阑尾切除术 age: 30, secondary_diagnoses: [] } result self.grouper.group(patient_with_op) # 断言结果中的ADRG应该是外科相关的例如‘GD1’ self.assertTrue(result[adrg].startswith(G)) # 假设外科ADRG以G开头 self.assertNotEqual(result[drg], 000) def test_cc_mcc_impact(self): 测试CC/MCC对最终DRG的影响 base_patient {main_diagnosis: I21.0, operations: [], age: 65} patient_without_cc {**base_patient, secondary_diagnoses: []} patient_with_cc {**base_patient, secondary_diagnoses: [E11.9]} # 糖尿病 result_no_cc self.grouper.group(patient_without_cc) result_with_cc self.grouper.group(patient_with_cc) # 同一个ADRG下有CC的DRG应该与无CC的DRG不同 self.assertEqual(result_no_cc[adrg], result_with_cc[adrg]) self.assertNotEqual(result_no_cc[drg], result_with_cc[drg]) if __name__ __main__: unittest.main()编写覆盖主要分组路径、边界条件如新生儿、极高龄患者和异常情况如编码缺失的测试用例是确保代码长期稳定运行的最有效手段。每次规则更新后都应跑一遍完整的测试集。这个Python实现的CHS-DRG分组器从设计到实现再到优化和排错是一个典型的将复杂业务规则转化为自动化系统的过程。它的核心价值在于将一致性、效率与可维护性结合了起来。人工分组难免会有疏漏和疲劳导致的错误而系统只要规则配置正确每一次计算都是严格一致的。处理速度从“天”级别提升到“分钟”甚至“秒”级别解放了人力。最后通过模块化设计和配置化管理系统能够相对平滑地适应CHS-DRG规则的年度更新。当然真实世界的挑战远不止于此比如诊断编码的归一化、手术操作链的复杂判断等但有了这个坚实的基础框架后续的扩展和深化就有了明确的路径。本文还有配套的精品资源点击获取