hccinfhir

HCCInFHIR

PyPI version Python 3.9+ License: Apache 2.0

A comprehensive Python library for calculating HCC (Hierarchical Condition Category) risk adjustment scores from healthcare claims data. Supports multiple data sources including FHIR resources, X12 837 claims, X12 834 enrollment files, and direct diagnosis processing.

πŸš€ Quick Start

pip install hccinfhir
from hccinfhir import HCCInFHIR

processor = HCCInFHIR(model_name="CMS-HCC Model V28")

result = processor.calculate_from_diagnosis(["E11.9", "I10", "N18.3"], age=67, sex="F")
print(f"Risk Score: {result.risk_score}")
print(f"HCCs: {result.hcc_list}")

πŸ“‹ Table of Contents

πŸ”„ Migrating from hccpy

HCCInFHIR is the evolution of hccpy. If you’re already using hccpy, the transition is straightforward:

hccpy:

from hccpy.hcc import HCCEngine

he = HCCEngine("28")
rp = he.profile(["E1169", "I5030", "I509", "I211", "I209", "R05"], age=70, sex="M")
print(rp["risk_score"])
print(rp["hcc_lst"])

hccinfhir:

from hccinfhir import HCCInFHIR

processor = HCCInFHIR(model_name="CMS-HCC Model V28")
result = processor.calculate_from_diagnosis(["E1169", "I5030", "I509", "I211", "I209", "R05"], age=70, sex="M")
print(result.risk_score)
print(result.hcc_list)

Why upgrade?

Β  hccpy hccinfhir
Diagnosis-to-RAF Simple and fast Same simplicity, same speed
Input formats Diagnosis codes only FHIR EOB, X12 837, X12 834, diagnosis codes
HCC models V22, V24, V28, ESRD V21 V22, V24, V28, ESRD V21, ESRD V24, RxHCC V08
Dual eligibility Manual elig parameter Auto-detection from 834 enrollment data
Payment adjustments CIF + normalization MACI, normalization, frailty scores
Data quality No workarounds Prefix override for incorrect source data
Custom data files Not supported Full support for custom coefficients, mappings, hierarchies
Output Dictionary Pydantic model (typed, serializable, dict-convertible)

Key differences to note:

✨ Key Features

πŸ“Š Data Sources & Use Cases

1. X12 837 Claims (Professional & Institutional)

2. X12 834 Enrollment Files

3. FHIR ExplanationOfBenefit Resources

4. Direct Diagnosis Codes

πŸ› οΈ Installation

Basic Installation

pip install hccinfhir

Development Installation

git clone https://github.com/yourusername/hccinfhir.git
cd hccinfhir
pip install -e .

Requirements

πŸ“– How-To Guides

Working with CMS Encounter Data (837 Claims)

Scenario: You’re a Medicare Advantage plan processing encounter data for CMS risk adjustment submissions.

from hccinfhir import HCCInFHIR, Demographics
from hccinfhir.extractor import extract_sld

# Step 1: Configure processor
# All data file parameters are optional and default to the latest 2026 valuesets
processor = HCCInFHIR(
    model_name="CMS-HCC Model V28",
    filter_claims=True,  # Apply CMS filtering rules

    # Optional: Override with custom data files (omit to use bundled 2026 defaults)
    # proc_filtering_filename="ra_eligible_cpt_hcpcs_2026.csv",  # CPT/HCPCS codes
    # dx_cc_mapping_filename="ra_dx_to_cc_2026.csv",            # ICD-10 to HCC
    # hierarchies_filename="ra_hierarchies_2026.csv",            # HCC hierarchies
    # is_chronic_filename="hcc_is_chronic.csv",                  # Chronic flags
    # coefficients_filename="ra_coefficients_2026.csv"           # RAF coefficients
)

# Step 2: Load 837 data
with open("encounter_data.txt", "r") as f:
    raw_837_data = f.read()

# Step 3: Extract service-level data
service_data = extract_sld(raw_837_data, format="837")

# Step 4: Define beneficiary demographics
demographics = Demographics(
    age=72,
    sex="M",
    dual_elgbl_cd="00",      # Non-dual eligible
    orec="0",                # Original reason for entitlement
    crec="0",                # Current reason for entitlement
    orig_disabled=False,
    new_enrollee=False,
    esrd=False
)

# Step 5: Calculate risk score
result = processor.run_from_service_data(service_data, demographics)

