企业官网建设流程全解析

简介：这份资源是一套基于Python实现的CHS-DRG分组器，面向医疗机构信息科人员、医保数据分析师及医疗信息化方向的学习者，用于解决主要诊断类别（MDC）对应关系、基础诊断相关组（ADRG）判定规则以及严重并发症与合并症（MCC/CC）排除条件的结构化提取与标准化输出问题。压缩包共约2000个文件、7.32MB，其中886个py为核心程序模块，624个pyc为编译字节码，343个zbak为备份文件，另有125个txt文档、18个dat数据集及1个db数据库，并附带版本管理配置与授权协议，目录结构完整，便于按模块阅读与二次开发。目前已有121人学习下载。读者可借此理解DRG分组运算的工程实现思路，掌握诊断与手术信息的映射处理、MCC/CC排除逻辑的代码组织方式，并获得可直接运行调试的脚本与数据样例，适合作为医疗质量分析与分组算法研究的实践参考。

1. 从一份病案首页到 DRG 分组：为什么 Python 是绕不开的那把刀

病案首页填完、编码员点下保存的那一刻，数据其实还是"生"的——诊断和手术是一串带扩展位的 ICD 码，年龄是出生日期，费用是一堆明细。真正决定这家医院能不能拿到合理医保支付的，是后面那一步：把首页数据喂进 DRG 分组器，吐出一个四位数的 DRG 组别和对应的权重。我见过太多团队卡在这一步，不是分组逻辑不会，而是数据提取和线性化处理没做干净，导致分组器要么报错，要么分出来的组"看着对、细看全是坑"。

CHS-DRG 的分组不是一次查表，而是"先分 MDC、再分 ADRG、最后进 DRG"的三级漏斗，每一级都依赖前一级清洗后的结构化输入。Python 在这里的价值不是"能写脚本"，而是能把 ICD 码的线性化、主诊断/主手术的判定、CC/MCC 排除表匹配这几件脏活串成一条可复现的流水线。这篇写给正在做医保结算清单或 DRG 分组工具的开发者，从数据提取讲到线性化处理，再到分组器骨架，每一步都给能跑的代码和参数说明。

2. CHS-DRG 分组器的数据底座：从病案首页提取什么、怎么提

2.1 先搞清楚分组器到底吃哪几列数据

很多人一上来就写分组逻辑，结果发现字段对不上。CHS-DRG 分组器的输入可以粗暴归成四类：患者基本信息（性别、年龄、新生儿体重、离院方式）、诊断信息（主诊断 + 其他诊断，带 ICD-10 码）、手术操作信息（主手术 + 其他手术，带 ICD-9-CM-3 码）、费用与资源信息（总费用、住院天数）。其中年龄不是直接给的，得从出生日期和入院日期算；新生儿体重只在特定 MDC 里参与判断，缺了会直接掉进错误组。

我一般会先定义一个提取配置，把"哪些列映射到分组器的哪个字段"显式写出来，而不是散落在代码各处。这样换一家医院的首页模板，只改配置不改逻辑。

# config.py # 病案首页列名 -> 分组器标准字段的映射配置 # 不同医院的首页导出列名差异极大，用配置隔离 FIELD_MAPPING = { "patient_id": "病案号", "gender": "性别", # 1=男 2=女 "birth_date": "出生日期", "admit_date": "入院日期", "discharge_date": "出院日期", "age": "年龄", # 若首页直接给年龄则用，否则由日期算 "neonate_weight": "新生儿出生体重", # 单位：克 "discharge_mode": "离院方式", # 1=医嘱离院 2=医嘱转院 ... "total_cost": "总费用", "main_diag": "主要诊断编码", "other_diags": "其他诊断编码", # 多值，分隔符需统一 "main_proc": "主要手术操作编码", "other_procs": "其他手术操作编码", } # 多值字段的分隔符，各院导出可能是逗号、分号或竖线 MULTI_VALUE_SEP = ","

