quality-measures

质量指标计算流程

概述

提供确定性的 Python 和 SQL 代码片段，用于计算 HEDIS 质量指标、检测护理缺口、计算利用率、识别高成本索赔人，以及根据理赔和临床数据计算风险分层指数。

用法

• 计算 HEDIS 指标率、检查连续参保情况、检测护理缺口并计算风险指数

核心概念

响应格式

• 先给出用户所需的命令或代码，再进行说明
• 结构为：确认输入 → 可运行代码 → 说明关键参数 → 注意事项
• 每项任务提供一个完整的可运行示例；不展示所有替代方案
• 代码注释应简洁实用（说明用途，而非存在原因）
• 目标：50-100 行代码，附简短说明

方法选择

场景	方法	关键考量
单项指标、临时计算	Python calculate_measure()	迭代快速，易于调试
企业级批处理（所有指标）	每项指标使用 SQL CTE	可扩展至数百万名成员
数据源：仅理赔数据	使用手术或收入代码作为分子	无临床数据可用
数据源：理赔数据 + EHR	补充实验室数值、生命体征	数据捕获率更高
参保检查：单一付款方	max_gap_days=45（HEDIS 默认值）	使用一张参保表
参保检查：多付款方	先合并参保时间段，再进行检查	避免重复计算间隔

1. 连续参保检查

1.1 Python 实现

"""Check continuous enrollment with allowable gap."""
import pandas as pd


def check_continuous_enrollment(
    enrollment: pd.DataFrame, member_id: str,
    start_date: str, end_date: str, max_gap_days: int = 45,
) -> dict:
    """Check if a member is continuously enrolled with allowable gap.

    Args:
        enrollment: DataFrame [member_id, enroll_start, enroll_end].
        member_id: Member to check.
        start_date: Measurement period start (e.g., '2025-01-01').
        end_date: Measurement period end / anchor date (e.g., '2025-12-31').
        max_gap_days: Maximum allowable gap in days (HEDIS default: 45).

    Returns:
        Dict with is_enrolled (bool), total_gap_days (int), gap_periods (list).
    """
    start, end = pd.Timestamp(start_date), pd.Timestamp(end_date)
    me = enrollment[enrollment["member_id"] == member_id].copy()
    me["enroll_start"] = pd.to_datetime(me["enroll_start"]).clip(lower=start)
    me["enroll_end"] = pd.to_datetime(me["enroll_end"]).clip(upper=end)
    me = me[me["enroll_start"] <= me["enroll_end"]].sort_values("enroll_start").reset_index(drop=True)
    if me.empty:
        return {"is_enrolled": False, "total_gap_days": (end - start).days, "gap_periods": []}
    if not (me["enroll_end"] >= end).any():
        return {"is_enrolled": False, "total_gap_days": -1, "gap_periods": []}
    gap_periods, total_gap = [], 0
    if me.iloc[0]["enroll_start"] > start:
        g = (me.iloc[0]["enroll_start"] - start).days
        total_gap += g
        gap_periods.append({"from": str(start.date()), "to": str(me.iloc[0]["enroll_start"].date()), "days": g})
    for i in range(1, len(me)):
        if me.iloc[i]["enroll_start"] > me.iloc[i-1]["enroll_end"] + pd.Timedelta(days=1):
            g = (me.iloc[i]["enroll_start"] - me.iloc[i-1]["enroll_end"]).days - 1
            total_gap += g
            gap_periods.append({"from": str(me.iloc[i-1]["enroll_end"].date()), "to": str(me.iloc[i]["enroll_start"].date()), "days": g})
    return {"is_enrolled": total_gap <= max_gap_days, "total_gap_days": total_gap, "gap_periods": gap_periods}

1.2 SQL 实现