# Step 6: Review results
print(f"Risk Score: {result.risk_score:.3f}")
print(f"Active HCCs: {result.hcc_list}")
print(f"Disease Interactions: {result.interactions}")
print(f"Diagnosis Mappings:")
for cc, dx_codes in result.cc_to_dx.items():
    print(f"  HCC {cc}: {', '.join(dx_codes)}")

# Export for CMS submission
encounter_summary = {
    "beneficiary_id": "12345",
    "risk_score": result.risk_score,
    "hcc_list": result.hcc_list,
    "model": "V28",
    "payment_year": 2026
}

Processing X12 834 Enrollment for Dual Eligibility

Scenario: You need to extract dual eligibility status from enrollment files to ensure accurate risk scores. This is critical because dual-eligible beneficiaries can receive 30-50% higher RAF scores due to different coefficient prefixes.

Why This Matters:

from hccinfhir import HCCInFHIR, Demographics
from hccinfhir.extractor_834 import (
    extract_enrollment_834,
    enrollment_to_demographics,
    is_losing_medicaid,
    medicaid_status_summary
)

# Step 1: Parse X12 834 enrollment file
with open("enrollment_834.txt", "r") as f:
    x12_834_data = f.read()

enrollments = extract_enrollment_834(x12_834_data)

# Step 2: Process each member
processor = HCCInFHIR(model_name="CMS-HCC Model V28")

for enrollment in enrollments:
    # Convert enrollment to Demographics for RAF calculation
    demographics = enrollment_to_demographics(enrollment)

    print(f"\\n=== Member: {enrollment.member_id} ===")
    print(f"MBI: {enrollment.mbi}")
    print(f"Medicaid ID: {enrollment.medicaid_id}")
    print(f"Dual Status: {enrollment.dual_elgbl_cd}")
    print(f"Full Benefit Dual: {enrollment.is_full_benefit_dual}")
    print(f"Partial Benefit Dual: {enrollment.is_partial_benefit_dual}")

    # Step 3: Check for Medicaid coverage loss (critical for RAF projections)
    if is_losing_medicaid(enrollment, within_days=90):
        print(f"⚠️  ALERT: Member losing Medicaid coverage!")
        print(f"   Coverage ends: {enrollment.coverage_end_date}")
        print(f"   Expected RAF impact: -30% to -50%")

    # Step 4: Get comprehensive Medicaid status
    status = medicaid_status_summary(enrollment)
    print(f"\\nMedicaid Status Summary:")
    print(f"  Has Medicare: {status['has_medicare']}")
    print(f"  Has Medicaid: {status['has_medicaid']}")
    print(f"  Dual Status Code: {status['dual_status']}")
    print(f"  Full Benefit Dual: {status['is_full_benefit_dual']}")
    print(f"  Partial Benefit Dual: {status['is_partial_benefit_dual']}")
    print(f"  Coverage End: {status['coverage_end_date']}")

    # Step 5: Calculate RAF with accurate dual status
    diagnosis_codes = ["E11.9", "I10", "N18.3"]  # From claims
    result = processor.calculate_from_diagnosis(diagnosis_codes, demographics)
    print(f"\\nRAF Score: {result.risk_score:.3f}")

California DHCS Medi-Cal Aid Codes (automatically mapped):

# Full Benefit Dual Aid Codes β†’ dual_elgbl_cd='02' or '04'
'4N', '4P'  # QMB Plus
'5B', '5D'  # SLMB Plus

# Partial Benefit Dual Aid Codes β†’ dual_elgbl_cd='01', '03', or '06'
'4M', '4O'  # QMB Only
'5A', '5C'  # SLMB Only
'5E', '5F'  # QI (Qualifying Individual)

Medicare Status Codes (REF*ABB segment):

'QMBPLUS', 'QMB+'    β†’ '02' (Full Benefit)
'SLMBPLUS', 'SLMB+'  β†’ '04' (Full Benefit)
'QMBONLY', 'QMB'     β†’ '01' (Partial Benefit)
'SLMBONLY', 'SLMB'   β†’ '03' (Partial Benefit)
'QI', 'QI1'          β†’ '06' (Partial Benefit)

Processing Clearinghouse 837 Claims

Scenario: Health plan receiving 837 files from clearinghouses for member risk scoring.

from hccinfhir import HCCInFHIR, Demographics
from hccinfhir.extractor import extract_sld_list

# Configure processor
processor = HCCInFHIR(
    model_name="CMS-HCC Model V28",
    filter_claims=True
)

# Process multiple 837 files
claim_files = ["inst_claims.txt", "prof_claims.txt"]
all_service_data = []

for file_path in claim_files:
    with open(file_path, "r") as f:
        claims_data = f.read()
    service_data = extract_sld_list([claims_data], format="837")
    all_service_data.extend(service_data)

