Building step therapy logic gates in Python adjudication scripts

The implementation decision on this page is narrow but consequential: how to model a step therapy sequence as a deterministic logic gate inside a Python adjudication script, rather than as a passive formulary dictionary lookup. A step therapy rule says a member must have tried and failed one or more preferred agents before a higher-tier drug is authorized, so the gate has to walk a member’s prior utilization in chronological order and decide whether the current 407-D7 Product/Service ID (NDC) is the next legitimate step. Get the state model wrong and the engine either approves an out-of-sequence fill (plan leakage and a clinical-safety miss) or rejects a member who genuinely completed the sequence (an abandoned prescription at the counter). This gate runs downstream of Step Therapy & Prior Auth Trigger Rules evaluation and inside the broader Formulary Validation & Rule Engine Design pipeline, so it must stay deterministic, replayable against a versioned formulary snapshot, and clear of PHI in every log line while holding a sub-50ms per-claim budget.

Modeling Decision: How to Represent the Step Sequence

The core choice is how to hold and evaluate the prior-utilization history. Three approaches dominate production adjudication scripts, and they differ sharply in latency, memory, and correctness under concurrency.

Approach How it works NCPDP fields read Latency (p99) Memory profile Correctness risk
Full-history DataFrame load Load the member’s entire claim history into pandas, filter and sort in-frame 302-C2, 401-D1, 407-D7 120–400ms High — heap fragmentation, GC pauses breach SLA Low logic risk, high SLA risk
Stateful linear sequence walk Stream a windowed slice of priors, walk them once against an ordered step list 302-C2, 401-D1, 407-D7 8–25ms Flat — only the windowed slice materializes Low, if the window and sort are correct
Precompiled step-index map Pre-hash the ordered sequence to a {ndc: index} map, advance a cursor per prior 302-C2, 401-D1, 407-D7 6–18ms Flat, plus a tiny static map Low; O(1) membership, but needs NDC normalization

The stateful linear walk is the right default: it is O(n) over a bounded window, holds a flat memory profile, and produces an order-independent verdict because the sequence order is explicit in the step list rather than implicit in load order. The precompiled index map is a micro-optimization on top of it for very long sequences. Everything below implements the linear walk with an index cursor, keyed on GPI-resolved therapeutic class rather than raw NDC — the 407-D7 NDC is normalized and mapped through the NDC-to-GPI Crosswalk Automation pipeline so that therapeutic equivalents count as the same step.

Step-therapy logic-gate decision flow A claim enters the gate. First the 401-D1 Date of Service is validated: an unparseable date is a hard reject with NCPDP code 15 (M/I Date of Service). A valid date proceeds to filter the member's prior claims to the lookback window. If there is no prior utilization the claim rejects with NCPDP 608 (Step Therapy Required). Otherwise the gate walks the required steps in chronological order, advancing a cursor on each 407-D7 NDC match. If no required step matched, the claim rejects with 608. If steps matched, the gate checks whether the current NDC is the next required step: if not, it rejects out of sequence with 608; if so, it approves with status 00 (Step Therapy Met). Claim in Valid date of service? Filter priors to lookback window reads 401-D1 Date of Service Any prior utilization? Walk required steps in order advance cursor on 407-D7 match Any required step matched? Current NDC is next required step? Approve — Step Therapy Met status 00 Reject — M/I Date of Service reject code 15 Reject — Step Therapy Required reject code 608 · no prior Reject — Step Therapy Required reject code 608 · no step met Reject — Out of Sequence reject code 608 yes yes yes yes no no no no

Figure: Step-sequence logic gate walking required steps in chronological order — an unparseable 401-D1 date rejects with NCPDP 15, while missing or out-of-sequence utilization rejects with 608 and a completed sequence approves with status 00.

Step-by-Step Implementation

The gate below is built in four moves: PHI-safe field extraction, windowed prior filtering, the sequence walk with a cursor, and a reject-code mapping. Each move is annotated with the NCPDP field it touches and the HIPAA constraint it honors.

1. Extract only the fields the gate needs, and hash the identifier immediately. A step therapy gate never needs the full claim payload. Pull 407-D7 (NDC), 401-D1 (Date of Service), and 302-C2 (Cardholder ID). The Cardholder ID is the only PHI element, and it is hashed at ingress for audit correlation so no raw member identifier ever reaches a log sink — the HIPAA minimum-necessary boundary enforced across Security & Compliance Boundaries for Claims Data.

2. Filter priors to the active clinical window before doing any work. Step therapy sequences are only valid inside a lookback window (commonly 90–365 days). Discarding expired records before the walk keeps the working set bounded and the memory profile flat.

3. Walk the ordered sequence with a cursor. Sort the windowed priors by 401-D1 and advance an index only when a prior matches the required step at the cursor. The cursor position after the walk is the member’s progress through the sequence.

