risk-adjustment

リスク調整 — パイプラインスキル

概要

CMS-HCC のリスク調整に使用できる、決定論的でコピーしてすぐに実行できる Python および SQL コードを提供します。 ICD-10 から HCC へのクロスウォーク、階層の解決、RAF スコアの計算、Rx／検査値の代理指標によるコーディングギャップの特定を行います。

使用方法

• ICD-10 から HCC へのクロスウォークを適用し、疾患階層をプログラムで解決します
• メンバーレベルの RAF スコアを計算し、Rx／検査値の代理指標からコーディングギャップを特定します

コアコンセプト

---

回答の形式

• ユーザーが必要とするコマンドまたはコードを最初に示し、その後に説明します
• 次の構成にします：入力の確認 → 動作するコード → 主要パラメーターの説明 → 注意点
• 各タスクにつき、完全に動作する例を1つ示します。あらゆる代替案は示しません
• コードコメントは最小限かつ実用的にします（存在理由ではなく、何をするかを記述）
• 目安：コード50～100行と、簡潔な説明

1. ICD-10 から HCC へのクロスウォーク

Python

import pandas as pd


def load_crosswalk(filepath: str) -> pd.DataFrame:
    """Load CMS ICD-10-to-CC crosswalk. Expected columns: [icd10, cc]."""
    xwalk = pd.read_csv(filepath, dtype=str)
    xwalk.columns = [c.strip().lower() for c in xwalk.columns]
    xwalk["icd10"] = xwalk["icd10"].str.replace(".", "", regex=False).str.upper()
    xwalk["cc"] = xwalk["cc"].astype(int)
    return xwalk


def map_diagnoses_to_ccs(dx_df: pd.DataFrame, xwalk: pd.DataFrame) -> pd.DataFrame:
    """Map member diagnoses to Condition Categories.

    Args:
        dx_df: [member_id, icd10, service_date, provider_type].
        xwalk: [icd10, cc].
    Returns:
        Deduplicated [member_id, icd10, cc].
    """
    dx = dx_df.copy()
    dx["icd10"] = dx["icd10"].str.replace(".", "", regex=False).str.upper()
    qualifying = {"MD", "DO", "NP", "PA", "CNS"}
    dx = dx[dx["provider_type"].str.upper().isin(qualifying)]
    mapped = dx.merge(xwalk, on="icd10", how="inner")
    return mapped.drop_duplicates(subset=["member_id", "cc"])

SQL

SELECT DISTINCT d.member_id, d.icd10, x.cc
FROM diagnoses d
JOIN icd10_cc_crosswalk x ON REPLACE(d.icd10, '.', '') = x.icd10
WHERE d.provider_type IN ('MD','DO','NP','PA','CNS')
  AND d.service_date BETWEEN '2025-01-01' AND '2025-12-31';

---

2. 階層の解決

Python

import pandas as pd
from collections import defaultdict

# V24 hierarchies: {higher_cc: [lower_ccs_it_supersedes]}
V24_HIERARCHIES = {
    17: [18, 19], 18: [19],           # Diabetes
    85: [86, 87], 86: [87],           # Heart failure
    111: [112],                        # COPD > asthma
    136: [137, 138], 137: [138],      # Renal
    8: [9,10,11,12], 9: [10,11,12], 10: [11,12], 11: [12],  # Cancer
    107: [108],                        # Vascular
    27: [28, 29], 28: [29],           # Liver
    51: [52],                          # Dementia
    82: [83, 84], 83: [84],           # Hemiplegia
}


def resolve_hierarchies(member_ccs: pd.DataFrame,
                        hierarchies: dict = None) -> pd.DataFrame:
    """Remove lower CCs when higher CC in same hierarchy is present.

    Args:
        member_ccs: [member_id, cc].
        hierarchies: {higher_cc: [lower_ccs]}. Defaults to V24.
    Returns:
        [member_id, hcc] with only surviving CCs.
    """
    if hierarchies is None:
        hierarchies = V24_HIERARCHIES

    superseded_by = defaultdict(set)
    for higher, lowers in hierarchies.items():
        for lower in lowers:
            superseded_by[lower].add(higher)

    results = []
    for mid, grp in member_ccs.groupby("member_id"):
        cc_set = set(grp["cc"])
        for cc in cc_set:
            if not superseded_by.get(cc, set()).intersection(cc_set):
                results.append({"member_id": mid, "hcc": cc})
    return pd.DataFrame(results)