# Member demographics (from enrollment system or 834 file)
demographics = Demographics(
    age=45,
    sex="F",
    dual_elgbl_cd="02",    # Full benefit dual from 834
    orig_disabled=True,
    new_enrollee=False
)

# Calculate risk score
result = processor.run_from_service_data(all_service_data, demographics)

print(f"Member Risk Score: {result.risk_score:.3f}")
print(f"Active HCCs: {result.hcc_list}")
print(f"Total Services: {len(result.service_level_data)}")

Using CMS BCDA API Data

Scenario: Building an application that processes Medicare beneficiary data from the BCDA API.

from hccinfhir import HCCInFHIR, Demographics
import requests

# Configure for BCDA data
processor = HCCInFHIR(
    model_name="CMS-HCC Model V24",  # BCDA typically uses V24
    filter_claims=True,
    dx_cc_mapping_filename="ra_dx_to_cc_2025.csv"
)

# Fetch EOB data from BCDA
# headers = {"Authorization": f"Bearer {access_token}"}
# response = requests.get("https://sandbox.bcda.cms.gov/api/v2/Patient/$export", headers=headers)
# eob_resources = response.json()

# For demo, use sample data
from hccinfhir import get_eob_sample_list
eob_resources = get_eob_sample_list(limit=50)

# Demographics (extract from EOB or enrollment system)
demographics = Demographics(
    age=68,
    sex="M",
    dual_elgbl_cd="00",
    new_enrollee=False,
    esrd=False
)

# Process FHIR data
result = processor.run(eob_resources, demographics)

print(f"Beneficiary Risk Score: {result.risk_score:.3f}")
print(f"HCC Categories: {', '.join(result.hcc_list)}")
print(f"Service Period: {min(svc.service_date for svc in result.service_level_data if svc.service_date)} to {max(svc.service_date for svc in result.service_level_data if svc.service_date)}")

Direct Diagnosis Code Processing

Scenario: Quick HCC mapping validation or research without claims data.

from hccinfhir import HCCInFHIR

processor = HCCInFHIR(model_name="CMS-HCC Model V28")

diagnosis_codes = [
    "E11.9",   # Type 2 diabetes
    "I10",     # Hypertension
    "N18.3",   # CKD stage 3
    "F32.9",   # Depression
    "M79.3"    # Panniculitis
]

# Pass demographics as keyword arguments
result = processor.calculate_from_diagnosis(
    diagnosis_codes,
    age=75, sex="F", dual_elgbl_cd="02"  # Full benefit dual
)

print("=== HCC Risk Analysis ===")
print(f"Risk Score: {result.risk_score:.3f}")
print(f"HCC Categories: {result.hcc_list}")
print(f"\\nDiagnosis Mappings:")
for cc, dx_list in result.cc_to_dx.items():
    print(f"  HCC {cc}: {', '.join(dx_list)}")
print(f"\\nApplied Coefficients:")
for coeff_name, value in result.coefficients.items():
    print(f"  {coeff_name}: {value:.3f}")
if result.interactions:
    print(f"\\nDisease Interactions:")
    for interaction, value in result.interactions.items():
        print(f"  {interaction}: {value:.3f}")

Demographics can also be passed as a Demographics object or a dict β€” all three forms are equivalent:

from hccinfhir import Demographics

# Keyword arguments (simplest)
result = processor.calculate_from_diagnosis(["E11.9"], age=75, sex="F")

# Dictionary
result = processor.calculate_from_diagnosis(["E11.9"], {"age": 75, "sex": "F"})

# Demographics object (full control)
result = processor.calculate_from_diagnosis(["E11.9"], Demographics(age=75, sex="F"))

βš™οΈ Configuration

Supported HCC Models

Model Name Model Years Use Case Supported
"CMS-HCC Model V22" 2024-2025 Community populations βœ…
"CMS-HCC Model V24" 2024-2026 Community populations (current) βœ…
"CMS-HCC Model V28" 2025-2026 Community populations (latest) βœ…
"CMS-HCC ESRD Model V21" 2024-2025 ESRD populations βœ…
"CMS-HCC ESRD Model V24" 2025-2026 ESRD populations βœ…
"RxHCC Model V08" 2024-2026 Part D prescription drug βœ…
"RxHCC Model V08 PDP_AND_MAPD" 2027 (proposed) Part D - Combined reference estimate βœ…
"RxHCC Model V08 PDP_ONLY" 2027 (proposed) Part D - Standalone PDP plans βœ…
"RxHCC Model V08 MAPD_ONLY" 2027 (proposed) Part D - MA-PD plans βœ…