4. Map the verdict to an exact NCPDP reject code. A completed sequence approves with 00; a missing or out-of-order sequence rejects with 608 (Step Therapy Required); an unparseable date rejects with 15 (M/I Date of Service). Precise codes matter — mismatched rejects break pharmacy-side messaging, mirroring the discipline in Schema Validation & Error Categorization.

python
import datetime
import hashlib
from typing import Dict, List, Optional, Iterator, Callable


def hash_cardholder(raw_id: str) -> str:
    # 302-C2 Cardholder ID is PHI. Hash at ingress; never log the raw value.
    return hashlib.sha256(raw_id.encode("utf-8")).hexdigest()[:16] if raw_id else ""


class STAdjudicationGate:
    __slots__ = ("st_sequence", "lookback_days", "reject_codes")

    def __init__(self, st_sequence: List[str], lookback_days: int = 365):
        # st_sequence holds GPI-resolved step keys (from 407-D7 NDC crosswalk),
        # so therapeutic equivalents collapse to a single step.
        self.st_sequence = st_sequence
        self.lookback_days = lookback_days
        self.reject_codes = {
            "NO_PRIOR": "608",       # NCPDP 608: Step Therapy Required
            "SEQUENCE_FAIL": "608",  # NCPDP 608: Step Therapy Required
            "INVALID_DOS": "15",     # NCPDP 15: M/I Date of Service
        }

    @staticmethod
    def _parse_dos(raw_date: str) -> Optional[datetime.date]:
        # 401-D1 Date of Service transmits as CCYYMMDD; parse in UTC context.
        if not raw_date:
            return None
        try:
            return datetime.datetime.strptime(raw_date, "%Y%m%d").date()
        except ValueError:
            return None

    def evaluate(self, claim: Dict[str, str], prior_claims: List[Dict[str, str]]) -> Dict[str, str]:
        current_step = claim.get("407-D7")          # Product/Service ID (NDC), pre-normalized
        member_hash = hash_cardholder(claim.get("302-C2", ""))  # PHI hashed, never raw
        dos_date = self._parse_dos(claim.get("401-D1", ""))

        if not dos_date:
            return {"status": "REJECT", "code": self.reject_codes["INVALID_DOS"],
                    "msg": "INVALID_DATE_OF_SERVICE", "member": member_hash}

        cutoff = dos_date - datetime.timedelta(days=self.lookback_days)

        # Window-filter priors for this member; expired records are dropped early.
        valid_priors = [
            c for c in prior_claims
            if (parsed := self._parse_dos(c.get("401-D1"))) and parsed >= cutoff
        ]
        valid_priors.sort(key=lambda x: self._parse_dos(x.get("401-D1")))

        if not valid_priors:
            return {"status": "REJECT", "code": self.reject_codes["NO_PRIOR"],
                    "msg": "NO_PRIOR_UTILIZATION", "member": member_hash}

        # Walk the ordered sequence with a cursor; advance only on the next required step.
        prior_steps = [c.get("407-D7") for c in valid_priors]
        cursor = 0
        for step in prior_steps:
            if cursor < len(self.st_sequence) and step == self.st_sequence[cursor]:
                cursor += 1

        if cursor == 0:
            return {"status": "REJECT", "code": self.reject_codes["SEQUENCE_FAIL"],
                    "msg": "STEP_SEQUENCE_VIOLATION", "member": member_hash}

        if cursor < len(self.st_sequence) and current_step != self.st_sequence[cursor]:
            return {"status": "REJECT", "code": self.reject_codes["SEQUENCE_FAIL"],
                    "msg": "OUT_OF_SEQUENCE", "member": member_hash}

        return {"status": "APPROVE", "code": "00", "msg": "STEP_THERAPY_MET", "member": member_hash}


def stream_adjudication(
    claims_stream: Iterator[Dict[str, str]],
    cache_lookup: Callable[[str], List[Dict[str, str]]],
    st_sequence: List[str],
) -> Iterator[Dict[str, str]]:
    """Generator wrapper for memory-efficient batch adjudication."""
    gate = STAdjudicationGate(st_sequence=st_sequence, lookback_days=365)
    for claim in claims_stream:
        # cache_lookup returns a time-partitioned prior-claim slice, not full history.
        priors = cache_lookup(claim.get("302-C2", ""))  # 302-C2 keys the cache; result excludes PHI
        yield gate.evaluate(claim, priors)

Note the gate is stateless and side-effect free — the only mutable state is the local cursor, which makes it thread-safe under the concurrent evaluation model used across the rule engine. Because the ordered st_sequence is injected (read from a versioned formulary snapshot), a sequence change is a config change, not a redeploy, and any historical claim can be replayed against the exact sequence that was live at adjudication time.

Verification and Testing Pattern

Correctness here means the gate returns the same verdict for the same inputs and never approves an out-of-sequence fill. Drive it against fixtures that encode known NCPDP field values — one fixture per branch of the decision tree.

python
import pytest

