quality-measures

name	quality-measures
description	Pipeline skill for computing HEDIS quality measures from claims and clinical data. Use when the user asks to calculate HEDIS measure rates, check continuous enrollment, build denominator/numerator logic, detect care gaps, compute utilization rates, identify high-cost claimants, or score risk stratification indices. Triggers include "calculate HEDIS", "measure rate", "continuous enrollment check", "care gap detection", "denominator query", "numerator logic", "utilization rate", "high-cost claimant", "Charlson score", "LACE score", "claims analysis", "quality measure SQL".
usage	Use when building or running quality measure calculation pipelines including enrollment checks, measure rates, and care gap detection.
version	1.0.0
validated_against	{"date":"2025-01-15T00:00:00.000Z","packages":{"pandas":"2.2","sqlalchemy":"2.0"}}
tags	["skill","category:pipeline","quality-measures","hedis","hcls"]

Quality Measures Calculation Pipeline

Overview

Provide deterministic Python and SQL code snippets for calculating HEDIS quality measures, detecting care gaps, computing utilization rates, identifying high-cost claimants, and scoring risk stratification indices from claims and clinical data.

Usage

• Calculate HEDIS measure rates, check continuous enrollment, detect care gaps, and score risk indices

Core Concepts

Response Format

• Lead with the command or code the user needs — explain after
• Structure as: confirm inputs → working code → key parameters explained → gotchas
• One complete working example per task; do not show every alternative
• Keep code comments minimal and functional (what, not why-it-exists)
• Target: 50-100 lines of code with brief surrounding explanation

Approach Selection

Scenario	Approach	Key consideration
Single measure, ad-hoc	Python calculate_measure()	Fast iteration, easy debugging
Enterprise batch (all measures)	SQL CTEs per measure	Scales to millions of members
Data source: claims only	Use procedure/revenue codes for numerator	No clinical data available
Data source: claims + EHR	Supplement with lab values, vitals	Higher capture rate
Enrollment check: single payer	max_gap_days=45 (HEDIS default)	One enrollment table
Enrollment check: multi-payer	Merge enrollment spans first, then check	Avoid double-counting gaps

1. Continuous Enrollment Check

1.1 Python Implementation

"""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 = [], 
     me.iloc[][] > start:
        g = (me.iloc[][] - start).days
        total_gap += g
        gap_periods.append({: (start.date()), : (me.iloc[][].date()), : g})
     i  (, (me)):
         me.iloc[i][] > me.iloc[i-][] + pd.Timedelta(days=):
            g = (me.iloc[i][] - me.iloc[i-][]).days - 
            total_gap += g
            gap_periods.append({: (me.iloc[i-][].date()), : (me.iloc[i][].date()), : g})
     {: total_gap <= max_gap_days, : total_gap, : gap_periods}

1.2 SQL Implementation

-- 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)  total_gap_days,
         (g.total_gap_days, )         is_continuously_enrolled
 anchor_check a
  total_gaps g  a.member_id  g.member_id;

2. HEDIS Measure Calculation

2.1 Generic Measure Calculator (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, ),
        gap_members=(gap_ids),
    )

2.2 Full Measure Calculation (SQL — CDC HbA1c Testing Example)

-- 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  (,)
       service_date    
)
 ()  denom,
       (  e.member_id       )  excluded,
       (  e.member_id    n.member_id       )  numer,
       ROUND((  e.member_id    n.member_id       )
              (()  (  e.member_id       ), ), )  rate
 denominator d
  exclusions e  d.member_id  e.member_id
  numerator n  d.member_id  n.member_id;

3. Care Gap Detection

"""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"]))
            ][].unique()
            eligible = eligible[eligible[].isin(dx_members)]
        year_claims = claims[claims[].between((year_start), (anchor))]
        closed = year_claims[
            year_claims[].isin(measure[])
        ][].unique()
         _, row  eligible[~eligible[].isin(closed)].iterrows():
            sw = measure.get(, )
            gaps.append({
                : row[], : measure[],
                : sw, : row.get(, ),
                : (sw *  + (row.get(, ) * , ) + , ),
            })
     pd.DataFrame(gaps).sort_values(, ascending=).reset_index(drop=)