Using Proposed 2027 Coefficients

The library includes proposed CMS coefficients for 2027 payment year (ra_proposed_coefficients_2027.csv). These are useful for:

from hccinfhir import HCCInFHIR, Demographics

# CMS-HCC with proposed 2027 coefficients
processor_2027 = HCCInFHIR(
    model_name="CMS-HCC Model V28",
    coefficients_filename="ra_proposed_coefficients_2027.csv"
)

demographics = Demographics(age=70, sex="M", dual_elgbl_cd="00")
diagnosis_codes = ["E11.9", "I10", "N18.3"]

result = processor_2027.calculate_from_diagnosis(diagnosis_codes, demographics)
print(f"2027 Proposed RAF Score: {result.risk_score:.3f}")

RxHCC Plan-Specific Variants

CMS is introducing plan-specific RxHCC coefficients for 2027, separating standalone PDP and MA-PD plans. The combined PDP_AND_MAPD estimate is also provided as a traditional reference:

# PDP and MA-PD combined (traditional reference estimate)
processor_pdp_mapd = HCCInFHIR(
    model_name="RxHCC Model V08 PDP_AND_MAPD",
    coefficients_filename="ra_proposed_coefficients_2027.csv"
)

# PDP-only plans (standalone Part D)
processor_pdp = HCCInFHIR(
    model_name="RxHCC Model V08 PDP_ONLY",
    coefficients_filename="ra_proposed_coefficients_2027.csv"
)

# MA-PD only plans (Medicare Advantage with Part D)
processor_mapd = HCCInFHIR(
    model_name="RxHCC Model V08 MAPD_ONLY",
    coefficients_filename="ra_proposed_coefficients_2027.csv"
)

# Compare scores across plan types
demographics = Demographics(age=70, sex="F", low_income=True)
diagnosis_codes = ["E11.9"]

for name, proc in [("PDP_AND_MAPD", processor_pdp_mapd),
                   ("PDP_ONLY", processor_pdp),
                   ("MAPD_ONLY", processor_mapd)]:
    result = proc.calculate_from_diagnosis(diagnosis_codes, demographics)
    print(f"{name}: {result.risk_score:.3f}")

Comparing 2026 vs 2027 Coefficients

from hccinfhir import HCCInFHIR, Demographics

# Current 2026 coefficients
processor_2026 = HCCInFHIR(
    model_name="CMS-HCC Model V28",
    coefficients_filename="ra_coefficients_2026.csv"
)

# Proposed 2027 coefficients
processor_2027 = HCCInFHIR(
    model_name="CMS-HCC Model V28",
    coefficients_filename="ra_proposed_coefficients_2027.csv"
)

demographics = Demographics(age=70, sex="M", dual_elgbl_cd="00")
diagnosis_codes = ["E11.9", "I10", "N18.3"]

result_2026 = processor_2026.calculate_from_diagnosis(diagnosis_codes, demographics)
result_2027 = processor_2027.calculate_from_diagnosis(diagnosis_codes, demographics)

print(f"2026 RAF Score: {result_2026.risk_score:.3f}")
print(f"2027 RAF Score: {result_2027.risk_score:.3f}")
print(f"Change: {((result_2027.risk_score / result_2026.risk_score) - 1) * 100:.1f}%")

Note: Proposed coefficients are subject to change. Always verify against final CMS publications for payment calculations.

Custom Data Files

The library includes bundled CMS reference data for 2025 and 2026. You can override all 5 data files with custom versions:

processor = HCCInFHIR(
    model_name="CMS-HCC Model V28",
    filter_claims=True,

    # All files support absolute paths, relative paths, or bundled filenames
    # See "Custom File Path Resolution" in Advanced Features for details

    # 1. CPT/HCPCS Procedure Codes (for CMS filtering)
    proc_filtering_filename="ra_eligible_cpt_hcpcs_2026.csv",

    # 2. Diagnosis to HCC Mapping (ICD-10 β†’ HCC)
    dx_cc_mapping_filename="ra_dx_to_cc_2026.csv",

    # 3. HCC Hierarchies (parent HCCs suppress child HCCs)
    hierarchies_filename="ra_hierarchies_2026.csv",

    # 4. Chronic Condition Flags
    is_chronic_filename="hcc_is_chronic.csv",

    # 5. RAF Coefficients (demographic + HCC + interaction coefficients)
    coefficients_filename="ra_coefficients_2026.csv"
)

πŸ’‘ Tip: For custom file paths (absolute, relative, or current directory), see Custom File Path Resolution in Advanced Features.

