质效精研 P36 | Python 赋能:质管人也要学一点代码,批量处理病历数据

1
2
3
4
import pandas as pddf = pd.concat([pd.read_excel(f) for f in files], ignore_index=True)print(df.shape)  # (12847, 86)

1
2
3
4
5
6
7
8
import pandas as pddf = pd.read_excel('5月病案首页.xlsx')print(df.head())      # 看前 5 行print(df.shape)       # (12847, 86)  12847 行,86 列print(df.columns)     # 所有列名print(df.dtypes)      # 每列的数据类型print(df.describe())  # 数值列的统计描述

1
2
3
4
5
import numpy as nparr = np.array([1, 2, 3, 4, 5])print(arr.mean())  # 3.0print(arr.std())   # 1.4142135623730951

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import matplotlib.pyplot as plt# 折线图:某指标 12 个月趋势months = ['1月', '2月', '3月', '4月', '5月', '6月', '7月', '8月', '9月', '10月', '11月', '12月']values = [85, 87, 86, 88, 90, 89, 91, 92, 93, 92, 94, 95]plt.rcParams['font.sans-serif'] = ['SimHei']plt.rcParams['axes.unicode_minus'] = Falseplt.figure(figsize=(10, 5))plt.plot(months, values, marker='o')plt.title('CMI 月度趋势')plt.xlabel('月份')plt.ylabel('CMI')plt.grid(True)plt.savefig('cmi_trend.png', dpi=200)plt.show()

1
2
3
4
5
6
7
8
9
10
11
12
13
from openpyxl import load_workbookfrom openpyxl.styles import Font, PatternFillwb = load_workbook('月报.xlsx')ws = wb.active# 设置标题字体ws['A1'].font = Font(bold=True, size=14)# 设置单元格颜色ws['A1'].fill = PatternFill(start_color='FFFF00', end_color='FFFF00', fill_type='solid')wb.save('月报_样式.xlsx')

