Automating tier mapping updates from CMS formulary files

Every quarter, CMS publishes a fresh HPMS formulary file, and the single implementation decision that decides whether that release lands cleanly is how you promote it into the tier snapshot the copay engine reads: full-table replace versus streaming delta into a new immutable version. This page owns that decision for the Tier Mapping & Copay Calculation Logic workflow. Get it wrong and an in-flight claim observes a half-applied publish — one member sees a Tier 2 copay while the next sees Non-Formulary for the same drug — which is unreproducible in a payer audit and shows up instantly at the counter as a mispriced fill or a spurious 70 Product/Service Not Covered reject. Get it right and a 500k-line quarterly shift deploys atomically, with a signed snapshot_version any auditor can replay. CMS formulary files carry zero Protected Health Information, but the pipeline still feeds the live adjudication path, so its correctness guarantees are exactly as strict as the claim path it serves.

Ingestion strategy: full reload vs streaming delta #

The naive approach — load the whole file into a pandas DataFrame, truncate the tier table, and bulk-insert — is correct only if nothing reads the table mid-swap. In a live PBM that never holds. The production choice is a streaming delta parse that builds a new versioned snapshot and swaps a pointer once, so readers only ever see version N or version N+1, never a torn state.

The delta path wins on every axis that matters for adjudication correctness. It keeps a lightweight baseline hash map of active tier states, processes only changed records, cuts memory ~85%, and — critically — produces an immutable snapshot rather than mutating a live table. That versioned snapshot is the same artifact Tier Mapping & Copay Calculation Logic reads under a monotonic version, and the same immutability contract the Rule Engine Threshold Tuning & Optimization engine depends on.

CMS HPMS field alignment #

The CMS HPMS formulary file is pipe-delimited and keyed differently from your adjudication schema. The mapping below is the load-bearing contract; a single misaligned column silently poisons every downstream copay calculation.

The National Drug Code arrives in varying widths and must be padded through the same deterministic NDC-to-GPI Crosswalk Automation rules the rest of the engine uses, because tiers are ultimately keyed on the resolved therapeutic-class identifier, not the raw NDC.

Step-by-step implementation #

The pipeline runs in five ordered stages: fetch, stream-normalize, delta-diff against the prior baseline, assemble a signed snapshot, and swap the pointer. The diagram below traces one row through normalization and delta hashing into the versioned tier table.

Figure: CMS formulary ingestion streams rows through TIER_MAP normalization and per-record delta hashing — unchanged rows are skipped, and only changed NDCs are sealed into an immutable, versioned tier snapshot with an atomic pointer swap.

Step 1 — Normalize each field deterministically. Strip and pad the NDC, map the tier literal to an integer, and coerce the CMS Y/N flags. Integer tiers eliminate string-comparison overhead on the real-time claims path.

import csv
import hashlib
import logging
from decimal import Decimal
from typing import Dict, Generator, Optional, Tuple

# CMS HPMS formulary files carry NO PHI (no 302-C2 Cardholder ID, no 310-CA
# Patient First Name). Even so, never echo raw claim bytes into these logs if
# this module is imported into an adjudication worker — keep it formulary-only.
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger(__name__)

# TIER_LEVEL_VALUE literals -> adjudication-native tier integers. Non-formulary
# and specialty variants collapse to their cost-share equivalents.
TIER_MAP = {
    "1": 1, "2": 2, "3": 3, "4": 4, "5": 5,
    "Tier 1": 1, "Tier 2": 2, "Tier 3": 3, "Tier 4": 4, "Tier 5": 5,
    "SP": 4, "Specialty": 4, "NF": 99, "Non-Formulary": 99,
}

def normalize_ndc(raw_ndc: str) -> Optional[str]:
    # 407-D7 Product/Service ID must be exactly 11 digits for the tier lookup.
    cleaned = raw_ndc.replace("-", "").replace(" ", "").strip()
    if not cleaned.isdigit():
        return None
    padded = cleaned.zfill(11)
    return padded if len(padded) == 11 else None

def coerce_flag(value: str) -> bool:
    # PRIOR_AUTHORIZATION_YN / STEP_THERAPY_YN -> bool (drives 75 / 608 rejects).
    return value.strip().upper() == "Y"

Step 2 — Stream rows and yield only deltas. A generator keeps memory O(1) in file size. Each normalized row is hashed and compared to the prior baseline; unchanged rows are dropped before they ever touch the snapshot writer.