File Format Requirements:

  1. proc_filtering (ra_eligible_cpt_hcpcs_2026.csv):
    cpt_hcpcs_code
    99213
    99214
    99215
    
  2. dx_cc_mapping (ra_dx_to_cc_2026.csv):
    diagnosis_code,cc,model_name
    E119,38,CMS-HCC Model V28
    I10,226,CMS-HCC Model V28
    
  3. hierarchies (ra_hierarchies_2026.csv):
    cc_parent,cc_child,model_domain,model_version,model_fullname
    17,18,CMS-HCC,V28,CMS-HCC Model V28
    17,19,CMS-HCC,V28,CMS-HCC Model V28
    
  4. is_chronic (hcc_is_chronic.csv):
    hcc,is_chronic,model_version,model_domain
    1,True,V28,CMS-HCC
    2,False,V28,CMS-HCC
    
  5. coefficients (ra_coefficients_2026.csv):
    coefficient,value,model_domain,model_version
    cna_f70_74,0.395,CMS-HCC,V28
    cna_hcc19,0.302,CMS-HCC,V28
    

πŸ“ Reference: See complete file formats and structure in the bundled data folder: src/hccinfhir/data

Demographics Configuration

from hccinfhir import Demographics

demographics = Demographics(
    # Required fields
    age=67,                    # Age in years
    sex="F",                   # "M" or "F" (also accepts "1" or "2")

    # Dual eligibility (critical for payment accuracy)
    dual_elgbl_cd="00",        # "00"=Non-dual, "01"=Partial, "02"=Full
                               # "03"=Partial, "04"=Full, "05"=QDWI
                               # "06"=QI, "08"=Other full benefit dual

    # Medicare entitlement
    orec="0",                  # Original reason for entitlement
                               # "0"=Old age, "1"=Disability, "2"=ESRD, "3"=Both
    crec="0",                  # Current reason for entitlement

    # Status flags
    orig_disabled=False,       # Original disability (affects category)
    new_enrollee=False,        # New to Medicare (<12 months)
    esrd=False,                # End-Stage Renal Disease (auto-detected from orec/crec)

    # Optional fields
    snp=False,                 # Special Needs Plan
    low_income=False,          # Low-income subsidy (Part D)
    lti=False,                 # Long-term institutionalized
    graft_months=None,         # Months since kidney transplant (ESRD models)
    fbd=False,                 # Full benefit dual (auto-set from dual_elgbl_cd)
    pbd=False,                 # Partial benefit dual (auto-set)

    # Auto-calculated (can override)
    category="CNA"             # Beneficiary category (auto-calculated if omitted)
)

πŸ“š API Reference

Main Classes

HCCInFHIR

Main processor class for HCC risk adjustment calculations.

Initialization:

HCCInFHIR(
    filter_claims: bool = True,
    model_name: ModelName = "CMS-HCC Model V28",
    proc_filtering_filename: str = "ra_eligible_cpt_hcpcs_2026.csv",
    dx_cc_mapping_filename: str = "ra_dx_to_cc_2026.csv",
    hierarchies_filename: str = "ra_hierarchies_2026.csv",
    is_chronic_filename: str = "hcc_is_chronic.csv",
    coefficients_filename: str = "ra_coefficients_2026.csv"
)

Methods:

Demographics

Patient demographic information for risk adjustment.

Key Fields:

RAFResult

Comprehensive risk adjustment calculation results.

Fields:

Utility Functions

from hccinfhir import (
    get_eob_sample,           # Get sample FHIR EOB
    get_837_sample,           # Get sample 837 claim
    get_834_sample,           # Get sample 834 enrollment
    get_eob_sample_list,      # Get multiple EOBs
    get_837_sample_list,      # Get multiple 837s
    list_available_samples,   # List all samples
)

from hccinfhir.extractor import (
    extract_sld,              # Extract from single resource
    extract_sld_list,         # Extract from multiple resources
)

from hccinfhir.extractor_834 import (
    extract_enrollment_834,       # Parse 834 enrollment file
    enrollment_to_demographics,   # Convert to Demographics
    is_losing_medicaid,           # Check Medicaid loss
    medicaid_status_summary,      # Get comprehensive status
)

from hccinfhir.filter import apply_filter  # Apply CMS filtering
from hccinfhir.model_calculate import calculate_raf  # Direct calculation

πŸ”§ Advanced Features

Payment RAF Adjustments

Apply CMS payment adjustments to RAF scores:

from hccinfhir import HCCInFHIR, Demographics