-- Continuous enrollment check with allowable gap (≤45 days)
WITH enrollment_segments AS (
    SELECT member_id, enroll_start, enroll_end,
           LEAD(enroll_start) OVER (PARTITION BY member_id ORDER BY enroll_start) AS next_start
    FROM enrollment
    WHERE enroll_end >= '2025-01-01' AND enroll_start <= '2025-12-31'
),
gaps AS (
    SELECT member_id,
           DATEDIFF(day, enroll_end, next_start) - 1 AS gap_days
    FROM enrollment_segments
    WHERE next_start IS NOT NULL
      AND DATEDIFF(day, enroll_end, next_start) > 1
),
total_gaps AS (
    SELECT member_id, SUM(gap_days) AS total_gap_days
    FROM gaps
    GROUP BY member_id
),
anchor_check AS (
    SELECT DISTINCT member_id
    FROM enrollment
    WHERE enroll_end >= '2025-12-31'
)
SELECT a.member_id,
       COALESCE(g.total_gap_days, 0) AS total_gap_days,
       CASE WHEN COALESCE(g.total_gap_days, 0) <= 45 THEN 1 ELSE 0 END AS is_continuously_enrolled
FROM anchor_check a
LEFT JOIN total_gaps g ON a.member_id = g.member_id;

2. HEDIS 指标计算

2.1 通用指标计算器（Python）

"""Generic HEDIS measure rate calculator."""
import pandas as pd
from dataclasses import dataclass


@dataclass
class MeasureResult:
    measure_id: str
    denominator_count: int
    exclusion_count: int
    numerator_count: int
    rate: float
    gap_members: list[str]


def calculate_measure(
    eligible: pd.DataFrame,
    exclusions: pd.DataFrame,
    numerator_events: pd.DataFrame,
    measure_id: str,
) -> MeasureResult:
    """Calculate a HEDIS measure rate.

    Args:
        eligible: Denominator members. Columns: [member_id].
        exclusions: Excluded members. Columns: [member_id, exclusion_reason].
        numerator_events: Members meeting numerator. Columns: [member_id, event_date].
        measure_id: Measure identifier (e.g., 'CDC-HbA1c-Testing').

    Returns:
        MeasureResult with rate and gap member list.
    """
    denom_ids = set(eligible["member_id"])
    excl_ids = set(exclusions["member_id"])
    eligible_denom = denom_ids - excl_ids
    numer_ids = set(numerator_events["member_id"]) & eligible_denom
    gap_ids = eligible_denom - numer_ids
    denom_count = len(eligible_denom)
    rate = len(numer_ids) / denom_count if denom_count > 0 else 0.0
    return MeasureResult(
        measure_id=measure_id,
        denominator_count=denom_count,
        exclusion_count=len(excl_ids & denom_ids),
        numerator_count=len(numer_ids),
        rate=round(rate, 4),
        gap_members=sorted(gap_ids),
    )

2.2 完整指标计算（SQL — CDC HbA1c 检测示例）

-- CDC: Comprehensive Diabetes Care — HbA1c Testing
WITH denominator AS (
    SELECT DISTINCT m.member_id
    FROM members m
    JOIN claims c ON m.member_id = c.member_id
    JOIN continuously_enrolled ce ON m.member_id = ce.member_id
    WHERE DATEDIFF(year, m.date_of_birth, '2025-12-31') BETWEEN 18 AND 75
      AND c.diagnosis_code LIKE 'E11%'
      AND c.service_date BETWEEN '2024-01-01' AND '2025-12-31'
      AND ce.is_continuously_enrolled = 1
),
exclusions AS (
    SELECT DISTINCT member_id FROM claims
    WHERE diagnosis_code IN ('Z51.5', 'N18.6')
       OR revenue_code IN ('0115','0125','0135','0145','0155','0235')
),
numerator AS (
    SELECT DISTINCT member_id FROM claims
    WHERE procedure_code IN ('83036','83037')
      AND service_date BETWEEN '2025-01-01' AND '2025-12-31'
)
SELECT COUNT(*) AS denom,
       SUM(CASE WHEN e.member_id IS NOT NULL THEN 1 ELSE 0 END) AS excluded,
       SUM(CASE WHEN e.member_id IS NULL AND n.member_id IS NOT NULL THEN 1 ELSE 0 END) AS numer,
       ROUND(SUM(CASE WHEN e.member_id IS NULL AND n.member_id IS NOT NULL THEN 1 ELSE 0 END)*1.0
             / NULLIF(COUNT(*) - SUM(CASE WHEN e.member_id IS NOT NULL THEN 1 ELSE 0 END), 0), 4) AS rate