SQL

WITH hierarchy_rules AS (
    SELECT 17 AS hi, 18 AS lo UNION ALL SELECT 17,19 UNION ALL SELECT 18,19 UNION ALL
    SELECT 85,86 UNION ALL SELECT 85,87 UNION ALL SELECT 86,87 UNION ALL
    SELECT 111,112 UNION ALL SELECT 136,137 UNION ALL SELECT 136,138 UNION ALL SELECT 137,138
),
superseded AS (
    SELECT mc.member_id, mc.cc
    FROM member_ccs mc
    JOIN hierarchy_rules h ON mc.cc = h.lo
    JOIN member_ccs mc2 ON mc.member_id = mc2.member_id AND mc2.cc = h.hi
)
SELECT mc.member_id, mc.cc AS hcc
FROM member_ccs mc
LEFT JOIN superseded s ON mc.member_id = s.member_id AND mc.cc = s.cc
WHERE s.cc IS NULL;

---

3. RAF スコアの計算

import pandas as pd

# V24 Community Non-Dual Aged (example subset)
DEMO_COEFF = {
    ("M","65-69"): 0.395, ("M","70-74"): 0.487, ("M","75-79"): 0.596,
    ("M","80-84"): 0.728, ("M","85-89"): 0.896, ("M","90-94"): 1.003, ("M","95+"): 1.073,
    ("F","65-69"): 0.339, ("F","70-74"): 0.421, ("F","75-79"): 0.532,
    ("F","80-84"): 0.668, ("F","85-89"): 0.854, ("F","90-94"): 0.979, ("F","95+"): 1.055,
}

HCC_COEFF = {
    8: 2.484, 9: 0.975, 10: 0.690, 17: 0.368, 18: 0.368, 19: 0.118,
    47: 0.545, 51: 0.437, 52: 0.294, 85: 0.441, 86: 0.335, 87: 0.237,
    96: 0.296, 111: 0.335, 112: 0.199, 136: 0.288, 137: 0.237, 138: 0.237,
}

INTERACTIONS = {
    frozenset(["diabetes","chf"]): 0.154,
    frozenset(["chf","copd"]): 0.175,
    frozenset(["chf","renal"]): 0.154,
    frozenset(["diabetes","chf","copd"]): 0.047,
}

HCC_GROUP = {
    17: "diabetes", 18: "diabetes", 85: "chf", 86: "chf",
    111: "copd", 112: "copd", 136: "renal", 137: "renal", 138: "renal",
}


def calculate_raf(demographics: pd.DataFrame, member_hccs: pd.DataFrame) -> pd.DataFrame:
    """Calculate RAF scores. demographics: [member_id, sex, age_group]. member_hccs: [member_id, hcc]."""
    hcc_map = member_hccs.groupby("member_id")["hcc"].apply(set).to_dict()
    rows = []
    for _, r in demographics.iterrows():
        mid, hccs = r["member_id"], hcc_map.get(r["member_id"], set())
        demo = DEMO_COEFF.get((r["sex"], r["age_group"]), 0.0)
        hcc_score = sum(HCC_COEFF.get(h, 0.0) for h in hccs)
        groups = {HCC_GROUP[h] for h in hccs if h in HCC_GROUP}
        interact = sum(v for k, v in INTERACTIONS.items() if k.issubset(groups))
        rows.append({"member_id": mid, "demo": round(demo, 3), "hcc_score": round(hcc_score, 3),
                      "interaction": round(interact, 3), "total_raf": round(demo + hcc_score + interact, 3),
                      "hcc_count": len(hccs), "hccs": sorted(hccs)})
    return pd.DataFrame(rows).sort_values("total_raf", ascending=False)

SQL

WITH hcc_scores AS (
    SELECT mh.member_id, SUM(c.coefficient) AS hcc_score, COUNT(*) AS hcc_count
    FROM member_hccs mh
    JOIN hcc_coefficients c ON mh.hcc = c.hcc
    WHERE c.model_version = 'V24' AND c.segment = 'CNA'
    GROUP BY mh.member_id
),
demo_scores AS (
    SELECT md.member_id, dc.coefficient AS demo_score
    FROM member_demographics md
    JOIN demographic_coefficients dc ON md.sex = dc.sex AND md.age_group = dc.age_group
    WHERE dc.model_version = 'V24' AND dc.segment = 'CNA'
)
SELECT d.member_id, d.demo_score, COALESCE(h.hcc_score, 0) AS hcc_score,
       ROUND(d.demo_score + COALESCE(h.hcc_score, 0), 3) AS total_raf
