quality-measures について
HEDIS 指標の計算や医療請求・臨床データの分析に使う、Python と SQL のコード例をまとめたスキルです。品質指標、ケアギャップ、利用率、高額請求者の特定、リスクスコアに対応します。
主な機能
- 01HEDIS 指標率の計算
- 02継続加入の確認
- 03ケアギャップの検出
- 04医療利用率の計算
- 05高額請求者の特定
- 06リスク層別化スコア
活用シーン
- 01HEDIS 指標の計算と継続加入の確認
- 02未実施の予防ケア・治療項目の特定
- 03メンバー別の救急・入院利用率の算出
- 04高額請求者の抽出と再入院リスク評価
インストール
コマンドをコピーしてターミナルで実行します。
$
npx skills add https:/インストール後、Claude Code で/quality-measuresを入力して読み込まれたか確認します。
詳細
品質指標の計算パイプライン
概要
HEDIS 品質指標の計算、ケアギャップの検出、利用率の算出、高額請求者の特定、請求データと臨床データに基づくリスク層別化指標のスコアリングを行う、決定論的な Python および SQL のコードスニペットを提供します。
使用方法
- HEDIS 指標の割合を計算し、継続加入状況を確認し、ケアギャップを検出して、リスク指標をスコアリングします
コアコンセプト
応答形式
- 必要なコマンドまたはコードを最初に示し、その後で説明します
- 次の順序で構成します：入力の確認 → 動作するコード → 主要なパラメーターの説明 → 注意点
- 各タスクにつき、完全に動作する例を1つ示します。代替案をすべて示す必要はありません
- コードコメントは最小限かつ実用的な内容にします（存在理由ではなく、処理内容を説明します）
- コードは50～100行を目安にし、前後の説明は簡潔にします
アプローチの選択
| シナリオ | アプローチ | 主な考慮事項 |
|---|---|---|
| 単一指標、アドホックな処理 | Python calculate_measure() | 反復が速く、デバッグが容易です |
| エンタープライズ向けバッチ（全指標） | 指標ごとの SQL CTE | 数百万のメンバーにスケールできます |
| データソース：請求データのみ | 分子に手技コード／収益コードを使用します | 臨床データは利用できません |
| データソース：請求データ＋EHR | 検査値やバイタル値で補完します | データ捕捉率が高まります |
| 加入状況の確認：単一の保険者 | max_gap_days=45（HEDIS のデフォルト） | 加入テーブルは1つです |
| 加入状況の確認：複数の保険者 | まず加入期間を統合してから確認します | ギャップの二重カウントを避けます |
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 の開始日の前日まで） 理由： 空白日数の計算で1日のずれが生じると、継続して加入しているメンバーを誤って除外したり、資格のないメンバーを含めたりします
-
誤り： 測定年の境界に合わせて切り詰めず、加入セグメントをそのまま使用する 正しい方法： 空白日数を計算する前に、すべての加入セグメントを測定年の開始日と終了日に合わせて切り詰めます 理由： 測定年を超えて延びるセグメントは、カバー率を過大に算出させ、加入資格の判定を誤らせます
-
誤り： 入院利用率の指標に診療日を使用する 正しい方法： 入院請求では入院日と退院日を使用します。診療日は医療専門職による外来請求に使用します 理由： 入院利用率（入院、再入院、在院日数）は、個々の診療項目ではなく入院イベントによって定義されます
-
誤り： 条件を満たすイベントが複数ある場合、分子で同じメンバーを複数回数える 正しい方法： 分子のイベントをメンバー ID で重複排除します。検査やサービスを何回受けたかにかかわらず、各メンバーは1回だけ数えます 理由： 同じメンバーのイベントを複数回数えると、分子が過大になり、指標の割合が実際より高くなります
-
誤り： 実行日または任意の日付時点のメンバー年齢を計算する 正しい方法： 指標ごとに指定された基準日（通常は測定年の12月31日）時点の年齢を計算します 理由： 誤った基準日を使用すると、年齢に基づく資格判定にずれが生じ、メンバーを誤って含めたり除外したりします
-
誤り： 条件を満たす診断について測定年だけを確認する 正しい方法： 条件を満たす疾患のあるメンバーを特定する際は、指標で指定された遡及期間（通常は2年間）を適用します 理由： 多くの HEDIS 指標では、測定年またはその前年の診断が必要です。1年分だけを確認すると、資格のあるメンバーを見落とします
コメント
Markdown 対応