这段配置的核心作用是解耦。参数上要注意两点：一是gender的编码各院可能是 1/2 也可能是 M/F，映射后要统一；二是多值字段的分隔符必须确认，我踩过用逗号切结果 ICD 码里本身带逗号的坑，所以后面还要做一次格式校验。

2.2 用 pandas 做提取与字段标准化

提取阶段最容易被忽略的是"空值语义"。首页里一个空的其他诊断，可能是"确实没有"，也可能是"没填"，这两者对分组的影响完全不同——前者不影响，后者可能导致 CC/MCC 漏判。我的做法是提取时保留原始空值，标准化阶段再统一成空列表，并在日志里记录每份病案的缺失字段数。

import pandas as pd from config import FIELD_MAPPING, MULTI_VALUE_SEP def extract_homepage(raw_df: pd.DataFrame) -> pd.DataFrame: """从原始首页 DataFrame 提取分组器所需字段并标准化""" df = pd.DataFrame() for std_field, src_col in FIELD_MAPPING.items(): if src_col not in raw_df.columns: # 缺列不静默跳过，记录后补空，避免后续 KeyError print(f"[WARN] 缺少列: {src_col} -> {std_field}") df[std_field] = None else: df[std_field] = raw_df[src_col] # 多值字段统一切成列表，空值 -> 空列表 for col in ["other_diags", "other_procs"]: df[col] = df[col].apply( lambda x: [] if pd.isna(x) or str(x).strip() == "" else [c.strip() for c in str(x).split(MULTI_VALUE_SEP) if c.strip()] ) # 数值字段强制转型，非法值置 NaN 并计数 for col in ["total_cost", "neonate_weight"]: df[col] = pd.to_numeric(df[col], errors="coerce") return df

逻辑说明：先按配置逐列搬运，缺列时打印警告而不是抛异常，保证批量处理不中断；多值字段统一成 Python 列表，后续线性化处理直接迭代；数值字段用errors="coerce"把脏值转成 NaN，方便统计。参数上，MULTI_VALUE_SEP如果某院用的是分号，改配置即可，不用动函数。

2.3 年龄与住院天数的派生计算

年龄在 DRG 分组里是硬门槛，比如某些 MDC 对年龄有明确切分。首页给的年龄有时是"岁"有时是"月"，还有的是"天"，直接拿来用会翻车。稳妥做法是用出生日期和入院日期算，算不出来再退回首页年龄字段。

import pandas as pd def derive_age_and_los(df: pd.DataFrame) -> pd.DataFrame: """派生年龄（岁）和住院天数""" admit = pd.to_datetime(df["admit_date"], errors="coerce") birth = pd.to_datetime(df["birth_date"], errors="coerce") discharge = pd.to_datetime(df["discharge_date"], errors="coerce") # 年龄按周岁算，不足一岁记 0，分组器对 <1 岁另有规则 age_days = (admit - birth).dt.days df["age_years"] = (age_days // 365).where(age_days >= 0, other=pd.NA) # 住院天数：出院-入院，至少 1 天 los = (discharge - admit).dt.days df["los"] = los.where(los >= 1, other=1) return df

这里的关键参数是年龄的取整方式。用// 365是向下取整，和多数分组器"周岁"口径一致；如果某地要求按虚岁或按月龄，改这一行即可。住院天数强制最小为 1，是因为 0 天在部分权重计算里会触发除零或异常分组。

3. ICD 码线性化处理：把带扩展位的编码压成分组器认的形态

3.1 线性化到底在做什么，为什么不能直接查表

CHS-DRG 分组器匹配的是"线性化后的 ICD 码"，不是首页上那串带扩展位的原始码。所谓线性化，核心是三件事：去掉不影响分组的扩展位、把编码统一成标准格式、把"类目码"和"亚目码"的层级关系理清。举个常见情况，首页可能填I50.900，分组器要的是I50.9或I50，具体压到哪一级取决于分组规则表里用的是哪一级。

