risk-adjustment

风险调整 — Pipeline 技能

概述

为 CMS-HCC 风险调整提供确定性、可直接复制粘贴的 Python 和 SQL 代码： ICD-10 到 HCC 交叉映射、层级解析、RAF 分数计算，以及通过 Rx/实验室代理指标识别编码缺口。

用法

• 以编程方式应用 ICD-10 到 HCC 交叉映射并解析疾病层级
• 计算成员级 RAF 分数，并通过 Rx/实验室代理指标识别编码缺口

核心概念

---

响应格式

• 先给出用户需要的命令或代码，再作解释
• 按以下顺序组织：确认输入 → 可用代码 → 说明关键参数 → 注意事项
• 每项任务提供一个完整可用示例；不要展示所有替代方案
• 代码注释保持精简且实用（说明作用，而非存在原因）
• 目标：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. 端到端 Pipeline

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 到 HCC 交叉映射前，未去除 ICD-10 代码中的句点 正确做法： 在与交叉映射表联接前，使用 str.replace(".", "").upper() 规范化代码 原因： CMS 交叉映射文件使用未格式化的代码（例如 E119，而非 E11.9）；不匹配会导致有效映射被静默丢弃

• 错误： 未先解析层级，就累加所有 CC 系数 正确做法： 评分前，始终调用 resolve_hierarchies()，仅保留每个层级组中严重程度最高的 CC 原因： 计入被更高级别 CC 取代的 CC 会抬高 RAF 分数，并导致收入预测不准确

• 错误： 对支付年份 2026 及以后使用 V24 层级和系数 正确做法： PY 2026 及以后使用 V28 表；PY 2025 使用混合权重（33% V24 / 67% V28） 原因： CMS 将于 2026 年全面采用 V28 权重；过时的系数会导致分数出现实质性偏差

• 错误： 将任何类型提供者的诊断都纳入交叉映射 正确做法： 映射前筛选符合资格的提供者类型（MD、DO、NP、PA、CNS） 原因： 对于风险调整，CMS 仅接受来自与符合资格的提供者进行面对面就诊的诊断

• 错误： 将 Rx 代理指标缺口视为确诊，并直接据此编码 正确做法： 使用 Rx/实验室代理指标识别疑似缺口，这些缺口需要在符合资格的就诊中由提供者提供文档记录 原因： 代理指标是筛查信号，并非诊断来源——缺少文档记录而提交会导致 RADV 审计不通过

• 错误： 未先将 result_value 转换为数值，就运行 identify_lab_gaps 正确做法： 使用 pd.to_numeric(errors="coerce") 进行转换，并在阈值比较前处理 NaN 原因： 实验室数值常包含文本限定符（如“>100”“见备注”）；字符串比较会产生错误结果