简介本资源是一个基于Python开发的CHS-DRG分组器实现系统面向医疗机构信息科人员、医保结算工程师及医疗大数据分析学习者解决DRG分组过程中MDC映射、ADRG判定规则解析与MCC/CC排除逻辑结构化提取等核心问题适用于医保支付改革背景下的分组算法验证与本地化适配场景。压缩包共2000个文件含886个可读可调试的Python源模块覆盖数据加载、规则解析、线性化输出全流程、624个编译字节码文件、343个备份配置zbak、125个文本格式规则说明与映射表以及dat二进制数据集和SQLite数据库各1个整体体积7.32MB工程结构完整、模块职责清晰。目前已有118人学习下载读者可直接复用全部886个.py脚本进行CHS-DRG标准数据解析获取标准化文本输出结果并基于ZD_INFO、SS_VALID、CCE等关键数据文件深入理解诊断与手术编码的匹配逻辑与分组权重生成机制。1. 项目概述从DRG到代码的桥梁最近在做一个挺有意思的项目核心就是用Python把CHS-DRG的分组逻辑给“翻译”出来做成一个能自动处理数据、跑通分组流程的系统。听起来可能有点专业但说白了就是医院里给病人看病、做手术医保怎么根据病情的复杂程度和资源消耗来付费背后就靠DRG这套规则。而CHS-DRG国家医疗保障疾病诊断相关分组就是咱们国内的标准。手动去对规则、算权重、分组工作量巨大还容易出错所以用代码把它自动化就成了一个刚需。这个“DRG分组器”项目目标很明确第一你得能从一堆乱七八糟的病案首页数据里把分组需要的关键信息比如主要诊断、手术操作、年龄等精准地“挖”出来这就是“数据提取”。第二DRG的分组规则本身是树状或者有复杂依赖关系的你得把它转换成计算机能高效处理的一条条“线性”判断逻辑这就是“线性化处理”。最后用Python把这些逻辑串起来形成一个输入病案数据、输出DRG分组结果的黑盒子。这活儿干好了无论是给医院做内部成本分析还是给医保部门做支付模拟甚至给相关软件公司做核心引擎都很有价值。我自己在啃这块硬骨头的时候踩了不少坑也总结了一些门道今天就跟大伙儿详细聊聊怎么用Python把这套系统给搭起来。2. 核心思路与架构设计2.1 理解CHS-DRG分组的基本逻辑在动手写代码之前必须得先吃透CHS-DRG分组的“游戏规则”。它不是简单的一对一匹配而是一个多步骤的决策树。简单来说流程是这样的病例数据准备拿到一份完整的病案首页数据包含患者基本信息、诊断信息ICD-10编码、手术操作信息ICD-9-CM-3编码等。主要诊断大类MDC判断根据病人的“主要诊断”编码先确定他属于26个MDC中的哪一个。比如主要诊断是“急性心肌梗死”那它就归入“MDC05 循环系统疾病及功能障碍”。外科手术判断与ADRG预分组在确定的MDC下看病人有没有做手术、做了什么手术。CHS-DRG分组器会先根据手术操作代码判断是否进入外科部分。然后结合主要诊断和主要手术将病例分入某一个“核心疾病诊断相关组”ADRG。一个ADRG包含多个临床过程相似、资源消耗相近的病例。内科驱动分组如果病例没有手术或不符合外科分组条件则根据主要诊断考虑并发症与合并症CC及严重并发症与合并症MCC的影响将其分入相应的内科ADRG。细分组DRG在ADRG的基础上再根据病人的年龄、体重新生儿、其他诊断并发症/合并症、出院情况等因素进一步细分为最终的DRG。这一步决定了医保支付的权重。我们的Python程序就是要模拟这个完整的决策过程。难点在于CHS-DRG的分组规则手册是一本厚厚的、充满“如果...那么...”语句的文档我们需要把它转化成无歧义的、可执行的逻辑。2.2 系统架构设计模块化与流程化基于上述逻辑我把整个系统设计成几个松耦合的模块这样便于开发、测试和维护。整体架构如下图所示用文字描述数据输入层负责接收原始数据。原始数据可能来自Excel、CSV、数据库或者医院信息系统的接口。这一层要做的是数据校验和初步清洗比如确保诊断和手术编码的格式符合ICD标准。核心处理引擎这是系统的心脏包含两个核心子模块。数据提取与标准化模块它的任务是从清洗后的数据中精准定位并提取分组所需的所有字段。例如从诊断列表里找出“主要诊断”从手术列表里找出“主要手术”并判断是否存在其他诊断用作CC/MCC判断。这里的一个关键点是编码的标准化比如是否带了小数点是否使用了扩展码都需要统一处理。规则线性化与执行模块这是技术核心。我们需要将CHS-DRG分组规则通常是PDF或Word文档进行解析。我的做法是先将规则的关键部分如MDC与诊断映射表、手术分组表、CC/MCC列表、年龄/体重分组节点结构化存入数据库如SQLite或配置文件如JSON/YAML。然后编写一个规则解释器按照“MDC判断 → 外科/内科路径选择 → ADRG查找 → DRG细分”的顺序线性地执行一系列查询和条件判断。决策与输出层引擎执行完毕后生成分组结果。结果应包括MDC代码、ADRG代码、DRG代码、DRG名称、权重值以及关键的分组路径日志例如为什么被分到某个外科ADRG依据了哪条手术码。输出可以是写入数据库、生成报告文件或通过API返回。辅助工具层包括日志记录方便追踪分组过程、排查问题、缓存机制对于频繁查询的规则表如CC列表进行缓存以提升性能、以及一个简单的管理界面用于更新规则库因为CHS-DRG版本会定期更新。注意规则库的维护是持续性的工作。CHS-DRG每年都可能更新因此规则解析和导入工具的设计要考虑到可维护性最好能做到半自动化更新。3. 关键技术实现细节3.1 数据提取从混沌到精准病案首页数据往往结构复杂字段多且不同来源的数据质量参差不齐。数据提取模块的鲁棒性直接决定了分组结果的准确性。字段映射与识别首先要定义一个清晰的输入数据规范。即使源数据字段名五花八门如“主要诊断”可能叫“PRIMARY_DIAG”、“ZD1”、“MAIN_DX”我们也要在程序内部建立一个映射字典将其统一到内部标准字段名。我通常会用一个配置字典来处理这种映射关系。# 示例字段映射配置 FIELD_MAPPING { ‘内部字段名_主要诊断‘: [‘PRIMARY_DIAG‘, ‘ZD1‘, ‘第一诊断‘, ‘main_diagnosis‘], ‘内部字段名_手术操作1‘: [‘OPERATION1‘, ‘SS1‘, ‘主要手术‘], ‘内部字段名_年龄‘: [‘AGE‘, ‘NL‘, ‘患者年龄‘], ‘内部字段名_出院诊断‘: [‘DISCHARGE_DIAGS‘], # 可能是一个用分隔符拼接的字符串 # ... 其他字段 } def standardize_input(raw_data_dict, mapping_config): 标准化输入数据字典 standardized {} for internal_field, possible_keys in mapping_config.items(): value None for key in possible_keys: if key in raw_data_dict: value raw_data_dict.get(key) break standardized[internal_field] value return standardized诊断与手术列表的处理这是一个重难点。一份病案通常有多个诊断和多个手术。我们需要提取主要诊断/手术通常源数据会标明哪个是“主要”。如果没有则需要根据业务逻辑推断如第一个诊断。识别其他诊断ODX所有非主要诊断的列表用于后续CC/MCC判断。需要处理字符串拼接如“I10;E11.9;J45.901”或列表形式的数据。编码清洗去除空格统一小数点格式。ICD-10编码通常是类似“I10”、“E11.9”的格式要确保其规范。import re def parse_diagnosis_codes(diag_string): 解析诊断字符串返回编码列表 if not diag_string: return [] # 假设用分号、逗号或空格分隔 split_pattern r‘[;,\s]‘ raw_codes re.split(split_pattern, diag_string.strip()) cleaned_codes [] for code in raw_codes: if code: # 基础清洗去空格转大写 clean_code code.strip().upper() # 可选简单格式校验此处为示例实际规则更复杂 if re.match(r‘^[A-Z][0-9][0-9A-Z](\.[0-9A-Z]{1,4})?$‘, clean_code): cleaned_codes.append(clean_code) else: # 记录日志无效编码 logging.warning(f“Invalid diagnosis code format: {code}“) return cleaned_codes # 示例分离主要诊断和其他诊断 std_data {‘主要诊断‘: ‘I10‘, ‘其他诊断‘: ‘E11.9;J45.901‘} primary_dx std_data[‘主要诊断‘] other_dx_list parse_diagnosis_codes(std_data.get(‘其他诊断‘, ‘‘))实操心得数据提取阶段最怕遇到编码错误或遗漏。一定要加入大量的日志记录和异常处理。对于无法识别的字段或明显不合规的编码如长度异常、包含非法字符要记录警告并采用默认值或抛出错误方便后期数据溯源和修正。不要试图在程序里“智能猜测”这会导致分组结果不可控。3.2 规则线性化将手册变成代码逻辑这是整个项目最核心、最烧脑的部分。CHS-DRG分组规则本质上是复杂的条件判断集合。线性化的目标是将其转化为一系列顺序执行的“if-elif-else”语句或者更优的转化为基于数据表的查询。1. MDC映射表的建立CHS-DRG有26个MDC每个MDC对应一系列诊断编码范围。手册里通常是一个表格。我们需要将其数字化。最实用的方法是建立一个“MDC查找表”。诊断编码前缀/范围MDC代码MDC名称A00-B99MDC01传染病和寄生虫病C00-D48MDC02肿瘤I00-I99MDC05循环系统疾病及功能障碍.........在程序中可以加载这个表比如从CSV或数据库然后写一个函数根据主要诊断编码的前几位去匹配。def find_mdc_for_diagnosis(dx_code, mdc_mapping_table): 根据诊断编码查找MDC if not dx_code: return None # 可能需要尝试不同长度的前缀进行匹配 for length in range(len(dx_code), 0, -1): prefix dx_code[:length] matched_mdc mdc_mapping_table.get(prefix) if matched_mdc: return matched_mdc # 如果都没匹配上可能属于“先期分组疾病”或其他特殊情况需要额外处理 return ‘MDC00‘ # 或一个特殊的‘未找到‘标识2. 外科分组与内科分组的逻辑分离在确定MDC后需要判断病例走外科路径还是内科路径。这依赖于“主要手术”操作码。CHS-DRG有一个“手术操作分类表”定义了哪些操作码属于外科手术并且这些手术码会映射到特定的ADRG。我的策略是建立两个关键表外科手术-ADRG映射表字段包括手术编码、所属MDC、目标ADRG。内科诊断-ADRG映射表字段包括诊断编码、所属MDC、目标ADRG有时还关联CC/MCC的影响。程序逻辑如下def determine_adrg(mdc, primary_dx, primary_op, other_dx_list, surgery_mapping, medical_mapping): 确定ADRG adrg None path ‘‘ # 1. 外科路径优先判断 if primary_op and primary_op in surgery_mapping: # 检查该手术是否属于当前MDC op_info surgery_mapping[primary_op] if op_info[‘mdc‘] mdc: adrg op_info[‘adrg‘] path ‘外科‘ return adrg, path # 2. 内科路径 if primary_dx in medical_mapping: dx_info medical_mapping[primary_dx] if dx_info[‘mdc‘] mdc: adrg dx_info[‘base_adrg‘] # 基础ADRG path ‘内科‘ # 内科ADRG可能还需要结合CC/MCC进行细分这里先返回基础ADRG return adrg, path # 3. 未找到可能是歧义病例或数据错误 return None, ‘未识别‘3. CC/MCC列表的集成与判断并发症和合并症列表是另一张大表。它定义了哪些诊断编码属于CC并发症与合并症哪些属于MCC严重并发症与合并症。在确定了基础ADRG后需要遍历病人的“其他诊断列表”检查是否有编码出现在当前ADRG对应的CC或MCC列表中。根据CC/MCC的存在情况以及患者的年龄最终决定细分到哪一个具体的DRG。这部分需要高效的查找。我会将CC/MCC列表加载到内存中的集合set或字典里实现O(1)时间复杂度的查找。# 假设已加载CC/MCC列表结构如{‘ADRG001‘: {‘mcc‘: {‘E10.10‘, ‘I21.9‘, ...}, ‘cc‘: {‘E11.9‘, ‘I10‘, ...}}} def evaluate_cc_mcc(adrg_code, other_dx_list, cc_mcc_library): 评估CC/MCC情况 if adrg_code not in cc_mcc_library: return ‘无CC/MCC‘, [] lib cc_mcc_library[adrg_code] mcc_set lib.get(‘mcc‘, set()) cc_set lib.get(‘cc‘, set()) found_mcc [] found_cc [] for dx in other_dx_list: if dx in mcc_set: found_mcc.append(dx) elif dx in cc_set: found_cc.append(dx) if found_mcc: return ‘有MCC‘, found_mcc elif found_cc: return ‘有CC‘, found_cc else: return ‘无CC/MCC‘, []4. 最终DRG决策结合基础ADRG、CC/MCC评估结果、患者年龄有时还有新生儿体重、出院情况等查询最终的“DRG细分规则表”得到唯一的DRG代码和权重。注意规则线性化不是一次性工作。CHS-DRG规则中充满了例外情况、排除条款和复杂依赖。在编码时必须反复对照官方手册为每一条逻辑分支编写详细的注释并创建大量的测试用例进行验证。建议使用单元测试框架如pytest为每一个MDC、每一个典型的ADRG都编写测试案例。4. 系统实现与核心代码解析4.1 项目结构与依赖一个清晰的项目结构能让协作和维护变得轻松。我的项目目录通常如下drg_grouper/ ├── config/ # 配置文件 │ ├── field_mapping.yaml # 字段映射配置 │ └── logging.conf # 日志配置 ├── data/ # 规则数据文件 (CSV, JSON) │ ├── mdc_mapping.csv │ ├── surgery_adrg_mapping.csv │ ├── medical_adrg_mapping.csv │ └── cc_mcc_library.json ├── core/ # 核心引擎 │ ├── __init__.py │ ├── data_extractor.py # 数据提取模块 │ ├── rule_engine.py # 规则执行引擎 │ └── models.py # 数据模型如Patient, DRGResult ├── utils/ # 工具函数 │ ├── file_loader.py # 加载CSV/JSON │ └── logger.py # 日志工具 ├── tests/ # 单元测试 │ ├── test_extractor.py │ └── test_engine.py ├── main.py # 主程序入口 └── requirements.txt # Python依赖主要Python依赖库pandas: 用于高效加载和处理规则表格数据虽然核心逻辑不用但数据准备阶段非常方便。pyyaml: 解析YAML格式的配置文件。openpyxl/xlrd: 如果源数据是Excel需要它们来读取。pytest: 用于编写和运行单元测试保证代码质量。logging: Python标准库用于记录程序运行日志必不可少。requirements.txt示例pandas1.4.0 PyYAML6.0 openpyxl3.0.0 pytest7.0.04.2 核心引擎类设计与实现我习惯用一个DRGGrouper类来封装整个分组逻辑这样使用起来比较面向对象状态清晰。# core/models.py from dataclasses import dataclass from typing import List, Optional dataclass class PatientCase: 病案数据模型 case_id: str primary_diagnosis: Optional[str] None other_diagnoses: List[str] None primary_operation: Optional[str] None other_operations: List[str] None age: Optional[int] None # ... 其他字段如性别、出生体重、出院情况等 def __post_init__(self): if self.other_diagnoses is None: self.other_diagnoses [] if self.other_operations is None: self.other_operations [] dataclass class DRGResult: 分组结果模型 case_id: str mdc_code: Optional[str] mdc_name: Optional[str] adrg_code: Optional[str] adrg_name: Optional[str] drg_code: Optional[str] drg_name: Optional[str] drg_weight: Optional[float] path_log: List[str] # 记录分组路径用于调试 cc_mcc_status: Optional[str] matched_cc_mcc_codes: List[str]# core/rule_engine.py import logging from .models import PatientCase, DRGResult class DRGEngine: DRG分组引擎核心类 def __init__(self, rule_loader): 初始化引擎 :param rule_loader: 规则加载器对象负责提供各种映射表 self.rule_loader rule_loader self.mdc_map rule_loader.load_mdc_mapping() self.surgery_map rule_loader.load_surgery_mapping() self.medical_map rule_loader.load_medical_mapping() self.cc_mcc_lib rule_loader.load_cc_mcc_library() self.drg_detail_map rule_loader.load_drg_detail_mapping() self.logger logging.getLogger(__name__) def group(self, patient_case: PatientCase) - DRGResult: 对单个病案进行分组 result DRGResult(case_idpatient_case.case_id, path_log[]) self.logger.info(f“开始分组病例: {patient_case.case_id}“) # 步骤1: 确定MDC mdc_info self._get_mdc(patient_case.primary_diagnosis) result.mdc_code mdc_info[‘code‘] result.mdc_name mdc_info[‘name‘] result.path_log.append(f“主要诊断‘{patient_case.primary_diagnosis}‘ 归入 {result.mdc_code}“) if not result.mdc_code: result.path_log.append(“错误无法确定MDC分组终止。“) return result # 步骤2: 确定ADRG (外科优先) adrg_info, path_type self._get_adrg( result.mdc_code, patient_case.primary_diagnosis, patient_case.primary_operation, patient_case.other_diagnoses ) result.adrg_code adrg_info.get(‘code‘) result.adrg_name adrg_info.get(‘name‘) result.path_log.append(f“通过{path_type}路径确定ADRG: {result.adrg_code}“) if not result.adrg_code: result.path_log.append(“错误无法确定ADRG分组终止。“) return result # 步骤3: 评估CC/MCC cc_mcc_status, matched_codes self._evaluate_cc_mcc( result.adrg_code, patient_case.other_diagnoses ) result.cc_mcc_status cc_mcc_status result.matched_cc_mcc_codes matched_codes result.path_log.append(f“CC/MCC状态: {cc_mcc_status}, 匹配编码: {matched_codes}“) # 步骤4: 确定最终DRG drg_info self._get_final_drg( result.adrg_code, cc_mcc_status, patient_case.age, # 其他因素如新生儿体重、出院情况等 ) result.drg_code drg_info.get(‘code‘) result.drg_name drg_info.get(‘name‘) result.drg_weight drg_info.get(‘weight‘) result.path_log.append(f“确定最终DRG: {result.drg_code}, 权重: {result.drg_weight}“) self.logger.info(f“病例 {patient_case.case_id} 分组完成DRG: {result.drg_code}“) return result def _get_mdc(self, primary_dx): # 实现具体的MDC查找逻辑见前文示例 pass def _get_adrg(self, mdc, primary_dx, primary_op, other_dx): # 实现ADRG判断逻辑见前文示例 pass def _evaluate_cc_mcc(self, adrg_code, other_dx_list): # 实现CC/MCC评估逻辑见前文示例 pass def _get_final_drg(self, adrg_code, cc_mcc_status, age, ...): # 实现最终DRG查询逻辑 # 根据adrg_code, cc_mcc_status, age等组合成key查询drg_detail_map key f“{adrg_code}|{cc_mcc_status}|{age_group}“ return self.drg_detail_map.get(key, {})4.3 规则加载器的实现规则加载器负责从文件CSV/JSON或数据库中将规则数据读入内存并组织成方便快速查找的数据结构如字典、集合。# utils/rule_loader.py import csv import json from pathlib import Path class RuleLoader: def __init__(self, data_dir): self.data_dir Path(data_dir) def load_mdc_mapping(self): mapping {} with open(self.data_dir / ‘mdc_mapping.csv‘, ‘r‘, encoding‘utf-8-sig‘) as f: reader csv.DictReader(f) for row in reader: # 假设CSV有‘dx_prefix‘, ‘mdc_code‘, ‘mdc_name‘列 mapping[row[‘dx_prefix‘]] {‘code‘: row[‘mdc_code‘], ‘name‘: row[‘mdc_name‘]} return mapping def load_surgery_mapping(self): mapping {} with open(self.data_dir / ‘surgery_adrg_mapping.csv‘, ‘r‘, encoding‘utf-8-sig‘) as f: reader csv.DictReader(f) for row in reader: # 假设CSV有‘op_code‘, ‘mdc‘, ‘adrg_code‘, ‘adrg_name‘列 mapping[row[‘op_code‘]] { ‘mdc‘: row[‘mdc‘], ‘adrg‘: {‘code‘: row[‘adrg_code‘], ‘name‘: row[‘adrg_name‘]} } return mapping def load_cc_mcc_library(self): # 从JSON加载结构更灵活 with open(self.data_dir / ‘cc_mcc_library.json‘, ‘r‘, encoding‘utf-8‘) as f: library json.load(f) # 将列表转换为集合提高查找效率 for adrg_info in library.values(): if ‘mcc‘ in adrg_info: adrg_info[‘mcc‘] set(adrg_info[‘mcc‘]) if ‘cc‘ in adrg_info: adrg_info[‘cc‘] set(adrg_info[‘cc‘]) return library5. 性能优化与实战技巧5.1 处理大规模数据批量与并发当需要对成千上万份病案进行分组时性能就成为关键。单线程顺序处理会非常慢。这里有几个优化策略使用Pandas进行向量化操作如果逻辑允许对于数据提取和清洗阶段Pandas的向量化操作远比循环快。例如清洗一列诊断编码import pandas as pd def clean_diagnosis_column(df, col_name): 清洗DataFrame中的诊断编码列 # 去除空格转大写 df[col_name] df[col_name].astype(str).str.strip().str.upper() # 使用正则表达式过滤有效格式简化版 pattern r‘^[A-Z][0-9][0-9A-Z](\.[0-9A-Z]{1,4})?$‘ # 将不符合格式的置为NaN df[col_name] df[col_name].where(df[col_name].str.match(pattern), pd.NA) return df核心分组逻辑的并发处理分组引擎本身通常涉及复杂的查表逻辑难以向量化。这时可以使用Python的concurrent.futures模块进行多进程并行。from concurrent.futures import ProcessPoolExecutor, as_completed def batch_group_cases(case_list, engine, max_workers4): 批量分组病例使用多进程 results [] # 注意在Windows上多进程要求if __name__ ‘__main__‘且engine对象可能需要序列化或重新初始化 with ProcessPoolExecutor(max_workersmax_workers) as executor: # 提交任务 future_to_case {executor.submit(engine.group, case): case for case in case_list} # 获取结果 for future in as_completed(future_to_case): case future_to_case[future] try: result future.result() results.append(result) except Exception as exc: logging.error(f“病例 {case.case_id} 分组失败: {exc}“) # 可以返回一个包含错误信息的结果对象 results.append(DRGResult(case_idcase.case_id, path_log[f“分组错误: {exc}“])) return results注意多进程并行时每个进程都会复制一份规则数据到内存中。如果规则库很大几百MB可能会消耗大量内存。可以考虑使用只读的共享内存或者使用像Redis这样的外部缓存来存储规则表所有工作进程从缓存读取。5.2 缓存机制的应用分组过程中有些查询是重复且昂贵的比如根据诊断前缀查找MDC。虽然我们已经用了字典O(1)复杂度但每次分组都要对主要诊断执行循环匹配。我们可以引入一个简单的缓存来存储最近查询过的诊断-MDC对。from functools import lru_cache class CachedDRGEngine(DRGEngine): def __init__(self, rule_loader): super().__init__(rule_loader) # 使用LRU缓存最多缓存1024个诊断的MDC结果 self._get_mdc_cached lru_cache(maxsize1024)(self._get_mdc_impl) def _get_mdc(self, primary_dx): # 对外接口内部调用缓存版本 if not primary_dx: return {‘code‘: None, ‘name‘: None} return self._get_mdc_cached(primary_dx) def _get_mdc_impl(self, primary_dx): # 实际的MDC查找逻辑与之前相同 for length in range(len(primary_dx), 0, -1): prefix primary_dx[:length] if prefix in self.mdc_map: return self.mdc_map[prefix] return {‘code‘: None, ‘name‘: None}对于CC/MCC查找由于是集合操作本身很快通常不需要缓存。缓存策略需要根据实际性能分析来定避免过度优化。5.3 日志与监控让系统可观测一个健壮的生产系统必须有完善的日志。日志不仅能帮助调试还能用于监控分组结果的分布和性能瓶颈。import logging import sys def setup_logging(log_levellogging.INFO, log_file‘drg_grouper.log‘): 配置日志 logger logging.getLogger(‘drg_grouper‘) logger.setLevel(log_level) # 避免重复添加handler if logger.handlers: return logger # 控制台Handler console_handler logging.StreamHandler(sys.stdout) console_format logging.Formatter(‘%(asctime)s - %(name)s - %(levelname)s - %(message)s‘) console_handler.setFormatter(console_format) logger.addHandler(console_handler) # 文件Handler file_handler logging.FileHandler(log_file, encoding‘utf-8‘) file_format logging.Formatter(‘%(asctime)s - %(name)s - %(levelname)s - %(module)s:%(lineno)d - %(message)s‘) file_handler.setFormatter(file_format) logger.addHandler(file_handler) return logger # 在引擎中使用 self.logger setup_logging() self.logger.debug(f“正在处理病例: {case_id}“) # 详细流程 self.logger.info(f“病例 {case_id} 分组结果为: {drg_code}“) # 关键信息 self.logger.warning(f“病例 {case_id} 主要诊断编码‘{dx}‘格式可疑“) # 警告 self.logger.error(f“病例 {case_id} 分组过程中发生异常“, exc_infoTrue) # 错误实操心得日志级别要合理运用。开发调试时用DEBUG生产环境用INFO或WARNING。对于每一条分组结果至少记录case_id和最终的DRG代码这样一旦后期发现分组有误可以根据日志追溯当时的分组路径和输入数据。6. 常见问题排查与调试技巧在实际运行中你肯定会遇到分组结果与预期不符的情况。这时候系统的可调试性就至关重要。6.1 建立测试用例库这是最有效的方法。从官方分组手册、历史数据或与业务专家讨论中收集一批“标准答案”已知的病案。为每个病案编写一个测试用例包括输入数据和期望的输出DRG。# tests/test_engine.py import pytest from core.models import PatientCase from core.rule_engine import DRGEngine from utils.rule_loader import RuleLoader pytest.fixture def engine(): loader RuleLoader(‘./data‘) return DRGEngine(loader) def test_mi_with_cc(engine): 测试急性心梗伴有并发症的病例 case PatientCase( case_id“TEST001“, primary_diagnosis“I21.9“, # 急性心肌梗死 other_diagnoses[“E11.9“], # 2型糖尿病 age65, primary_operationNone ) result engine.group(case) # 根据CHS-DRG规则期望的DRG可能是‘FM11‘ (伴CC/MCC的心肌梗死) assert result.drg_code “FM11“ assert result.mdc_code “MDC05“ # 可以增加更多断言检查路径日志等 print(result.path_log)使用pytest运行测试可以快速定位是规则数据错误还是引擎逻辑有bug。6.2 分组路径追踪与可视化在DRGResult中我们设计了path_log字段。在调试时把这个日志打印出来就能清晰地看到分组决策的每一步。病例 TEST001 分组路径: - 主要诊断‘I21.9‘ 归入 MDC05 - 通过内科路径确定ADRG: F21 (急性心肌梗死) - CC/MCC状态: 有CC, 匹配编码: [‘E11.9‘] - 确定最终DRG: FM11, 权重: 1.23如果结果不对就顺着日志一步步检查MDC对吗诊断编码清洗是否正确走的路径对吗外科还是内科主要手术码识别对了吗ADRG对吗诊断/手术码与映射表匹配吗CC/MCC判断对吗其他诊断编码都正确提取了吗CC列表里包含这个编码吗最终DRG的查询keyADRG状态年龄拼对了吗6.3 数据质量检查清单很多分组错误源于输入数据质量差。在数据提取模块后可以增加一个数据质量检查环节def validate_case_data(patient_case): 数据质量校验 warnings [] if not patient_case.primary_diagnosis: warnings.append(“主要诊断为空“) else: if not is_valid_icd10(patient_case.primary_diagnosis): warnings.append(f“主要诊断编码‘{patient_case.primary_diagnosis}‘格式无效“) if patient_case.age is not None and (patient_case.age 0 or patient_case.age 150): warnings.append(f“年龄‘{patient_case.age}‘超出合理范围“) # 检查诊断编码是否重复 all_dx [patient_case.primary_diagnosis] patient_case.other_diagnoses if len(all_dx) ! len(set(all_dx)): warnings.append(“诊断列表中存在重复编码“) return warnings在正式分组前运行此检查将警告信息记录到结果或日志中能极大帮助后续的问题定位。6.4 规则版本管理CHS-DRG规则每年都可能更新。必须建立严格的规则版本管理。将不同版本的规则数据CSV/JSON存放在以版本号命名的目录下如data/v1.0/,data/v1.1/。在程序配置中指定当前使用的规则版本。当切换版本时必须用完整的测试用例库进行回归测试确保新旧版本下已知病例的分组结果要么一致要么变化符合新版规则预期。踩坑实录有一次更新规则库后发现大量病例的DRG权重变了。排查了半天发现不是分组逻辑问题而是新版规则的权重表文件drg_weight.csv中DRG代码的列名从drg改成了drg_code导致程序读取权重时全部为None。教训是对输入文件的格式列名、分隔符、编码要做严格的校验和兼容性处理。7. 扩展方向与应用场景一个基础的DRG分组器跑通后可以考虑很多扩展方向让它从一个工具变成一个平台。1. 规则可视化与编辑器对于医保政策研究员或临床专家直接看代码或CSV是不友好的。可以开发一个Web界面以树状图或流程图的形式展示MDC-ADRG-DRG的分组逻辑并允许他们在线对规则进行微调、注释和模拟测试。2. 分组预演与成本模拟医院在收治复杂病人前可以输入其预估的诊断和手术系统模拟出可能进入的DRG组及支付标准帮助临床科室进行成本预估和诊疗方案优化。3. 大数据分析与标杆比对积累大量分组结果后可以进行分析。例如分析本院各科室、各病种的DRG组分布、平均权重、费用结构并与区域或全国的平均水平进行比对发现自身优势病种或潜在的成本控制点。4. 与医院信息系统HIS集成通过API或中间库的方式与HIS系统对接实现病案首页数据的自动抓取和分组结果的回写形成闭环用于实时监控和医保结算预审。5. 机器学习辅助优化虽然分组规则是确定的但机器学习可以用于上游的数据质量提升。例如训练模型自动校验诊断和手术编码的合理性或者基于历史数据预测某份病案进入高权重DRG的概率并提示医生补充关键诊断或病历描述。实现这些扩展技术栈可能就需要从前端的Vue/React到后端的FastAPI/Django再到数据分析的Pandas/Scikit-learn。但核心永远是这个稳定、准确、高效的Python DRG分组引擎。把它打磨好是这一切的基础。本文还有配套的精品资源点击获取