assets/2026-07-01-医数智联系列开篇-医院数字化工作者的工具箱_2026-07-01_18-28-11.jpg

医数智联·第 18 期:Python + DRG 入组分析 —— 5 类异常筛查的代码版,告别复杂 SQL

[!ABSTRACT] 核心摘要

  • 本期主题:Python 版 DRG 入组异常筛查——SQL 不熟也能做医保结算防御
  • 核心结论:pandas 的 merge + groupby 替代复杂 SQL,代码可读性提升 10 倍
  • 5 大异常 Python 版:低码高套、歧义病例、误入组、主诊错误、入组编码缺失
  • 典型效果:用 100 行 Python 代码,替代 500 行 SQL,还可直接画图
  • 阅读时长:约 10 分钟

一、场景引入:为什么用 Python 做 DRG 分析?

[!example] 一个真实的周四上午
第 10 期讲过用 SQL 筛查 DRG 入组异常,但很多医院人反馈:

「SQL 的窗口函数、CTE 递归看着头大。」

「好几个 JOIN 嵌套,改一个地方全乱。」

「我要的不只是表格,还要图表,SQL 完不成。」

Python + pandas 能解决这一切。

这是 SQL 在医院数据分析中的典型痛点——逻辑复杂时,可读性和可维护性变差。

维度 SQL Python + pandas
学习曲线 陡(JOIN/窗口函数) 平缓(像读英语)
可读性 嵌套深时差 始终线性,像脚本
调试 难(只能整段跑) 易(单步执行)
可视化 需另起工具 直接出图
复用 复制粘贴 函数化,一次写永久用
部署 依赖数据库 独立脚本,任何电脑跑

[!quote] 狼叔的判断
“SQL 适合’最后一道查询’,Python 适合’整条流水线’。” 这一期,带你把第 10 期的 5 类异常筛查全部用 Python 重写。

二、工具原理:pandas 实现 DRG 分析的核心模式

2.1 5 类分析模式

DRG 异常类型 核心 pandas 操作
低码高套 groupby + transform 算中位数,过滤
歧义病例 merge + filter 主诊与操作匹配
误入组 apply 自定义函数判断 MDC 组
主诊错误 merge 主诊表 + 副诊表,识别 MCC 漏选
编码缺失 isnull() 检查必填字段

2.2 数据准备

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import pandas as pd
import numpy as np

# 读首页主表
首页 = pd.read_excel('本月首页.xlsx')

# 读诊断表(每个病案号多条诊断)
诊断 = pd.read_excel('诊断表.xlsx')

# 读手术表
手术 = pd.read_excel('手术表.xlsx')

# 读 DRG 入组结果(医保局反馈)
drg = pd.read_excel('DRG入组结果.xlsx')

# 读 ICD 字典(标准名称、是否 MCC)
icd = pd.read_excel('ICD字典.xlsx')

三、医院实战:5 类异常 Python 版

3.1 异常 ①:低码高套筛查

[!example] 实战 ①:低码高套
逻辑:同主诊 ICD,实际权重 > 历史中位权重 × 1.5 → 疑似高套

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def detect_低码高套(首页, drg, 阈值倍数=1.5):
    """低码高套筛查"""

    # 1. 计算每个主诊的历史中位权重(用过去 3 个月的数据)
    历史数据 = pd.read_excel('过去3月首页.xlsx')  # 包含历史 DRG 权重
    主诊中位权重 = 历史数据.groupby('主诊ICD')['DRG权重'].median().to_dict()

    # 2. 合并本月数据 + 历史中位
    合并 = 首页.merge(drg, on='病案号', how='left')
    合并['历史中位权重'] = 合并['主诊ICD'].map(主诊中位权重)

    # 3. 计算「权重倍数」并筛异常
    合并['权重倍数'] = 合并['DRG权重'] / 合并['历史中位权重']
    合并['阈值倍数'] = 阈值倍数

    # 4. 标记疑似高套
    异常 = 合并[合并['权重倍数'] > 阈值倍数].copy()
    异常['异常类型'] = '低码高套'
    异常['异常说明'] = (
        '实际权重 ' +
        异常['DRG权重'].round(2).astype(str) +
        ' 高于历史中位 ' +
        异常['历史中位权重'].round(2).astype(str) +
        ' 的 ' +
        异常['权重倍数'].round(1).astype(str) + ' 倍'
    )

    return 异常[['病案号', '科室', '主诊ICD', '主诊名称', 'DRG权重', '历史中位权重', '权重倍数', '异常类型', '异常说明']]