我见过最典型的翻车是：编码员填了带星剑号的码，或者填了"未特指"的.9，线性化时没处理，结果 CC/MCC 判定全错。所以线性化不是简单的字符串截断，得有一张规则表驱动。

3.2 用规则表驱动线性化，而不是硬编码截断

硬编码code[:5]这种写法迟早出事，因为不同章节的编码长度规则不一样。正确做法是维护一张"原始码模式 -> 线性化码"的映射表，Python 只负责查表和兜底。

import re # 线性化规则：按正则匹配，命中则替换 # 顺序敏感，先匹配具体规则，再走通用兜底 LINEARIZE_RULES = [ (re.compile(r"^([A-Z]\d{2})\.(\d)00$"), r"\1.\2"), # I50.900 -> I50.9 (re.compile(r"^([A-Z]\d{2})\.(\d{2})0$"), r"\1.\2"), # I50.910 -> I50.91 (re.compile(r"^([A-Z]\d{2})\.(\d)$"), r"\1.\2"), # 已是标准形态 (re.compile(r"^([A-Z]\d{2})(\d)$"), r"\1.\2"), # I509 -> I50.9 ] def linearize_code(code: str) -> str: """将单个 ICD 码线性化为分组器标准形态""" if not code or not isinstance(code, str): return "" code = code.strip().upper().replace(" ", "") for pattern, repl in LINEARIZE_RULES: if pattern.match(code): return pattern.sub(repl, code) # 兜底：无法识别的码原样返回并记录，人工复核 print(f"[REVIEW] 未匹配线性化规则: {code}") return code

逻辑说明：规则按从具体到通用排列，re.sub用捕获组保留章节和亚目部分。参数上，LINEARIZE_RULES的顺序不能乱，否则I50.900可能被通用规则先截成I50.9之外的东西。兜底返回原码并打[REVIEW]标记，是为了让编码员能捞出来人工核对，而不是悄悄放过。

3.3 批量线性化与主诊断/主手术的判定

单码线性化只是第一步，真正影响分组的是"哪个是主诊断、哪个是主手术"。首页通常已经标了主诊断，但主手术有时需要按规则判定——比如多个手术时，取与主诊断最相关、或创伤最大的那个。我一般先信任首页标注，再用规则做一致性校验。

def linearize_record(record: dict) -> dict: """对一份病案的所有诊断和手术做线性化""" record["main_diag_lin"] = linearize_code(record.get("main_diag", "")) record["other_diags_lin"] = [linearize_code(c) for c in record.get("other_diags", [])] record["main_proc_lin"] = linearize_code(record.get("main_proc", "")) record["other_procs_lin"] = [linearize_code(c) for c in record.get("other_procs", [])] # 一致性校验：主诊断不应出现在其他诊断里 if record["main_diag_lin"] in record["other_diags_lin"]: print(f"[WARN] 病案 {record.get('patient_id')} 主诊断重复出现在其他诊断") return record

参数说明：main_diag_lin和other_diags_lin分开存，是为了后续 CC/MCC 匹配时能明确排除主诊断本身。一致性校验那几行看着简单，但能捞出一批首页填写错误，属于低成本高回报的检查。

4. 分组器骨架：MDC、ADRG、DRG 三级漏斗的 Python 实现

4.1 三级漏斗的分组顺序不能颠倒

CHS-DRG 的分组是严格顺序的：先根据主诊断定 MDC（主要诊断大类），再在 MDC 内根据主诊断和主手术定 ADRG（核心疾病诊断相关组），最后在 ADRG 内根据 CC/MCC 和年龄等条件定 DRG。顺序颠倒会导致 ADRG 匹配不到，因为 ADRG 表是按 MDC 分区的。

