| 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
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)
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
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."""
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
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