def compute_record_hash(ndc: str, tier: int, pa: bool, st: bool, ql: int) -> str:
    payload = f"{ndc}|{tier}|{int(pa)}|{int(st)}|{ql}"
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()

def parse_cms_formulary(
    file_path: str,
    baseline: Dict[str, dict],
) -> Generator[Tuple[str, dict], None, None]:
    """Stream a CMS HPMS formulary file, normalize fields, yield only tier deltas.

    Feeds the versioned snapshot that Tier Mapping & Copay Calculation Logic
    reads. Operates in O(1) memory relative to file size.
    """
    with open(file_path, "r", encoding="utf-8-sig", newline="") as f:
        reader = csv.DictReader(f, delimiter="|")   # CMS HPMS is pipe-delimited
        for row in reader:
            ndc = normalize_ndc(row.get("NDC", ""))
            if not ndc:                              # unparseable NDC -> skip, log at DEBUG
                logger.debug("Dropped row with unparseable NDC")
                continue

            tier = TIER_MAP.get(row.get("TIER_LEVEL_VALUE", "").strip(), 99)
            pa = coerce_flag(row.get("PRIOR_AUTHORIZATION_YN", "N"))
            st = coerce_flag(row.get("STEP_THERAPY_YN", "N"))
            ql = int(row.get("QUANTITY_LIMIT_AMOUNT", "0") or "0")

            record_hash = compute_record_hash(ndc, tier, pa, st, ql)
            if ndc in baseline and baseline[ndc]["hash"] == record_hash:
                continue                             # unchanged -> not a delta

            yield ndc, {
                "ndc": ndc,                          # 407-D7
                "tier": tier,                        # -> 505-F5 Patient Pay Amount
                "pa_required": pa,                   # -> 75 Prior Auth Required
                "st_required": st,                   # -> 608 Step Therapy Required
                "quantity_limit": ql,                # -> 76 Plan Limitations Exceeded
                "hash": record_hash,
            }

Step 3 — Seal a versioned snapshot and swap atomically. Apply the deltas onto the prior baseline, stamp a monotonic version and a content digest, and swap the reader pointer in one operation. Copay-affecting values that ride the snapshot — flat copays, caps — are stored as Decimal strings, never float, so cost-share math downstream stays exact.

import json
import time
from pathlib import Path

def seal_snapshot(
    baseline: Dict[str, dict],
    deltas: Dict[str, dict],
    formulary_id: str,
    prior_version: int,
    out_dir: str,
) -> dict:
    """Merge deltas into an immutable, content-addressed tier snapshot."""
    merged = {**baseline, **deltas}                  # deltas win per NDC
    version = prior_version + 1                       # monotonic, audit-replayable

    # Copay caps stored as Decimal STRINGS — decimal.Decimal, never float, so
    # 505-F5 Patient Pay Amount reconstructs exactly on read.
    body = {
        ndc: {**rec, "copay_cap": str(Decimal(rec.get("copay_cap", "0")))}
        for ndc, rec in merged.items()
    }
    canonical = json.dumps(body, sort_keys=True, separators=(",", ":"))
    digest = hashlib.sha256(canonical.encode("utf-8")).hexdigest()

    snapshot = {
        "formulary_id": formulary_id,
        "version": version,
        "sealed_at": int(time.time()),
        "content_digest": digest,
        "record_count": len(body),
        "delta_count": len(deltas),
        "records": body,
    }

    dst = Path(out_dir) / f"tier_snapshot_v{version}.json"
    tmp = dst.with_suffix(".tmp")
    tmp.write_text(json.dumps(snapshot))
    tmp.replace(dst)                                 # atomic rename == the swap
    (Path(out_dir) / "current").write_text(str(version))
    logger.info("Sealed formulary %s v%d: %d records, %d deltas, digest=%s",
                formulary_id, version, len(body), len(deltas), digest[:12])
    return snapshot

Step 4 — Push deltas to the hot read layer. Load only the changed NDCs into the Redis-backed or DynamoDB tier cache the adjudication workers query, using the new version as a namespace so a worker never mixes rows across snapshots. Warm the cache before flipping current, so the first claim under version N+1 hits a fully populated table.

Verification and testing pattern #