我用一个分组器类把三级逻辑串起来，每级失败都返回明确的错误码，而不是抛异常中断整批处理。

class DRGGrouper: def __init__(self, mdc_table, adrg_table, drg_table, cc_exclude_table): self.mdc_table = mdc_table # 主诊断 -> MDC self.adrg_table = adrg_table # (MDC, 主诊断, 主手术) -> ADRG self.drg_table = drg_table # (ADRG, CC/MCC标志) -> DRG self.cc_exclude_table = cc_exclude_table # 主诊断 -> 需排除的 CC 码 def group(self, record: dict) -> dict: mdc = self._match_mdc(record) if not mdc: return {"drg": None, "error": "MDC_NOT_FOUND"} adrg = self._match_adrg(record, mdc) if not adrg: return {"drg": None, "error": "ADRG_NOT_FOUND"} drg = self._match_drg(record, adrg) return {"drg": drg, "mdc": mdc, "adrg": adrg, "error": None} def _match_mdc(self, record): return self.mdc_table.get(record["main_diag_lin"]) def _match_adrg(self, record, mdc): key = (mdc, record["main_diag_lin"], record["main_proc_lin"]) return self.adrg_table.get(key)

逻辑说明：group方法按 MDC -> ADRG -> DRG 顺序调用，每级返回 None 就带错误码退出。参数上，四张表建议用 dict 或 DataFrame 索引，匹配复杂度 O(1)。错误码MDC_NOT_FOUND和ADRG_NOT_FOUND要分开统计，前者多半是主诊断码有问题，后者多半是主手术缺失或码不对。

4.2 CC/MCC 判定：排除表是分组的胜负手

CC（并发症合并症）和 MCC（严重并发症合并症）的判定直接决定 DRG 落在哪个权重档。判定逻辑是：遍历其他诊断，看是否在 CC/MCC 表里，同时要排除"主诊断本身对应的 CC 码"——这就是cc_exclude_table的作用。

def judge_cc_mcc(record: dict, cc_table: dict, exclude_table: dict) -> str: """返回 'MCC' / 'CC' / 'NONE'""" main_diag = record["main_diag_lin"] excluded = set(exclude_table.get(main_diag, [])) has_mcc, has_cc = False, False for code in record["other_diags_lin"]: if code in excluded: continue level = cc_table.get(code) if level == "MCC": has_mcc = True elif level == "CC": has_cc = True if has_mcc: return "MCC" if has_cc: return "CC" return "NONE"

参数说明：cc_table是码到等级的映射，exclude_table是主诊断到应排除码列表的映射。注意excluded用 set 是为了 O(1) 查找，其他诊断多的时候性能差别明显。返回NONE表示无 CC/MCC，分组时走基础档。

4.3 把三级结果拼成最终 DRG 并输出

最后一步是把 ADRG 和 CC/MCC 标志拼成 DRG 码，通常 DRG 码是 ADRG 加一位数字后缀。这一步查drg_table即可，但要处理"该 ADRG 不接受 CC/MCC"的情况。

def _match_drg(self, record, adrg): cc_level = judge_cc_mcc(record, self.cc_table, self.cc_exclude_table) # 部分 ADRG 只有基础档，CC/MCC 不改变分组 key = (adrg, cc_level) drg = self.drg_table.get(key) if drg is None: drg = self.drg_table.get((adrg, "NONE")) return drg

逻辑说明：先按(adrg, cc_level)查，查不到再退回(adrg, "NONE")，兼容那些不区分 CC/MCC 的 ADRG。参数上，drg_table的键设计成元组，避免字符串拼接带来的歧义。

5. 避坑与排查：线性化和分组里最容易翻车的五件事

5.1 现象：分组结果大批落在错误组，错误码全是 ADRG_NOT_FOUND

原因：主手术码线性化后和 ADRG 表里的码对不上，最常见的是手术码少了小数点或多了扩展位。解决：把[REVIEW]标记的码全部导出，和 ADRG 表里的手术码做一次集合差集，差异码集中处理。我一般会写个一次性脚本对比，而不是逐个猜。