# 运行
高套异常 = detect_低码高套(首页, drg)
print(f"发现 {len(高套异常)} 例低码高套")
print(高套异常.head(10))

输出示例:

1
2
3
4
发现 23 例低码高套
    病案号   科室 主诊ICD  主诊名称  DRG权重  历史中位权重  权重倍数  异常类型           异常说明
0  BA001    骨科  S72.001  股骨颈骨折  2.85       1.65     1.7   低码高套   实际权重 2.85 高于历史中位 1.65 的 1.7 倍
1  BA015    骨科  S72.001  股骨颈骨折  3.12       1.65     1.9   低码高套   实际权重 3.12 高于历史中位 1.65 的 1.9 倍

3.2 异常 ②:歧义病例筛查

[!example] 实战 ②:歧义病例
逻辑:主诊是内科病,但主要操作是外科大手术 → 歧义

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def detect_歧义病例(首页, 手术):
    """歧义病例筛查:内科主诊 + 外科操作"""

    # 1. 取出每条病案的主要操作
    主要操作 = 手术[手术['主要操作标记'] == 1][['病案号', '手术ICD', '手术名称']]

    # 2. 合并
    合并 = 首页.merge(主要操作, on='病案号', how='inner')

    # 3. 定义内科主诊关键词 + 外科大手术 ICD 前缀
    内科关键词 = ['感染', '炎症', '衰竭', '综合征', '高血压', '糖尿病']
    外科大手术前缀 = ['01', '02', '03', '04', '05']  # ICD 章节 1-5 是外科

    # 4. 标记歧义:内科主诊 + 外科大手术
    合并['是否歧义'] = (
        合并['主诊名称'].str.contains('|'.join(内科关键词), na=False) &
        合并['手术ICD'].str[:2].isin(外科大手术前缀)
    )

    # 5. 输出
    异常 = 合并[合并['是否歧义']].copy()
    异常['异常类型'] = '歧义病例'
    异常['异常说明'] = '内科主诊[' + 异常['主诊名称'] + ']但有外科大手术[' + 异常['手术名称'] + ']'

    return 异常[['病案号', '科室', '主诊名称', '手术名称', '异常类型', '异常说明']]

歧义 = detect_歧义病例(首页, 手术)
print(f"发现 {len(歧义)} 例歧义病例")

3.3 异常 ③:误入组筛查(主诊 vs 操作不匹配)

[!example] 实战 ③:误入组
逻辑:根据主诊 ICD + 主要操作,推算理论 MDC 组,与实际 MDC 组对比

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def detect_误入组(首页, 手术, drg):
    """误入组筛查"""

    # 1. 取出主要操作
    主要操作 = 手术[手术['主要操作标记'] == 1][['病案号', '手术ICD']]

    # 2. 合并
    合并 = 首页.merge(主要操作, on='病案号', how='left').merge(drg, on='病案号')

    # 3. 定义"理论 MDC 组"判断函数
    def 推算理论MDC(row):
        if pd.isna(row['手术ICD']):
            return 'MDCE'  # 内科组
        章节 = str(row['手术ICD'])[:2]
        if 章节 in ['01', '02', '03']:
            return 'MDCA'  # 神经系统手术
        elif 章节 in ['04', '05', '06']:
            return 'MDCB'  # 眼科/耳鼻喉
        elif 章节 in ['07', '08']:
            return 'MDCC'  # 呼吸/循环
        else:
            return 'MDCD'  # 消化系统

    合并['理论MDC'] = 合并.apply(推算理论MDC, axis=1)
    合并['是否误入组'] = 合并['理论MDC'] != 合并['实际MDC组']

    异常 = 合并[合并['是否误入组']].copy()
    异常['异常类型'] = '误入组'
    异常['异常说明'] = (
        '实际入组[' + 异常['实际MDC组'] +
        '],但理论应该入[' + 异常['理论MDC'] + ']'
    )

    return 异常[['病案号', '主诊ICD', '手术ICD', '实际MDC组', '理论MDC', '异常类型', '异常说明']]