Correctness here is verified against known CMS fixtures: a small pinned formulary file whose expected tier integers, PA/ST flags, and delta counts you assert exactly. The two properties that must hold are (a) idempotency — re-ingesting the same file yields zero deltas — and (b) tier fidelity — every mapped integer matches the CMS TIER_LEVEL_VALUE.

import pytest

FIXTURE = "NDC|TIER_LEVEL_VALUE|PRIOR_AUTHORIZATION_YN|STEP_THERAPY_YN|QUANTITY_LIMIT_AMOUNT\n" \
          "00093-7146-01|2|N|N|0\n" \
          "12345-6789|NF|N|N|0\n" \
          "00071-0155-23|SP|Y|Y|30\n"

def test_ndc_padding_and_tier_map(tmp_path):
    f = tmp_path / "cms.txt"
    f.write_text(FIXTURE)
    out = {ndc: rec for ndc, rec in parse_cms_formulary(str(f), baseline={})}

    assert out["00093714601"]["tier"] == 2          # 407-D7 padded to 11 digits
    assert out["00012345678"]["tier"] == 99         # NF -> Non-Formulary (70 reject)
    assert out["00071015523"]["tier"] == 4          # SP -> Specialty cost-share
    assert out["00071015523"]["pa_required"] is True   # -> 75 Prior Auth Required
    assert out["00071015523"]["st_required"] is True   # -> 608 Step Therapy Required

def test_reingest_is_idempotent(tmp_path):
    f = tmp_path / "cms.txt"
    f.write_text(FIXTURE)
    baseline = {ndc: rec for ndc, rec in parse_cms_formulary(str(f), baseline={})}
    # Second pass against the same baseline must produce NO deltas.
    deltas = list(parse_cms_formulary(str(f), baseline=baseline))
    assert deltas == []

Beyond unit fixtures, run a pre-flight suite against the sealed snapshot before flipping current: confirm mapped tier integers correspond to active copay group IDs (mismatches throw 75), that PRIOR_AUTHORIZATION_YN / STEP_THERAPY_YN align with the Step Therapy & Prior Auth Trigger Rules routing tables, and that overlapping effective windows resolve with a GREATEST(EFF_DT)/LEAST(END_DT) strategy so no drug routes to two tiers at once. Cross-reference quantity limits against the Quantity Limit & Days Supply Validation rules that consume them.

Gotchas and PHI guardrails #

• Never mutate the live table. A truncate-and-reload leaves a window where readers see a half-applied formulary. Always seal a new immutable version and swap the current pointer — the whole reason tier assignments use the versioned snapshot pattern is payer-audit replay.
• float copay caps drift. A specialty plan cap of 0.1 + 0.2 will not equal 0.3, and a member can be pushed a cent over their 505-F5 Patient Pay Amount. Store and compute every monetary value with decimal.Decimal.
• Silent NDC padding failures. A NDC that fails .isdigit() (embedded *, alphanumeric placeholder) is dropped — count and alert on drop rate, because a spike means an upstream format change, not clean data.
• Overlapping effective windows. CMS occasionally publishes overlapping EFF_DT/END_DT for the same NDC; without explicit resolution the drug dual-routes and produces inconsistent copays across adjacent claims.
• PHI leakage on shared logging. Formulary files hold no PHI, but this module is often imported into adjudication workers. Keep its logger formulary-scoped and never let it interpolate a claim payload — no 302-C2 Cardholder ID or 310-CA Patient First Name may ever reach these logs. Strip and redact per HIPAA minimum-necessary if the boundaries ever blur.
• Rejection-rate watch. Track NCPDP 75 and 70 rates for 24 hours after each swap and alert on >0.5% deviation from the prior version’s baseline — a tier that mapped wrong shows up as a reject spike before members feel it.

Related #

• Tier Mapping & Copay Calculation Logic — the engine that reads the sealed snapshot to compute 505-F5 Patient Pay Amount.
• Rule Engine Threshold Tuning & Optimization — the versioned-config discipline this ingestion feeds.
• Quantity Limit & Days Supply Validation — consumes the quantity-limit fields normalized here.
• Step Therapy & Prior Auth Trigger Rules — routes the PA/ST flags this pipeline coerces.
• NDC-to-GPI Crosswalk Automation — the identifier resolution that makes tier lookups trustworthy.

← Back to Tier Mapping & Copay Calculation Logic