Risk Adjustment

Risk Adjustment — Pipeline Skill

Overview

Provide deterministic, copy-paste-ready Python and SQL code for CMS-HCC risk adjustment: ICD-10-to-HCC crosswalk, hierarchy resolution, RAF score calculation, and coding gap identification from Rx/lab proxies.

Usage

• Apply ICD-10-to-HCC crosswalk and resolve disease hierarchies programmatically
• Calculate member-level RAF scores and identify coding gaps from Rx/lab proxies

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

1. ICD-10-to-HCC Crosswalk

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. Hierarchy Resolution

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 Score Calculation

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. Coding Gap Identification

4a. Rx Proxy Gaps

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. Lab Proxy Gaps

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": "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"] 100", "see note"); string comparison produces wrong results

## Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [awslabs](https://github.com/awslabs)
- **Source:** [awslabs/hcls-agent-skills](https://github.com/awslabs/hcls-agent-skills)
- **License:** MIT-0

Install and usage instructions live in the source repository linked above.