| name | dip-case-grouping |
| description | DIP病历入组技能。根据病案数据的主诊断和手术操作匹配DIP分组规则，为病历自动生成DIP病种编码和分值。当用户需要为病历进行DIP分组、匹配病种编码或执行病历入组时调用。 |
DIP病历入组技能
本技能用于根据导入的病案数据，自动匹配DIP分组规则，为每条病历生成对应的DIP病种编码和分值。
前置条件
重要：执行本技能前，请确保已完成以下数据导入
- DIP分组规则已导入 - 需要先调用
dip-group-import 技能导入DIP分组目录
- 病案数据已导入 - 需要先调用
case-data-import 技能导入病案出院病人数据
数据源文件
病案数据文件
病案数据位于 基础数据/病案出院病人.xlsx，包含以下字段：
| 字段名 | 说明 | 用于匹配 |
|---|
| 住院号 | 患者唯一标识 | 关联键 |
| 患者姓名 | 患者姓名 | - |
| 医师 | 主治医师 | - |
| 科室编号 | 科室编码 | - |
| 出院科室 | 出院科室名称 | - |
| 入院科室 | 入院科室名称 | - |
| 入院日期 | 入院时间 | - |
| 出院日期 | 出院时间 | - |
| 诊断类型 | 主要诊断/其他诊断 | 区分主诊断 |
| 诊断ICD | 诊断ICD-10编码 | 匹配键 |
| 诊断名称 | 诊断描述 | - |
| icd | ICD编码（用于DIP分组） | 匹配键 |
| icd3 | ICD-3位编码 | 备用匹配 |
| 手术icd | 手术ICD编码 | 匹配键 |
| 手术名称 | 手术名称 | - |
DIP分组规则文件
DIP分组规则位于 基础数据/平顶山2025年DIP2.0分组目录库.xlsx，目录Sheet包含：
| 字段名 | 说明 |
|---|
| 病种编码 | DIP病种编码（匹配结果） |
| 病种类型 | 1=核心病种，2=综合病种 |
| 主诊断代码 | 主诊断ICD-10编码 |
| 主诊断名称 | 主诊断名称 |
| 主要操作代码 | 主要操作/手术编码 |
| 主要操作名称 | 主要操作/手术名称 |
| 其他操作代码 | 其他操作编码 |
| 其他操作名称 | 其他操作名称 |
| 病种分值 | 病种分值/权重 |
匹配规则
核心匹配逻辑
DIP入组采用优先级匹配策略，按以下顺序进行匹配：
优先级1: 主诊断代码 + 主要操作代码（精确匹配）
↓ 未匹配
优先级2: 主诊断代码 + 其他操作代码（精确匹配）
↓ 未匹配
优先级3: 主诊断代码（无操作代码匹配）
↓ 未匹配
优先级4: 主诊断ICD-3位编码模糊匹配
↓ 未匹配
未入组: 标记为"未入组"病历
匹配规则详解
规则1：主诊断+主要操作精确匹配（最高优先级）
当病历有手术/操作时，优先匹配：
def match_diagnosis_with_operation(diagnosis_code, operation_code, dip_rules):
match = dip_rules[
(dip_rules['main_diagnosis_code'] == diagnosis_code) &
(dip_rules['main_operation_code'] == operation_code)
]
return match
规则2：主诊断+其他操作匹配
如果主要操作未匹配，尝试匹配其他操作：
def match_diagnosis_with_other_operation(diagnosis_code, operation_code, dip_rules):
match = dip_rules[
(dip_rules['main_diagnosis_code'] == diagnosis_code) &
(dip_rules['other_operation_code'] == operation_code)
]
return match
规则3：仅主诊断匹配
对于内科病例（无手术/操作），仅匹配主诊断：
def match_diagnosis_only(diagnosis_code, dip_rules):
match = dip_rules[
(dip_rules['main_diagnosis_code'] == diagnosis_code) &
(dip_rules['main_operation_code'].isna()) &
(dip_rules['other_operation_code'].isna())
]
return match
规则4：ICD-3位编码模糊匹配
当精确匹配失败时，使用ICD-3位编码进行模糊匹配：
def match_icd3_fuzzy(icd3_code, operation_code, dip_rules):
dip_rules['diagnosis_icd3'] = dip_rules['main_diagnosis_code'].str[:3]
if pd.notna(operation_code):
match = dip_rules[
(dip_rules['diagnosis_icd3'] == icd3_code) &
(dip_rules['main_operation_code'] == operation_code)
]
if len(match) == 0:
match = dip_rules[
(dip_rules['diagnosis_icd3'] == icd3_code) &
(dip_rules['main_operation_code'].isna())
]
else:
match = dip_rules[
(dip_rules['diagnosis_icd3'] == icd3_code) &
(dip_rules['main_operation_code'].isna())
]
return match
数据预处理
步骤1：提取主要诊断
病案数据中同一住院号可能有多条诊断记录，需要提取主要诊断：
def extract_main_diagnosis(df_case):
main_diagnosis = df_case[df_case['诊断类型'] == '主要诊断'].copy()
main_diagnosis = main_diagnosis.groupby('住院号').first().reset_index()
return main_diagnosis
步骤2：聚合手术信息
将同一患者的手术信息聚合：
def aggregate_surgery_info(df_case):
surgery_info = df_case[['住院号', '手术icd', '手术名称']].dropna(subset=['手术icd'])
surgery_info = surgery_info.groupby('住院号').agg({
'手术icd': lambda x: ','.join(x.dropna().unique()),
'手术名称': lambda x: ','.join(x.dropna().unique())
}).reset_index()
return surgery_info
步骤3：合并患者完整信息
def merge_patient_info(main_diagnosis, surgery_info):
patient_info = main_diagnosis.merge(
surgery_info,
on='住院号',
how='left'
)
return patient_info
完整入组流程
import pandas as pd
import numpy as np
def dip_case_grouping(case_file, dip_rule_file, output_file=None):
df_case = pd.read_excel(case_file)
xlsx = pd.ExcelFile(dip_rule_file)
df_dip = pd.read_excel(xlsx, sheet_name='目录')
column_mapping = {
'序号': 'id',
'病种编码': 'dip_code',
'病种类型（1.核心病种；2.综合病种）': 'dip_type',
'主诊断代码': 'main_diagnosis_code',
'主诊断名称': 'main_diagnosis_name',
'主要操作代码': 'main_operation_code',
'主要操作名称': 'main_operation_name',
'其他操作代码': 'other_operation_code',
'其他操作名称': 'other_operation_name',
'病种分值': 'score'
}
df_dip = df_dip.rename(columns=column_mapping)
df_dip = df_dip.dropna(subset=['dip_code', 'main_diagnosis_code'])
main_diagnosis = df_case[df_case['诊断类型'] == '主要诊断'].copy()
main_diagnosis = main_diagnosis.groupby('住院号').first().reset_index()
surgery_info = df_case[['住院号', '手术icd', '手术名称']].dropna(subset=['手术icd'])
surgery_info = surgery_info.groupby('住院号').first().reset_index()
patient_info = main_diagnosis.merge(surgery_info, on='住院号', how='left')
results = []
for _, row in patient_info.iterrows():
patient_id = row['住院号']
diagnosis_code = row.get('icd', row.get('诊断ICD'))
icd3_code = row.get('icd3', diagnosis_code[:3] if pd.notna(diagnosis_code) else None)
operation_code = row.get('手术icd')
matched = None
match_type = None
if pd.notna(operation_code):
matched = df_dip[
(df_dip['main_diagnosis_code'] == diagnosis_code) &
(df_dip['main_operation_code'] == operation_code)
]
if len(matched) > 0:
match_type = '主诊断+主要操作'
if matched is None or len(matched) == 0:
if pd.notna(operation_code):
matched = df_dip[
(df_dip['main_diagnosis_code'] == diagnosis_code) &
(df_dip['other_operation_code'] == operation_code)
]
if len(matched) > 0:
match_type = '主诊断+其他操作'
if matched is None or len(matched) == 0:
matched = df_dip[
(df_dip['main_diagnosis_code'] == diagnosis_code) &
(df_dip['main_operation_code'].isna()) &
(df_dip['other_operation_code'].isna())
]
if len(matched) > 0:
match_type = '仅主诊断'
if matched is None or len(matched) == 0:
if pd.notna(icd3_code):
df_dip['diagnosis_icd3'] = df_dip['main_diagnosis_code'].str[:3]
if pd.notna(operation_code):
matched = df_dip[
(df_dip['diagnosis_icd3'] == icd3_code) &
(df_dip['main_operation_code'] == operation_code)
]
if len(matched) > 0:
match_type = 'ICD3+操作'
if len(matched) == 0:
matched = df_dip[
(df_dip['diagnosis_icd3'] == icd3_code) &
(df_dip['main_operation_code'].isna())
]
if len(matched) > 0:
match_type = 'ICD3模糊'
if matched is not None and len(matched) > 0:
best_match = matched.iloc[0]
results.append({
'住院号': patient_id,
'患者姓名': row.get('患者姓名'),
'主诊断代码': diagnosis_code,
'手术代码': operation_code,
'DIP编码': best_match['dip_code'],
'DIP类型': '核心病种' if best_match['dip_type'] == 1 else '综合病种',
'病种分值': best_match['score'],
'匹配方式': match_type,
'入组状态': '已入组'
})
else:
results.append({
'住院号': patient_id,
'患者姓名': row.get('患者姓名'),
'主诊断代码': diagnosis_code,
'手术代码': operation_code,
'DIP编码': None,
'DIP类型': None,
'病种分值': None,
'匹配方式': None,
'入组状态': '未入组'
})
result_df = pd.DataFrame(results)
if output_file:
result_df.to_excel(output_file, index=False)
return result_df
if __name__ == '__main__':
case_file = '基础数据/病案出院病人.xlsx'
dip_rule_file = '基础数据/平顶山2025年DIP2.0分组目录库.xlsx'
output_file = '基础数据/processed/病历DIP入组结果.xlsx'
result = dip_case_grouping(case_file, dip_rule_file, output_file)
print(f'总病历数: {len(result)}')
print(f'已入组: {len(result[result["入组状态"] == "已入组"])}')
print(f'未入组: {len(result[result["入组状态"] == "未入组"])}')
print(f'入组率: {len(result[result["入组状态"] == "已入组"]) / len(result) * 100:.2f}%')
输出结果
入组结果文件
输出文件：基础数据/processed/病历DIP入组结果.xlsx
| 字段 | 说明 |
|---|
| 住院号 | 患者住院号 |
| 患者姓名 | 患者姓名 |
| 主诊断代码 | 主诊断ICD编码 |
| 手术代码 | 手术/操作编码 |
| DIP编码 | 匹配的DIP病种编码 |
| DIP类型 | 核心病种/综合病种 |
| 病种分值 | 病种分值 |
| 匹配方式 | 匹配规则说明 |
| 入组状态 | 已入组/未入组 |
入组统计报告
执行完成后生成统计报告：
=== DIP病历入组统计报告 ===
总病历数: XXX
已入组: XXX
未入组: XXX
入组率: XX.XX%
=== 匹配方式分布 ===
主诊断+主要操作: XXX
主诊断+其他操作: XXX
仅主诊断: XXX
ICD3+操作: XXX
ICD3模糊: XXX
=== 未入组原因分析 ===
诊断编码不在规则中: XXX
操作编码不匹配: XXX
数据库更新
入组完成后，更新病案信息表：
UPDATE biz_case_info ci
JOIN (
SELECT patient_id, dip_code, dip_name, dip_score
FROM 入组结果表
) r ON ci.patient_id = r.patient_id
SET
ci.dip_code = r.dip_code,
ci.dip_name = r.dip_name,
ci.dip_score = r.dip_score,
ci.update_time = NOW();
注意事项
- 数据完整性：确保病案数据中的诊断编码格式正确（ICD-10标准）
- 编码规范：诊断编码使用ICD-10标准，操作编码使用ICD-9-CM-3标准
- 匹配优先级：严格按照优先级顺序匹配，确保最精确匹配
- 未入组处理：未入组病历需要人工审核，可能需要补充诊断或调整规则
- 分值计算：入组后可计算DIP金额 = 病种分值 × 分值单价
- 数据期间：注意DIP规则的数据期间与病案数据期间匹配
常见问题处理
1. 入组率低
- 检查诊断编码格式是否正确
- 检查DIP规则是否覆盖该病种
- 考虑使用ICD-3位编码模糊匹配
2. 手术病历未入组
- 确认手术编码格式正确
- 检查DIP规则中是否包含该操作
- 尝试匹配其他操作代码
3. 多个匹配结果
- 优先选择核心病种（dip_type=1）
- 优先选择有操作代码的匹配
- 按病种分值排序选择
4. 诊断编码格式不一致