SEQUENCE = ["GPI_STEP1", "GPI_STEP2", "GPI_STEP3"]  # ordered, GPI-resolved


def dos(days_ago: int) -> str:
    d = datetime.date(2026, 7, 1) - datetime.timedelta(days=days_ago)
    return d.strftime("%Y%m%d")  # 401-D1 CCYYMMDD


@pytest.fixture
def gate():
    return STAdjudicationGate(st_sequence=SEQUENCE, lookback_days=365)


def test_completed_sequence_approves(gate):
    claim = {"407-D7": "GPI_STEP3", "302-C2": "M001", "401-D1": dos(0)}
    priors = [
        {"407-D7": "GPI_STEP1", "302-C2": "M001", "401-D1": dos(120)},
        {"407-D7": "GPI_STEP2", "302-C2": "M001", "401-D1": dos(60)},
    ]
    result = gate.evaluate(claim, priors)
    assert result["status"] == "APPROVE"
    assert result["code"] == "00"


def test_out_of_sequence_rejects_608(gate):
    # Member jumps to STEP3 with only STEP1 on file -> out of sequence.
    claim = {"407-D7": "GPI_STEP3", "302-C2": "M001", "401-D1": dos(0)}
    priors = [{"407-D7": "GPI_STEP1", "302-C2": "M001", "401-D1": dos(30)}]
    result = gate.evaluate(claim, priors)
    assert result["status"] == "REJECT"
    assert result["code"] == "608"


def test_expired_prior_is_ignored(gate):
    # STEP1 fill sits outside the 365-day window -> treated as no prior.
    claim = {"407-D7": "GPI_STEP2", "302-C2": "M001", "401-D1": dos(0)}
    priors = [{"407-D7": "GPI_STEP1", "302-C2": "M001", "401-D1": dos(400)}]
    result = gate.evaluate(claim, priors)
    assert result["msg"] == "NO_PRIOR_UTILIZATION"


def test_bad_dos_rejects_15(gate):
    claim = {"407-D7": "GPI_STEP1", "302-C2": "M001", "401-D1": "20261301"}  # invalid month
    assert gate.evaluate(claim, [])["code"] == "15"


def test_verdict_is_deterministic_under_prior_reordering(gate):
    # Shuffling prior input order must not change the verdict — the walk sorts by 401-D1.
    claim = {"407-D7": "GPI_STEP3", "302-C2": "M001", "401-D1": dos(0)}
    a = [{"407-D7": "GPI_STEP1", "302-C2": "M001", "401-D1": dos(120)},
         {"407-D7": "GPI_STEP2", "302-C2": "M001", "401-D1": dos(60)}]
    assert gate.evaluate(claim, a)["status"] == gate.evaluate(claim, list(reversed(a)))["status"]

The last test is the load-bearing one: it pins the order-independence property that separates a correct gate from a lookup that silently depends on how the cache returned rows.

Gotchas and PHI Guardrails

  • NDC 11-to-10 digit drift. Inbound 407-D7 fields arrive from legacy switch transmissions with inconsistent zero-padding, so 00093-0058-01 and 93-58-1 name the same product but fail an equality check. Normalize every NDC to 11-digit zfill form and resolve through the crosswalk before it reaches st_sequence; the gate compares GPI-resolved keys, never raw NDC strings.
  • Repeating the same step. A member who refills STEP1 twice should not advance the cursor twice. The cursor only advances when the prior equals the step at the current index, so consecutive STEP1 fills leave the cursor at 1 — verify this explicitly with a fixture, because a naive count(ndc in sequence) model gets it wrong.
  • Overlapping prescribers. Concurrent therapy from two prescribers (411-DB) can inject duplicate step fills on the same date and produce a false out-of-sequence reject. Deduplicate priors on (GPI, 401-D1) before the sort.
  • Timezone drift on the window boundary. 401-D1 carries no timezone; compute the lookback cutoff in UTC so a fill on the 365th day is not silently dropped by a local-time off-by-one. See Python’s datetime module documentation for timezone-aware handling when reconciling against EHR feeds.
  • Cache stampede on high-volume members. The cache_lookup in the streaming wrapper must use jittered TTLs (3600 + random.randint(0, 300)) and fall back to a cold time-partitioned read on a miss; never block the adjudication thread on a synchronous rebuild, and route persistent backend failure through Fallback Routing Logic Design rather than defaulting to APPROVE.
  • PHI in log output — the most common defect. Log only the hashed member token, the transaction ID, the outcome, and the reject code. A raw 302-C2 or a full claim dict in a debug line is a reportable breach. Every verdict dict above carries member as a truncated hash, never the raw identifier, and raw claim bytes never persist to stdout or structured logs.

Once a claim clears this gate, the verdict feeds threshold calibration in Rule Engine Threshold Tuning & Optimization and member cost-sharing in Tier Mapping & Copay Calculation Logic; for standard field definitions consult the official NCPDP Standards documentation.

← Back to Step Therapy & Prior Auth Trigger Rules