FROM denominator d
LEFT JOIN exclusions e ON d.member_id = e.member_id
LEFT JOIN numerator n ON d.member_id = n.member_id;

3. 护理缺口检测

"""Detect and prioritize open care gaps across members."""
import pandas as pd
from datetime import date


def detect_care_gaps(
    members: pd.DataFrame,
    measures: list[dict],
    claims: pd.DataFrame,
    measurement_year: int = 2025,
) -> pd.DataFrame:
    """Detect open care gaps for a population.

    Args:
        members: DataFrame [member_id, date_of_birth, gender, risk_score].
        measures: List of dicts with keys: measure_id, age_min, age_max,
            gender (str|None), diagnosis_codes (list), numerator_codes (list),
            star_weight (1 or 3).
        claims: DataFrame [member_id, service_date, diagnosis_code, procedure_code].
        measurement_year: Calendar year for measurement.

    Returns:
        DataFrame: [member_id, measure_id, star_weight, risk_score, priority_score].
    """
    anchor = date(measurement_year, 12, 31)
    year_start = date(measurement_year, 1, 1)
    gaps = []
    for measure in measures:
        eligible = members.copy()
        eligible["age"] = eligible["date_of_birth"].apply(lambda d: (anchor - d).days // 365)
        eligible = eligible[eligible["age"].between(measure["age_min"], measure["age_max"])]
        if measure.get("gender"):
            eligible = eligible[eligible["gender"] == measure["gender"]]
        if measure.get("diagnosis_codes"):
            dx_members = claims[
                claims["diagnosis_code"].str.startswith(tuple(measure["diagnosis_codes"]))
            ]["member_id"].unique()
            eligible = eligible[eligible["member_id"].isin(dx_members)]
        year_claims = claims[claims["service_date"].between(str(year_start), str(anchor))]
        closed = year_claims[
            year_claims["procedure_code"].isin(measure["numerator_codes"])
        ]["member_id"].unique()
        for _, row in eligible[~eligible["member_id"].isin(closed)].iterrows():
            sw = measure.get("star_weight", 1)
            gaps.append({
                "member_id": row["member_id"], "measure_id": measure["measure_id"],
                "star_weight": sw, "risk_score": row.get("risk_score", 0),
                "priority_score": round(sw * 30 + min(row.get("risk_score", 0) * 25, 25) + 20, 1),
            })
    return pd.DataFrame(gaps).sort_values("priority_score", ascending=False).reset_index(drop=True)

4. 利用率计算

-- Utilization rates per 1000 members: ED, inpatient, 30-day readmissions
WITH members AS (
    SELECT COUNT(DISTINCT member_id) AS member_count FROM continuously_enrolled WHERE is_continuously_enrolled = 1
),
ed AS (
    SELECT COUNT(*) AS cnt FROM claims WHERE revenue_code IN ('0450','0451','0452','0456','0459') AND service_date BETWEEN '2025-01-01' AND '2025-12-31'
),
ip AS (
    SELECT COUNT(DISTINCT claim_id) AS cnt FROM claims WHERE claim_type = 'inpatient' AND admit_date BETWEEN '2025-01-01' AND '2025-12-31'
),
readmit AS (
    SELECT COUNT(*) AS cnt FROM (
        SELECT member_id, admit_date, LAG(discharge_date) OVER (PARTITION BY member_id ORDER BY admit_date) AS prev_dc
        FROM claims WHERE claim_type = 'inpatient' AND admit_date BETWEEN '2025-01-01' AND '2025-12-31'
    ) s WHERE DATEDIFF(day, prev_dc, admit_date) <= 30 AND prev_dc IS NOT NULL
)
SELECT ROUND(ed.cnt*1000.0/m.member_count,1) AS ed_per_1000,
       ROUND(ip.cnt*1000.0/m.member_count,1) AS ip_per_1000,
       ROUND(readmit.cnt*1000.0/m.member_count,1) AS readmit_per_1000
FROM members m, ed, ip, readmit;

5. 高成本索赔人识别

"""Identify high-cost claimants and analyze cost drivers."""
import pandas as pd
import numpy as np


def identify_high_cost_claimants(
    claims: pd.DataFrame,
    threshold_percentile: float = 95,
    measurement_year: int = 2025,
) -> dict:
    """Identify high-cost claimants above a percentile threshold.

    Args:
        claims: DataFrame [member_id, service_date, paid_amount, claim_type, diagnosis_code].
        threshold_percentile: Percentile cutoff (default 95th).
        measurement_year: Calendar year to analyze.

    Returns:
        Dict with threshold, count, pct of total spend, and member DataFrame.
    """
    year_claims = claims[
        pd.to_datetime(claims["service_date"]).dt.year == measurement_year
    ]
    member_costs = (
        year_claims.groupby("member_id")
        .agg(total_paid=("paid_amount", "sum"), claim_count=("paid_amount", "count"))
        .reset_index()
    )
    threshold = np.percentile(member_costs["total_paid"], threshold_percentile)
    high_cost = member_costs[member_costs["total_paid"] >= threshold].sort_values(
        "total_paid", ascending=False
    )
    return {
        "threshold_amount": round(threshold, 2),
        "high_cost_count": len(high_cost),
        "high_cost_pct_of_total": round(
            high_cost["total_paid"].sum() / member_costs["total_paid"].sum() * 100, 1
        ),
        "high_cost_members": high_cost,
    }

6. 风险分层评分

6.1 Charlson 合并症指数（Python）

"""Calculate Charlson Comorbidity Index from diagnosis codes."""

# ICD-10 prefix → (condition, weight). Subset of key mappings.
CHARLSON_MAP = {
    "I21": ("mi", 1), "I22": ("mi", 1), "I50": ("chf", 1),
    "I70": ("pvd", 1), "I71": ("pvd", 1), "I6": ("cvd", 1),
    "F01": ("dementia", 1), "F03": ("dementia", 1), "G30": ("dementia", 1),
    "J4": ("copd", 1), "M05": ("ctd", 1), "M06": ("ctd", 1),
    "K25": ("pud", 1), "K26": ("pud", 1),
    "K70": ("mild_liver", 1), "K73": ("mild_liver", 1), "K74": ("mild_liver", 1),
    "E109": ("dm_uncomp", 1), "E119": ("dm_uncomp", 1),
    "E102": ("dm_comp", 2), "E112": ("dm_comp", 2),
    "G81": ("hemiplegia", 2), "N18": ("renal", 2),
    "C": ("cancer", 2),
    "C77": ("metastatic", 6), "C78": ("metastatic", 6), "C79": ("metastatic", 6),
    "B20": ("hiv", 6),
}


def charlson_score(diagnosis_codes: list[str]) -> dict:
    """Calculate Charlson Comorbidity Index. Returns {score, conditions}."""
    matched: dict[str, int] = {}
    for code in [c.replace(".", "") for c in diagnosis_codes]:
        for prefix, (cond, wt) in CHARLSON_MAP.items():
            if code.startswith(prefix):
                if cond == "cancer" and "metastatic" in matched:
                    continue
                if cond == "metastatic":
                    matched.pop("cancer", None)
                if cond == "dm_uncomp" and "dm_comp" in matched:
                    continue
                if cond == "dm_comp":
                    matched.pop("dm_uncomp", None)
                if cond not in matched or wt > matched[cond]:
                    matched[cond] = wt
                break
    return {"score": sum(matched.values()), "conditions": matched}

6.2 LACE 再入院风险评分

"""Calculate LACE readmission risk score."""


def lace_score(length_of_stay: int, acuity: str, charlson: int, ed_visits_6mo: int) -> dict:
    """Calculate LACE index for 30-day readmission risk.

    Returns dict with total score, component scores, and risk tier.
    """
    l = 7 if length_of_stay >= 14 else (5 if length_of_stay >= 7 else (4 if length_of_stay >= 4 else min(length_of_stay, 3)))
    a = {"emergent": 3, "urgent": 2, "elective": 0}.get(acuity.lower(), 0)
    c = 5 if charlson >= 4 else min(charlson, 3)
    e = min(ed_visits_6mo, 4)
    total = l + a + c + e
    tier = "high" if total >= 10 else ("moderate" if total >= 5 else "low")
    return {"total": total, "components": {"L": l, "A": a, "C": c, "E": e}, "risk_tier": tier}

7. 按计划/服务提供者进行指标分层

-- Measure rate stratified by health plan and provider
SELECT health_plan, assigned_pcp_npi,
       COUNT(*) - SUM(CASE WHEN e.member_id IS NOT NULL THEN 1 ELSE 0 END) AS eligible_denom,
       SUM(CASE WHEN e.member_id IS NULL AND n.member_id IS NOT NULL THEN 1 ELSE 0 END) AS numerator,
       ROUND(SUM(CASE WHEN e.member_id IS NULL AND n.member_id IS NOT NULL THEN 1 ELSE 0 END) * 100.0
             / NULLIF(COUNT(*) - SUM(CASE WHEN e.member_id IS NOT NULL THEN 1 ELSE 0 END), 0), 1) AS rate_pct
FROM denominator_members m
LEFT JOIN exclusion_members e ON m.member_id = e.member_id
LEFT JOIN numerator_members n ON m.member_id = n.member_id
GROUP BY health_plan, assigned_pcp_npi
ORDER BY rate_pct ASC;

8. 参数参考

参数	默认值	说明
measurement_year	当前日历年	HEDIS 报告年度
max_gap_days	45	允许的最大参保间隔
anchor_date	指标测量年度的 12 月 31 日	成员必须持续参保至该日期
dx_lookback_years	2	回溯查找符合条件诊断的年数
threshold_percentile	95	识别高成本索赔人的百分位数阈值
readmission_window	30	出院后用于标记再入院的天数

9. 常见错误

• 错误： 计算参保间隔天数时，将分段结束日期计入间隔 正确： 将间隔计算为两个分段之间的日历天数（A 段结束后的次日到 B 段开始前一日） 原因： 间隔计算中的差一错误会错误排除连续参保的成员，或纳入不符合条件的成员

• 错误： 不经处理直接使用参保分段，未将其截断至测量年度边界 正确： 计算间隔前，将所有参保分段截断至测量年度的开始和结束日期 原因： 延伸至年度范围之外的分段会夸大覆盖率计算结果，并导致参保判定错误

• 错误： 使用服务日期计算住院利用率指标 正确： 对住院理赔使用入院日期和出院日期；服务日期适用于专业医疗或门诊理赔 原因： 住院利用率（入院、再入院、住院时长）按入院事件定义，而非单项服务明细

• 错误： 成员有多个符合条件的事件时，在分子中重复计数 正确： 按成员 ID 对分子事件去重；无论成员接受了多少次检测或服务，每位成员只计一次 原因： 同一成员的多个事件会使分子膨胀，导致指标比率虚高

• 错误： 按运行日期或任意日期计算成员年龄 正确： 按指标指定的锚定日期计算年龄（通常为测量年度的 12 月 31 日） 原因： 使用错误的锚定日期会改变基于年龄的资格判定，从而错误纳入或排除成员

• 错误： 仅查看测量年度内是否有符合条件的诊断 正确： 按指标指定的回溯期（通常为 2 年）识别患有符合条件疾病的成员 原因： 许多 HEDIS 指标要求在测量年度或前一年出现诊断；只查看一年会漏掉符合条件的成员