processor = HCCInFHIR(model_name="CMS-HCC Model V28")
demographics = Demographics(age=70, sex="F")
diagnosis_codes = ["E11.9", "I50.22", "N18.3"]

# Apply payment adjustments
result = processor.calculate_from_diagnosis(
    diagnosis_codes,
    demographics,
    maci=0.059,         # MA Coding Intensity Adjustment (5.9% reduction for 2026)
    norm_factor=1.015,  # Normalization factor (1.5% for 2026)
    frailty_score=0.0   # Frailty adjustment (if applicable)
)

print(f"Base RAF Score: {result.risk_score:.3f}")
print(f"Payment RAF Score: {result.risk_score_payment:.3f}")
print(f"Payment Adjustment: {((result.risk_score_payment / result.risk_score) - 1) * 100:.1f}%")

Common Adjustment Values:

Demographic Prefix Override

Problem: Demographic data quality issues leading to incorrect RAF calculations.

Solution: Manually specify the coefficient prefix.

from hccinfhir import HCCInFHIR, Demographics

# ESRD patient with incorrect orec/crec codes
processor = HCCInFHIR(model_name="CMS-HCC ESRD Model V24")
demographics = Demographics(
    age=65,
    sex="F",
    orec="0",  # Should be '2' or '3', but data is wrong
    crec="0"
)
diagnosis_codes = ["N18.6", "E11.22", "I12.0"]

# Force ESRD dialysis coefficients
result = processor.calculate_from_diagnosis(
    diagnosis_codes,
    demographics,
    prefix_override='DI_'  # ESRD Dialysis prefix
)

print(f"RAF Score with override: {result.risk_score:.3f}")

Common Prefix Values:

CMS-HCC Models:

ESRD Models:

RxHCC Models:

See CLAUDE.md for complete reference.

Custom File Path Resolution

The library uses intelligent path resolution to locate data files with the following priority:

  1. Absolute path - If you provide an absolute path, it uses that exact location
  2. Relative to current working directory - Checks ./your_file.csv or ./custom_data/your_file.csv
  3. Bundled package data - Falls back to built-in CMS reference files

This allows flexible deployment scenarios without changing code.

πŸ“ Data File Reference: See the bundled CMS reference files for format examples: src/hccinfhir/data

Basic Examples

from hccinfhir import HCCInFHIR

# Option 1: Use bundled data (default - no setup needed)
processor = HCCInFHIR(
    model_name="CMS-HCC Model V28",
    dx_cc_mapping_filename="ra_dx_to_cc_2026.csv"  # βœ… Loads from package
)

# Option 2: Relative path from current directory
# Assumes: ./custom_data/my_dx_mapping.csv exists
processor = HCCInFHIR(
    model_name="CMS-HCC Model V28",
    dx_cc_mapping_filename="custom_data/my_dx_mapping.csv"  # βœ… ./custom_data/
)

# Option 3: Absolute path (production deployments)
processor = HCCInFHIR(
    model_name="CMS-HCC Model V28",
    dx_cc_mapping_filename="/var/data/cms/dx_mapping_2026.csv"  # βœ… Absolute
)

# Option 4: Mix bundled and custom files
processor = HCCInFHIR(
    model_name="CMS-HCC Model V28",
    dx_cc_mapping_filename="ra_dx_to_cc_2026.csv",  # Bundled default
    coefficients_filename="custom_coefficients.csv"  # Custom from current dir
)

Real-World Scenarios

Scenario 1: Development Environment

# Use bundled files for testing
processor = HCCInFHIR(model_name="CMS-HCC Model V28")

Scenario 2: Custom Coefficients for Research

# Keep standard mappings, customize coefficients
# File: ./research/adjusted_coefficients.csv
processor = HCCInFHIR(
    model_name="CMS-HCC Model V28",
    coefficients_filename="research/adjusted_coefficients.csv"
)

Scenario 3: Production with Centralized Data

# All custom files in shared network location
data_path = "/mnt/shared/cms_data/2026"
processor = HCCInFHIR(
    model_name="CMS-HCC Model V28",
    proc_filtering_filename=f"{data_path}/cpt_hcpcs.csv",
    dx_cc_mapping_filename=f"{data_path}/dx_to_cc.csv",
    hierarchies_filename=f"{data_path}/hierarchies.csv",
    is_chronic_filename=f"{data_path}/chronic_flags.csv",
    coefficients_filename=f"{data_path}/coefficients.csv"
)

Scenario 4: Docker Container with Mounted Volume