4. Utilization Rate Computation

-- 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  member_id   admit_date)  prev_dc
         claims  claim_type    admit_date    
    ) s  DATEDIFF(, prev_dc, admit_date)    prev_dc  
)
 ROUND(ed.cntm.member_count,)  ed_per_1000,
       ROUND(ip.cntm.member_count,)  ip_per_1000,
       ROUND(readmit.cntm.member_count,)  readmit_per_1000
 members m, ed, ip, readmit;

5. High-Cost Claimant Identification

"""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,
    }

6. Risk Stratification Scoring

6.1 Charlson Comorbidity Index (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", ), : (, ),
    : (, ), : (, ),
    : (, ), : (, ),
    : (, ),
    : (, ), : (, ), : (, ),
    : (, ),
}


 () -> :
    
    matched: [, ] = {}
     code  [c.replace(, )  c  diagnosis_codes]:
         prefix, (cond, wt)  CHARLSON_MAP.items():
             code.startswith(prefix):
                 cond ==     matched:
                    
                 cond == :
                    matched.pop(, )
                 cond ==     matched:
                    
                 cond == :
                    matched.pop(, )
                 cond   matched  wt > matched[cond]:
                    matched[cond] = wt
                
     {: (matched.values()), : matched}

6.2 LACE Readmission Risk Score

"""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, : c, : e}, : tier}

7. Measure Stratification by Plan/Provider

-- 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    ), ), )  rate_pct
 denominator_members m
  exclusion_members e  m.member_id  e.member_id
  numerator_members n  m.member_id  n.member_id
  health_plan, assigned_pcp_npi
  rate_pct ;

8. Parameter Reference

Parameter	Default	Description
measurement_year	Current calendar year	HEDIS reporting year
max_gap_days	45	Maximum allowable enrollment gap
anchor_date	Dec 31 of measurement year	Date member must be enrolled through
dx_lookback_years	2	Years to look back for qualifying diagnoses
threshold_percentile	95	Percentile cutoff for high-cost identification
readmission_window	30	Days after discharge for readmission flag

9. Common Mistakes

• Wrong: Including segment end dates when calculating enrollment gap days Right: Calculate gaps as the calendar days between segments (day after end of segment A to day before start of segment B) Why: Off-by-one errors in gap calculation incorrectly exclude continuously enrolled members or include ineligible ones

• Wrong: Using enrollment segments as-is without clipping to the measurement year boundaries Right: Clip all enrollment segments to the measurement year start and end dates before calculating gaps Why: Segments extending beyond the year inflate coverage calculations and produce incorrect enrollment determinations

• Wrong: Using service date for inpatient utilization metrics Right: Use admit date and discharge date for inpatient claims — service date is for professional/outpatient claims Why: Inpatient utilization (admissions, readmissions, length of stay) is defined by admission events, not individual service lines

• Wrong: Counting a member multiple times in the numerator when they have multiple qualifying events Right: Deduplicate numerator events by member ID — each member counts once regardless of how many tests or services they received Why: Multiple events for the same member inflate the numerator and produce artificially high measure rates

• Wrong: Calculating member age as of the run date or an arbitrary date Right: Calculate age as of the measure-specific anchor date (typically December 31 of the measurement year) Why: Using the wrong anchor date shifts age-based eligibility and includes or excludes members incorrectly

• Wrong: Only looking at the measurement year for qualifying diagnoses Right: Apply the measure-specified lookback period (typically 2 years) for identifying members with qualifying conditions Why: Many HEDIS measures require a diagnosis in the measurement year OR the year prior; using only one year misses eligible members