FROM demo_scores d LEFT JOIN hcc_scores h ON d.member_id = h.member_id
ORDER BY total_raf DESC;

---

4. コーディングギャップの特定

4a. Rx 代理指標によるギャップ

import pandas as pd

RX_MAP = {
    "metformin":      {"condition": "diabetes",       "icd_pfx": "E11", "hccs": [17,18,19]},
    "glipizide":      {"condition": "diabetes",       "icd_pfx": "E11", "hccs": [17,18,19]},
    "insulin":        {"condition": "diabetes",       "icd_pfx": "E11", "hccs": [17,18,19]},
    "furosemide":     {"condition": "heart_failure",  "icd_pfx": "I50", "hccs": [85,86,87]},
    "spironolactone": {"condition": "heart_failure",  "icd_pfx": "I50", "hccs": [85,86,87]},
    "albuterol":      {"condition": "copd_asthma",    "icd_pfx": "J44", "hccs": [111,112]},
    "donepezil":      {"condition": "dementia",       "icd_pfx": "F03", "hccs": [51,52]},
    "memantine":      {"condition": "dementia",       "icd_pfx": "F03", "hccs": [51,52]},
}


def identify_rx_gaps(rx_df: pd.DataFrame, dx_df: pd.DataFrame, year: int) -> pd.DataFrame:
    """Members with Rx evidence but no matching diagnosis in payment year."""
    rx_yr = rx_df[rx_df["fill_date"].dt.year == year]
    dx_yr = dx_df[dx_df["service_date"].dt.year == year].copy()
    dx_yr["pfx"] = dx_yr["icd10"].str[:3]

    gaps = []
    for drug, m in RX_MAP.items():
        with_rx = set(rx_yr[rx_yr["drug_name"].str.lower().str.contains(drug, na=False)]["member_id"])
        with_dx = set(dx_yr[dx_yr["pfx"] == m["icd_pfx"]]["member_id"])
        for mid in with_rx - with_dx:
            gaps.append({"member_id": mid, "condition": m["condition"],
                         "evidence": drug, "target_hccs": m["hccs"], "gap_type": "rx_proxy"})
    return pd.DataFrame(gaps).drop_duplicates(subset=["member_id", "condition"])

4b. 検査値の代理指標によるギャップ

import pandas as pd

LAB_MAP = {
    "HbA1c": {"thresh": 6.5, "op": ">=", "condition": "diabetes",       "icd_pfx": "E11", "hccs": [17,18,19]},
    "eGFR":  {"thresh": 60,  "op": "<",  "condition": "ckd",            "icd_pfx": "N18", "hccs": [136,137,138]},
    "BNP":   {"thresh": 100, "op": ">=", "condition": "heart_failure",  "icd_pfx": "I50", "hccs": [85,86,87]},
    "BMI":   {"thresh": 40,  "op": ">=", "condition": "morbid_obesity", "icd_pfx": "E66", "hccs": [22]},
}


def identify_lab_gaps(labs: pd.DataFrame, dx_df: pd.DataFrame, year: int) -> pd.DataFrame:
    """Members with abnormal labs but no matching diagnosis."""
    labs_yr = labs[labs["result_date"].dt.year == year]
    dx_yr = dx_df[dx_df["service_date"].dt.year == year].copy()
    dx_yr["pfx"] = dx_yr["icd10"].str[:3]

    gaps = []
    for test, m in LAB_MAP.items():
        t = labs_yr[labs_yr["test_name"].str.upper() == test.upper()].copy()
        t["val"] = pd.to_numeric(t["result_value"], errors="coerce")
        abn = t[t["val"] >= m["thresh"]] if m["op"] == ">=" else t[t["val"] < m["thresh"]]
        with_dx = set(dx_yr[dx_yr["pfx"] == m["icd_pfx"]]["member_id"])
        for mid in set(abn["member_id"]) - with_dx:
            gaps.append({"member_id": mid, "condition": m["condition"],
                         "evidence": test, "target_hccs": m["hccs"], "gap_type": "lab_proxy"})
    return pd.DataFrame(gaps).drop_duplicates(subset=["member_id", "condition"])