# Files mounted at /app/data
processor = HCCInFHIR(
    model_name="CMS-HCC Model V28",
    dx_cc_mapping_filename="/app/data/dx_to_cc_custom.csv",
    coefficients_filename="/app/data/coefficients_custom.csv"
    # Other files use bundled defaults
)

Error Handling

from hccinfhir import HCCInFHIR

try:
    processor = HCCInFHIR(
        model_name="CMS-HCC Model V28",
        dx_cc_mapping_filename="nonexistent.csv"
    )
except FileNotFoundError as e:
    print(f"File not found: {e}")
    # Error shows all locations checked:
    # - Current directory: /path/to/cwd
    # - Package data: hccinfhir.data

Batch Processing

from hccinfhir import HCCInFHIR, Demographics

processor = HCCInFHIR(model_name="CMS-HCC Model V28")

# Process multiple beneficiaries
beneficiaries = [
    {"id": "001", "age": 67, "sex": "F", "dual": "00", "dx": ["E11.9", "I10"]},
    {"id": "002", "age": 45, "sex": "M", "dual": "02", "dx": ["N18.4", "F32.9"]},
    {"id": "003", "age": 78, "sex": "F", "dual": "01", "dx": ["F03.90", "I48.91"]},
]

results = []
for ben in beneficiaries:
    demographics = Demographics(
        age=ben["age"],
        sex=ben["sex"],
        dual_elgbl_cd=ben["dual"]
    )
    result = processor.calculate_from_diagnosis(ben["dx"], demographics)
    results.append({
        "beneficiary_id": ben["id"],
        "risk_score": result.risk_score,
        "risk_score_payment": result.risk_score_payment,
        "hcc_list": result.hcc_list
    })

# Export results
import json
with open("risk_scores.json", "w") as f:
    json.dump(results, f, indent=2)

Large-Scale Processing with Databricks

For processing millions of beneficiaries, use PySpark’s pandas_udf for distributed computation. The hccinfhir logic is well-suited for batch operations with clear, simple transformations.

Performance Benchmark:

Databricks Performance Chart

Tested with ACO data on Databricks Runtime 17.3 LTS, Worker: i3.4xlarge (122GB, 16 cores)

The chart shows execution time varies based on condition complexity - members with more diagnoses require additional internal processing loops. While the relationship isn’t perfectly linear, 1 million members can be processed in under 2 minutes with this configuration.

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, FloatType, ArrayType, StringType
from pyspark.sql import functions as F
from pyspark.sql.functions import pandas_udf
import pandas as pd

from hccinfhir import HCCInFHIR, Demographics

# Define the return schema
hcc_schema = StructType([
    StructField("risk_score", FloatType(), True),
    StructField("risk_score_demographics", FloatType(), True),
    StructField("risk_score_chronic_only", FloatType(), True),
    StructField("risk_score_hcc", FloatType(), True),
    StructField("hcc_list", ArrayType(StringType()), True)
])

# Initialize processor (will be serialized to each executor)
hcc_processor = HCCInFHIR(model_name="CMS-HCC Model V28")

# Create the pandas UDF
@pandas_udf(hcc_schema)
def calculate_hcc(
    age_series: pd.Series,
    sex_series: pd.Series,
    diagnosis_series: pd.Series
) -> pd.DataFrame:
    results = []

    for age, sex, diagnosis_codes in zip(age_series, sex_series, diagnosis_series):
        try:
            demographics = Demographics(age=int(age), sex=sex)

            # diagnosis_codes can be passed directly - accepts any iterable including numpy arrays
            result = hcc_processor.calculate_from_diagnosis(diagnosis_codes, demographics)

            results.append({
                'risk_score': float(result.risk_score),
                'risk_score_demographics': float(result.risk_score_demographics),
                'risk_score_chronic_only': float(result.risk_score_chronic_only),
                'risk_score_hcc': float(result.risk_score_hcc),
                'hcc_list': result.hcc_list
            })
        except Exception as e:
            # Log error and return nulls for failed rows
            print(f"ERROR processing row: {e}")
            results.append({
                'risk_score': None,
                'risk_score_demographics': None,
                'risk_score_chronic_only': None,
                'risk_score_hcc': None,
                'hcc_list': None
            })

    return pd.DataFrame(results)

# Apply the UDF to your DataFrame
# Assumes df has columns: age, patient_gender, diagnosis_codes (array of strings)
df = df.withColumn(
    "hcc_results",
    calculate_hcc(
        F.col("age"),
        F.col("patient_gender"),
        F.col("diagnosis_codes")
    )
)