误入组 = detect_误入组(首页, 手术, drg)
print(f"发现 {len(误入组)} 例误入组")

3.4 异常 ④:主诊错误筛查(MCC 漏选)

[!example] 实战 ④:主诊错误
逻辑:副诊有 MCC,但主诊选了非 MCC → 主诊可能选错

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
def detect_主诊错误(首页, 诊断, icd):
    """主诊错误筛查:副诊有 MCC,主诊选了非 MCC"""

    # 1. 取出所有副诊
    副诊 = 诊断[诊断['诊断类型'] == '副诊'][['病案号', 'ICD编码']]

    # 2. 合并 ICD 字典,识别 MCC
    副诊_mcc = 副诊.merge(
        icd[['ICD编码', '是否MCC']],
        left_on='ICD编码',
        right_on='ICD编码',
        how='left'
    )

    # 3. 找出有 MCC 副诊的病案
    有MCC副诊 = 副诊_mcc[副诊_mcc['是否MCC'] == 1]['病案号'].unique()

    # 4. 检查这些病案的主诊是否也是 MCC
    候选病案 = 首页[首页['病案号'].isin(有MCC副诊)].copy()
    候选病案 = 候选病案.merge(
        icd[['ICD编码', '是否MCC']],
        left_on='主诊ICD',
        right_on='ICD编码',
        how='left',
        suffixes=('', '_主诊')
    )

    # 5. 标记:有 MCC 副诊,但主诊不是 MCC
    异常 = 候选病案[
        (候选病案['是否MCC_主诊'] != 1) &
        (候选病案['是否MCC_主诊'].notna())
    ].copy()

    异常['异常类型'] = '主诊错误'
    异常['异常说明'] = '病案存在 MCC 副诊,但主诊未选 MCC 疾病'

    return 异常[['病案号', '主诊ICD', '主诊名称', '异常类型', '异常说明']]

主诊错误 = detect_主诊错误(首页, 诊断, icd)
print(f"发现 {len(主诊错误)} 例主诊错误")

3.5 异常 ⑤:入组编码缺失

[!example] 实战 ⑤:编码缺失
逻辑:做了手术但手术 ICD 为空

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def detect_编码缺失(首页, 手术):
    """入组编码缺失"""

    # 1. 首页有「手术名称」字段
    有手术首页 = 首页[首页['手术名称'].notna()]

    # 2. 关联手术表,看 ICD 是否缺失
    合并 = 有手术首页.merge(手术, on='病案号', how='left')
    缺失 = 合并[合并['手术ICD'].isna()]

    异常 = 缺失.copy()
    异常['异常类型'] = '编码缺失'
    异常['异常说明'] = '做了手术但手术 ICD 未编码'

    return 异常[['病案号', '手术名称', '异常类型', '异常说明']]

编码缺失 = detect_编码缺失(首页, 手术)
print(f"发现 {len(编码缺失)} 例编码缺失")