SQL — Rx ギャップの検出

WITH rx_diabetes AS (
    SELECT DISTINCT member_id FROM rx_claims
    WHERE LOWER(drug_name) IN ('metformin','glipizide','insulin glargine','insulin lispro')
      AND fill_date BETWEEN '2025-01-01' AND '2025-12-31'
),
dx_diabetes AS (
    SELECT DISTINCT member_id FROM diagnoses
    WHERE icd10 LIKE 'E11%' AND service_date BETWEEN '2025-01-01' AND '2025-12-31'
      AND provider_type IN ('MD','DO','NP','PA')
)
SELECT r.member_id, 'diabetes' AS condition, 'rx_proxy' AS gap_type
FROM rx_diabetes r LEFT JOIN dx_diabetes d ON r.member_id = d.member_id
WHERE d.member_id IS NULL;

---

5. エンドツーエンドのパイプライン

def project_risk_scores(dx_df, xwalk_df, demo_df, hierarchies=None):
    """Full pipeline: crosswalk → hierarchy → RAF → summary."""
    ccs = map_diagnoses_to_ccs(dx_df, xwalk_df)
    hccs = resolve_hierarchies(ccs, hierarchies)
    scores = calculate_raf(demo_df, hccs)
    print(f"Members: {len(scores)}, Mean RAF: {scores['total_raf'].mean():.3f}, "
          f"Median: {scores['total_raf'].median():.3f}")
    return scores

---

6. パラメーターのリファレンス

関数	主要パラメーター	デフォルト	ガイダンス
load_crosswalk	CMS ファイル	年次リリース	支払年度が一致していることを確認します
resolve_hierarchies	hierarchies	V24	PY 2026 以降は V28 を使用し、移行期間中は両方を使用します
calculate_raf	係数テーブル	V24 CNA	対象の人口セグメントに合わせます
identify_rx_gaps	year	現在	ギャップは年度ごとに異なります
identify_lab_gaps	検査値の閾値	臨床基準	ガイドラインに応じて調整します

---

よくある間違い

• 誤り： ICD-10 コードからピリオドを削除せずに、ICD-10 から HCC へのクロスウォークを適用する 正しい方法： クロスウォークとの結合前に str.replace(".", "").upper() でコードを正規化します 理由： CMS のクロスウォークファイルでは書式なしのコード（例：E11.9 ではなく E119）が使われます。一致しない場合、有効なマッピングが警告なしに除外されます

• 誤り： 階層を解決する前に、すべての CC 係数を合計する 正しい方法： スコア計算の前に必ず resolve_hierarchies() を呼び出し、各階層グループで重症度が最も高い CC のみを残します 理由： 置き換えられた CC を数えると RAF スコアが過大になり、収益予測が不正確になります

• 誤り： 支払年度 2026 以降に V24 の階層と係数を使用する 正しい方法： PY 2026 以降は V28 テーブルを使用し、PY 2025 は加重値（V24 を 33％、V28 を 67％）を使用します 理由： CMS は 2026 年に V28 の完全加重へ移行します。古い係数を使うと、スコアに大きな誤差が生じます

• 誤り： クロスウォークのマッピングに、あらゆる医療提供者種別の診断を含める 正しい方法： マッピング前に、要件を満たす医療提供者種別（MD、DO、NP、PA、CNS）に絞り込みます 理由： CMS はリスク調整に、適格な医療提供者との対面診療で得られた診断のみを認めています

• 誤り： Rx 代理指標によるギャップを確定診断として扱い、そのままコーディングする 正しい方法： Rx／検査値の代理指標は、要件を満たす診療での医療提供者による記録が必要な、ギャップの疑いの特定に使用します 理由： 代理指標はスクリーニングのシグナルであり、診断の根拠ではありません。記録なしで提出すると RADV 監査に不合格となります

• 誤り： result_value を数値に変換せずに identify_lab_gaps を実行する 正しい方法： pd.to_numeric(errors="coerce") で型変換し、閾値との比較前に NaN を処理します 理由： 検査値にはテキストの修飾子（「>100」、「備考参照」など）が含まれることがよくあります。文字列の比較では誤った結果が生じます