# Expand the struct into separate columns
df = df.select(
    "*",
    F.col("hcc_results.risk_score").alias("risk_score"),
    F.col("hcc_results.risk_score_demographics").alias("risk_score_demographics"),
    F.col("hcc_results.risk_score_chronic_only").alias("risk_score_chronic_only"),
    F.col("hcc_results.risk_score_hcc").alias("risk_score_hcc"),
    F.col("hcc_results.hcc_list").alias("hcc_list")
).drop("hcc_results")

Performance Tips:

Extended Schema with Demographics:

# Include additional demographic parameters
@pandas_udf(hcc_schema)
def calculate_hcc_full(
    age_series: pd.Series,
    sex_series: pd.Series,
    dual_status_series: pd.Series,
    diagnosis_series: pd.Series
) -> pd.DataFrame:
    results = []

    for age, sex, dual_status, diagnosis_codes in zip(
        age_series, sex_series, dual_status_series, diagnosis_series
    ):
        try:
            demographics = Demographics(
                age=int(age),
                sex=sex,
                dual_elgbl_cd=dual_status if dual_status else "00"
            )
            result = hcc_processor.calculate_from_diagnosis(diagnosis_codes, demographics)

            results.append({
                'risk_score': float(result.risk_score),
                'risk_score_demographics': float(result.risk_score_demographics),
                'risk_score_chronic_only': float(result.risk_score_chronic_only),
                'risk_score_hcc': float(result.risk_score_hcc),
                'hcc_list': result.hcc_list
            })
        except Exception as e:
            results.append({
                'risk_score': None,
                'risk_score_demographics': None,
                'risk_score_chronic_only': None,
                'risk_score_hcc': None,
                'hcc_list': None
            })

    return pd.DataFrame(results)

Converting to Dictionaries

All Pydantic models support dictionary conversion for JSON serialization, database storage, or legacy code:

from hccinfhir import HCCInFHIR, Demographics

processor = HCCInFHIR(model_name="CMS-HCC Model V28")
demographics = Demographics(age=67, sex="F")
result = processor.calculate_from_diagnosis(["E11.9"], demographics)

# Convert to dictionary
result_dict = result.model_dump()
print(result_dict["risk_score"])  # Dictionary access

# JSON-safe conversion
result_json = result.model_dump(mode='json')

# Partial conversion
summary = result.model_dump(include={"risk_score", "hcc_list", "model_name"})

# Exclude large nested data
compact = result.model_dump(exclude={"service_level_data"})

# Convert to JSON string
json_string = result.model_dump_json()

# API response (FastAPI)
from fastapi import FastAPI
app = FastAPI()

@app.post("/calculate")
def calculate_risk(diagnosis_codes: list, demographics: dict):
    demo = Demographics(**demographics)
    result = processor.calculate_from_diagnosis(diagnosis_codes, demo)
    return result.model_dump(mode='json')  # Automatic JSON serialization

πŸ“ Sample Data

Comprehensive sample data for testing and development:

from hccinfhir import (
    get_eob_sample,
    get_837_sample,
    get_834_sample,
    list_available_samples
)

# FHIR EOB samples (3 individual + 200 batch)
eob = get_eob_sample(1)  # Cases 1, 2, 3 (returns single dict)
eob_list = get_eob_sample_list(limit=50)  # Returns list

# Usage: processor.run() expects a list, so wrap single EOB
result = processor.run([eob], demographics)  # Note: [eob] not eob

# X12 837 samples (13 different scenarios)
claim = get_837_sample(0)  # Cases 0-12 (returns string)
claims = get_837_sample_list([0, 1, 2])  # Returns list

# X12 834 enrollment samples (6 CA DHCS scenarios)
enrollment_834 = get_834_sample(1)  # Cases 1-6 available (returns string)

# List all available samples
info = list_available_samples()
print(f"EOB samples: {info['eob_case_numbers']}")
print(f"837 samples: {info['837_case_numbers']}")
print(f"834 samples: {info['834_case_numbers']}")

πŸ§ͺ Testing

# Activate virtual environment
hatch shell

# Install in development mode
pip install -e .

# Run all tests (238 tests)
pytest tests/

# Run specific test file
pytest tests/test_model_calculate.py -v

# Run with coverage
pytest tests/ --cov=hccinfhir --cov-report=html

πŸ“„ License

Apache License 2.0. See LICENSE for details.

πŸ“ž Support

πŸ‘₯ Contributors

We’re grateful to all contributors who have helped improve this project:

Want to contribute? We’re always looking for great minds to contribute to this project! Simply make a PR or open a ticket and we’ll get connected.


Made with ❀️ by the HCCInFHIR team