5.2 现象：CC/MCC 判定结果比预期多，权重虚高

原因：cc_exclude_table没配全，主诊断对应的排除码漏了，导致本应排除的其他诊断被算成 CC。解决：核对排除表是否覆盖了所有主诊断章节，尤其是那些"主诊断本身即 CC"的情况。这个坑很隐蔽，因为分组不报错，只是钱算多了。

5.3 现象：年龄字段大量为空，分组掉进新生儿或老年组

原因：出生日期或入院日期格式不统一，pd.to_datetime解析失败变成 NaT，年龄算出来是 NA。解决：提取阶段先统计日期字段的格式分布，对多种格式分别指定format参数解析，解析失败的单独输出人工处理。

5.4 现象：多值诊断切分后码被截断

原因：分隔符配置错误，或者 ICD 码本身含分隔符。解决：先抽样看原始字段，确认分隔符；切分后校验每个码是否符合 ICD 码正则，不符合的整条记录打标复核。

5.5 现象：批量处理跑到一半内存爆掉

原因：一次性把全量首页读进内存，且线性化时对每行都做正则匹配。解决：用pd.read_csv(chunksize=...)分块读，线性化结果做缓存（同一个码不重复匹配），我实测缓存能把大样本的处理时间压掉一大半。

6. 让分组器可验证：用构造样本做回归测试的一个习惯

分组器写完，最怕的是"改了一处规则，别处悄悄坏了"。我的习惯是维护一组构造样本——不是真实病案，而是手工编的、覆盖各 MDC 和 CC/MCC 分支的最小用例，每次改规则先跑这组样本。下面是一个最小测试骨架。

import pytest from grouper import DRGGrouper # 构造样本：覆盖基础档、CC 档、MCC 档各一例 SAMPLES = [ { "patient_id": "T001", "main_diag_lin": "I50.9", "other_diags_lin": [], "main_proc_lin": "", "expect_adrg": "FR1", "expect_cc": "NONE", }, { "patient_id": "T002", "main_diag_lin": "I50.9", "other_diags_lin": ["N18.5"], # 假设为 CC "main_proc_lin": "", "expect_adrg": "FR1", "expect_cc": "CC", }, ] def test_group_consistency(grouper: DRGGrouper): for s in SAMPLES: result = grouper.group(s) assert result["adrg"] == s["expect_adrg"], f"{s['patient_id']} ADRG 不符" cc = judge_cc_mcc(s, grouper.cc_table, grouper.cc_exclude_table) assert cc == s["expect_cc"], f"{s['patient_id']} CC 判定不符"

这段测试的价值不在覆盖率，而在"锁行为"。参数上，expect_adrg和expect_cc是人工确认过的期望值，一旦规则改动导致不符，测试立刻红。我一般还会加一个"全量样本分组结果快照"的对比，把上次分组结果存成文件，这次跑完 diff，差异超过阈值就人工看。

一个具体技巧：线性化函数的缓存用functools.lru_cache包一层，因为同一批病案里重复码极多，缓存命中率通常很高。

from functools import lru_cache @lru_cache(maxsize=10000) def linearize_code_cached(code: str) -> str: return linearize_code(code)

maxsize设 10000 是因为 ICD 码总量有限，基本能全缓存。注意lru_cache要求参数可哈希，字符串没问题，但如果以后要传列表就得改设计。

最后说个我自己的教训：早期我图快，把线性化规则直接写死在分组逻辑里，结果一次编码版本更新，规则要改，我花了整整两天在几千行代码里找散落的截断逻辑。从那以后，凡是"码到码"的转换，一律走独立规则表加缓存，规则表和分组逻辑物理隔离。这个习惯让我后面每次版本切换都能在一小时内完成回归。希望帮到你。

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