3.6 主函数:汇总 5 类异常

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def validate_DRG(首页, 手术, 诊断, drg, icd):
    """主函数:汇总 5 类异常"""

    print("=" * 50)
    print("开始 DRG 入组异常筛查")
    print("=" * 50)

    异常1 = detect_低码高套(首页, drg)
    print(f"✓ 低码高套:{len(异常1)} 例")

    异常2 = detect_歧义病例(首页, 手术)
    print(f"✓ 歧义病例:{len(异常2)} 例")

    异常3 = detect_误入组(首页, 手术, drg)
    print(f"✓ 误入组:{len(异常3)} 例")

    异常4 = detect_主诊错误(首页, 诊断, icd)
    print(f"✓ 主诊错误:{len(异常4)} 例")

    异常5 = detect_编码缺失(首页, 手术)
    print(f"✓ 编码缺失:{len(异常5)} 例")

    # 汇总
    全部异常 = pd.concat([异常1, 异常2, 异常3, 异常4, 异常5], ignore_index=True)

    # 按异常类型统计 + 画图
    类型统计 = 全部异常['异常类型'].value_counts()

    import matplotlib.pyplot as plt
    plt.rcParams['font.sans-serif'] = ['SimHei']

    fig, ax = plt.subplots(figsize=(10, 6))
    类型统计.plot(kind='bar', ax=ax, color='steelblue', edgecolor='black')
    ax.set_title('本月 DRG 入组异常分布', fontsize=14, fontweight='bold')
    ax.set_xlabel('异常类型')
    ax.set_ylabel('例数')
    plt.tight_layout()
    plt.savefig('DRG异常分布.png', dpi=300)

    # 输出 Excel
    with pd.ExcelWriter('DRG入组异常.xlsx') as writer:
        全部异常.to_excel(writer, sheet_name='全部异常', index=False)
        类型统计.reset_index().to_excel(writer, sheet_name='类型统计', index=False)

    print(f"\n总异常数:{len(全部异常)}")
    print(f"结果已保存到 DRG入组异常.xlsx")

    return 全部异常

# 运行
异常清单 = validate_DRG(首页, 手术, 诊断, drg, icd)

四、避坑指南:5 个让 Python 脚本「报错」的常见错误

[!warning] 雷区 ①:merge 后字段名重复
症状:ValueError: columns overlap but no suffix specified。
解决:加 suffixes:

1
df.merge(other, on='病案号', suffixes=('', '_y'))

[!warning] 雷区 ②:apply 函数没用 vectorize
症状:数据量 1 万行,apply 跑了 30 秒。
解决:用 np.where 或 np.select 替代:

1
2
3
4
5
# 慢
df['等级'] = df['分值'].apply(lambda x: '高' if x > 80 else '低')

# 快
df['等级'] = np.where(df['分值'] > 80, '高', '低')

[!warning] 雷区 ③:历史数据不足
症状:detect_低码高套 用过去 3 个月数据,但新主诊没历史。
解决:缺失填充 1.0(表示无历史可比):

1
主诊中位权重 = 主诊中位权重.fillna(主诊中位权重.median())

[!warning] 雷区 ④:中文匹配漏掉
症状:主诊名称有「肺部感染」但 str.contains('肺部') 没命中(可能中间有空格)。
解决:正则 + 容错:str.contains('肺部.*感染', regex=True, na=False)。

[!warning] 雷区 ⑤:输出 Excel 中文乱码
症状:Excel 打开中文是乱码。
解决:指定编码:df.to_excel('结果.xlsx', encoding='utf-8-sig') 或用 openpyxl 引擎。

五、Python vs SQL 怎么选?

[!TIP] 选型决策

1
2
3
4
5
6
7
8
9
10
11
12
你要做什么?
├─ 简单查询(单表 + WHERE + GROUP BY)
│   └─ ✅ SQL(快)
│
├─ 复杂分析(多表 JOIN + 窗口函数)
│   └─ ✅ Python(可读性好,可画图)
│
├─ 自动出报告(包含图表)
│   └─ ✅ Python(一站式)
│
└─ 实时性高(每天 1 万次查询)
    └─ ✅ SQL(数据库直接跑)

六、下期预告

[!quote] 第 19 期
Python + 病历关键词提取 —— 让医学 NLP 不再神秘

这一期,带你用 jieba(中文分词)+ sklearn(机器学习)做病历关键词提取,自动识别主诉、诊断、治疗等关键信息。

[!TIP] 互动
如果你觉得这篇「医数智联·第 18 期」对你有用,请点赞、在看、转发给科里的同事。
我是白衣狼,一个用数字化工具把医院管理做「轻」的实战派。咱们下期见。