1
2
3
4
5
6
7
8
9
10
11
12
import pymysqlconn = pymysql.connect(    host='10.1.2.3',         # HIS 数据库地址    user='quality_reader',   # 只读账号    password='******',    database='his_db',    charset='utf8mb4')df = pd.read_sql('SELECT * FROM diagnosis LIMIT 1000', conn)conn.close()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
import pandas as pdimport globimport os# 1. 找到所有 Excel 文件files = glob.glob('data/病案首页_2026年5月_*.xlsx')print(f'共找到 {len(files)} 个文件')# 2. 循环读取并合并dfs = []for f in files:    dept_name = os.path.basename(f).replace('病案首页_2026年5月_', '').replace('.xlsx', '')    df = pd.read_excel(f)    df['科室'] = dept_name  # 加 1 列「科室」,方便后续分组    dfs.append(df)# 3. 合并所有 DataFramedf_all = pd.concat(dfs, ignore_index=True)print(f'合并后总行数:{len(df_all)}')# 4. 保存df_all.to_excel('output/5月病案首页_全院合并.xlsx', index=False)

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
import pandas as pddf = pd.read_excel('output/5月病案首页_全院合并.xlsx')# 步骤 1:去重print(f'去重前行数:{len(df)}')df = df.drop_duplicates(subset=['住院号'], keep='first')print(f'去重后行数:{len(df)}')# 步骤 2:缺失值处理missing_rate = df.isnull().sum() / len(df) * 100print('缺失率 > 5% 的列:')print(missing_rate[missing_rate > 5].sort_values(ascending=False))df = df.dropna(subset=['主要诊断编码'])median_age = df['年龄'].median()df['年龄'] = df['年龄'].fillna(median_age)df['婚姻状况'] = df['婚姻状况'].fillna('未知')# 步骤 3:异常值处理abnormal_age = df[(df['年龄'] < 0) | (df['年龄'] > 150)]print(f'年龄异常记录:{len(abnormal_age)} 条')df = df[(df['年龄'] >= 0) & (df['年龄'] <= 150)]df = df[(df['住院天数'] >= 0) & (df['住院天数'] <= 365)]df = df[df['性别'].isin(['男', '女', '未知'])]print(f'清洗后行数:{len(df)}')

1
2
3
4
5
6
7
8
9
10
11
12
13
14
import pandas as pddf_patient = pd.read_excel('data/患者信息.xlsx')df_diagnosis = pd.read_excel('data/诊断记录.xlsx')# 内连接(只保留有诊断的患者)df_merged = pd.merge(df_patient, df_diagnosis, on='patient_id', how='inner')# 左连接(保留所有患者,没诊断的填 NaN)df_left = pd.merge(df_patient, df_diagnosis, on='patient_id', how='left')print(f'患者表行数:{len(df_patient)}')print(f'诊断表行数:{len(df_diagnosis)}')print(f'内连接后:{len(df_merged)}')

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
import pandas as pddf = pd.read_excel('output/5月病案首页_全院合并.xlsx')required_fields = [    '住院号', '姓名', '性别', '年龄', '入院日期', '出院日期',    '主要诊断编码', '主要诊断名称',    '主要手术操作编码', '主要手术操作名称',    '入院病情', '出院病情', '科别', '床位医师']issues = []for field in required_fields:    if field not in df.columns:        print(f'警告:字段「{field}」不存在')        continue    missing_count = df[field].isnull().sum()    empty_count = (df[field].astype(str).str.strip() == '').sum()    total_problems = missing_count + empty_count    issues.append({        '字段': field,        '缺失数': missing_count,        '空白数': empty_count,        '总问题数': total_problems,        '问题率(%)': round(total_problems / len(df) * 100, 2)    })df_issues = pd.DataFrame(issues).sort_values('总问题数', ascending=False)print(df_issues)df_issues.to_excel('output/病案首页必填字段核查报告.xlsx', index=False)

1
2
3
4
       字段   缺失数   空白数   总问题数  问题率(%)   主要手术操作名称   1245    89    1334    10.38   主要手术操作编码   1203    89    1292    10.06   入院病情         56      0     56      0.44

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import pandas as pddf = pd.read_excel('data/2026年不良事件.xlsx')# 透视表:按级别 × 类别pivot1 = df.pivot_table(    index='事件级别', columns='事件类别', values='事件ID',    aggfunc='count', fill_value=0, margins=True, margins_name='合计')print('事件级别 × 事件类别:')print(pivot1)# 按科室统计,降序dept_stats = df.groupby('科室').agg(    事件总数=('事件ID', 'count'),    严重事件数=('事件级别', lambda x: (x == '严重').sum()),    已结案数=('状态', lambda x: (x == '已结案').sum())).sort_values('事件总数', ascending=False)print('科室统计:')print(dept_stats)with pd.ExcelWriter('output/不良事件统计月报.xlsx', engine='openpyxl') as writer:    pivot1.to_excel(writer, sheet_name='级别×类别')    dept_stats.to_excel(writer, sheet_name='科室统计')

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import pandas as pddf_use = pd.read_excel('data/5月抗菌药物使用记录.xlsx')df_ddd = pd.read_excel('data/抗菌药物DDD字典.xlsx')df_merged = pd.merge(df_use, df_ddd, on='药物编码', how='left')df_merged['DDDs'] = df_merged['用药总量(克)'] / df_merged['DDD值(克)']drug_ddds = df_merged.groupby(['药物名称', 'ATC分类']).agg(    用药总量=('用药总量(克)', 'sum'),    DDDs=('DDDs', 'sum')).sort_values('DDDs', ascending=False)print(drug_ddds.head(20))df_dept = df_merged.groupby('科室').agg(    DDDs合计=('DDDs', 'sum'),    住院人天=('住院人天', 'sum'))df_dept['DDDs_100人天'] = df_dept['DDDs合计'] / df_dept['住院人天'] * 100print('科室抗菌药物使用强度:')print(df_dept.sort_values('DDDs_100人天', ascending=False))

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
import pandas as pddf = pd.read_excel('data/临床路径数据_2026.xlsx')grouped = df.groupby(['科室', '月份', '路径名称']).agg(    符合路径=('是否符合', lambda x: (x == '是').sum()),    实际入径=('是否入径', lambda x: (x == '是').sum()),    完成路径=('是否完成', lambda x: (x == '是').sum()),    发生变异=('是否变异', lambda x: (x == '是').sum()))grouped['入径率'] = grouped['实际入径'] / grouped['符合路径'] * 100grouped['完成率'] = grouped['完成路径'] / grouped['实际入径'] * 100grouped['变异率'] = grouped['发生变异'] / grouped['实际入径'] * 100grouped = grouped.round(2)high_variation = grouped[grouped['变异率'] > 30].sort_values('变异率', ascending=False)print('高变异率路径:')print(high_variation)grouped.to_excel('output/临床路径入径率分析.xlsx')

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
import pandas as pddf = pd.read_excel('data/5月DRG入组结果.xlsx')# 总体入组率total_rate = (df['DRG编码'].notnull().sum() / len(df)) * 100print(f'全院 DRG 入组率:{total_rate:.2f}%')# 各科室入组率dept_rate = df.groupby('科室').apply(    lambda x: pd.Series({        '病例数': len(x),        '入组数': x['DRG编码'].notnull().sum(),        '入组率(%)': round(x['DRG编码'].notnull().sum() / len(x) * 100, 2)    })).sort_values('入组率(%)', ascending=False)warning = dept_rate[dept_rate['入组率(%)'] < 95]print('入组率 < 95% 的科室:')print(warning)# CMI 计算total_weight = df['DRG权重'].sum()cni_count = len(df[df['DRG编码'].notnull()])cmi = total_weight / cni_countprint(f'全院 CMI:{cmi:.4f}')dept_rate.to_excel('output/DRG入组率核查.xlsx')

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
import pandas as pdimport matplotlib.pyplot as pltimport osplt.rcParams['font.sans-serif'] = ['SimHei']plt.rcParams['axes.unicode_minus'] = Falsedf = pd.read_excel('data/2025年指标数据.xlsx')os.makedirs('output/趋势图', exist_ok=True)indicators = ['CMI', '入组率', '平均住院日', '抗菌药物使用强度', '手术并发症发生率']fig, axes = plt.subplots(2, 3, figsize=(15, 10))axes = axes.flatten()for i, ind in enumerate(indicators):    if ind not in df.columns:        continue    ax = axes[i]    ax.plot(df['月份'], df[ind], marker='o', linewidth=2)    ax.set_title(ind, fontsize=14)    ax.set_xlabel('月份')    ax.set_ylabel(ind)    ax.grid(True, alpha=0.3)    ax.tick_params(axis='x', rotation=45)for j in range(len(indicators), len(axes)):    axes[j].set_visible(False)plt.tight_layout()plt.savefig('output/趋势图/月度指标趋势图.png', dpi=200, bbox_inches='tight')plt.show()print('趋势图已生成')

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
import pandas as pdfrom datetime import datetimedf = pd.read_excel('output/5月病案首页_全院合并.xlsx')report = []report.append(f'数据质量报告 - {datetime.now().strftime("%Y-%m-%d")}')report.append(f'总记录数:{len(df)}')report.append('=' * 50)# 完整性report.append('\n【维度 1:完整性】')required_fields = ['主要诊断编码', '主要手术操作编码', '入院病情', '出院病情']for field in required_fields:    if field in df.columns:        missing_rate = df[field].isnull().sum() / len(df) * 100        status = '通过' if missing_rate < 1 else '不通过'        report.append(f'  {field}:缺失率 {missing_rate:.2f}% [{status}]')# 一致性report.append('\n【维度 2:一致性】')invalid_date = (df['入院日期'] >= df['出院日期']).sum()report.append(f'  入院日期 >= 出院日期:{invalid_date} 条')icd_pattern = df['主要诊断编码'].astype(str).str.match(r'^[A-Z]\d{2}(\.\d{1,2})?$')invalid_icd = (~icd_pattern & df['主要诊断编码'].notnull()).sum()report.append(f'  主要诊断编码格式不规范:{invalid_icd} 条')# 及时性report.append('\n【维度 3:及时性】')df['归档时效'] = (df['归档日期'] - df['出院日期']).dt.total_seconds() / 3600on_time_rate = (df['归档时效'] <= 24).sum() / len(df) * 100report.append(f'  24 小时内归档率:{on_time_rate:.2f}%')# 唯一性report.append('\n【维度 4:唯一性】')duplicate_id = df.duplicated(subset=['住院号'], keep=False).sum()report.append(f'  住院号重复:{duplicate_id} 条')with open('output/数据质量报告_5月.txt', 'w', encoding='utf-8') as f:    f.write('\n'.join(report))print('\n'.join(report))

1
2
3
4
5
6
7
8
9
10
11
12
13
数据质量报告 - 2026-06-24总记录数:12847==================================================【维度 1:完整性】  主要诊断编码:缺失率 0.12% [通过]  主要手术操作编码:缺失率 10.06% [不通过]【维度 2:一致性】  入院日期 >= 出院日期:3 条  主要诊断编码格式不规范:127 条【维度 3:及时性】  24 小时内归档率:89.32%【维度 4:唯一性】  住院号重复:0 条

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
from jinja2 import Templateimport pandas as pdfrom datetime import datetimedf = pd.read_excel('output/5月病案首页_全院合并.xlsx')total = len(df)grouped_rate = df['DRG编码'].notnull().sum() / total * 100cmi = df['DRG权重'].sum() / df['DRG编码'].notnull().sum()data = {    'report_month': '2026年5月',    'generate_date': datetime.now().strftime('%Y-%m-%d'),    'total_cases': total,    'drg_rate': f'{grouped_rate:.2f}',    'cmi': f'{cmi:.4f}',    'avg_los': f'{df["住院天数"].mean():.2f}'}template_str = """<!DOCTYPE html><html><head><meta charset="utf-8"><title>质控月报</title></head><body><h1>{{ report_month }} 全院质控月报</h1><p>生成日期:{{ generate_date }}</p><h2>一、总体情况</h2><ul>  <li>出院病例总数:<b>{{ total_cases }}</b> 例</li>  <li>DRG 入组率:<b>{{ drg_rate }}%</b></li>  <li>CMI:<b>{{ cmi }}</b></li>  <li>平均住院日:<b>{{ avg_los }}</b> 天</li></ul></body></html>"""template = Template(template_str)html_content = template.render(**data)with open('output/月报.html', 'w', encoding='utf-8') as f:    f.write(html_content)# 转 PDF(需要安装 weasyprint)# from weasyprint import HTML# HTML(string=html_content).write_pdf('output/月报.pdf')

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
import jiebaimport resample_text = """患者于 2026-05-10 在全身麻醉下行腹腔镜下胆囊切除术,术中出血约 50ml,术后安返病房。术后第 2 天出现发热,体温最高 38.5℃,考虑切口感染,给予头孢呋辛钠抗感染治疗后好转。"""# 中文分词words = jieba.lcut(sample_text)print('分词结果:')print('/'.join(words))# 提取关键信息surgery_match = re.search(r'行(.*?术)[,。]', sample_text)if surgery_match:    print(f'手术名称:{surgery_match.group(1)}')dates = re.findall(r'\d{4}-\d{2}-\d{2}', sample_text)print(f'涉及日期:{dates}')temps = re.findall(r'(\d+\.?\d*)℃', sample_text)print(f'体温记录:{temps}')complication_keywords = ['感染', '出血', '发热', '疼痛', '肿胀', '渗出']found = [k for k in complication_keywords if k in sample_text]print(f'并发症关键词:{found}')

1
2
3
4
手术名称:腹腔镜下胆囊切除术涉及日期:['2026-05-10']体温记录:['38.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
# streamlit_app.pyimport streamlit as stimport pandas as pdst.title('病案首页核查工具')uploaded = st.file_uploader('上传病案首页 Excel', type=['xlsx'])if uploaded:    df = pd.read_excel(uploaded)    st.success(f'共读取 {len(df)} 条记录')    st.subheader('必填字段核查')    required = ['主要诊断编码', '主要手术操作编码', '入院病情']    for field in required:        if field in df.columns:            missing = df[field].isnull().sum()            rate = missing / len(df) * 100            if rate > 5:                st.error(f'{field}:缺失 {missing} 条 ({rate:.2f}%) - 不通过')            else:                st.success(f'{field}:缺失 {missing} 条 ({rate:.2f}%) - 通过')    if st.checkbox('显示原始数据'):        st.dataframe(df.head(100))