Generating Rebate Invoices from Adjudicated Claims

The decision this page pins down is exact: at what grain do you group an adjudicated, accrual-finalized claim population into manufacturer invoice lines, how do you sum 442-E7 Quantity Dispensed and apply the rebate rate so the totals are bit-for-bit stable across runs, and precisely which member fields must be stripped before the utilization file leaves the boundary. Get the grain wrong and a manufacturer disputes lines it cannot foot against its own contract; sum units as float and two runs of the same job disagree by a penny; leave one PHI field in the export and a routine billing file becomes a reportable disclosure. This guide implements the aggregation inside the builder described in Rebate Invoice Generation, keyed on the GPI resolved by the NDC-to-GPI crosswalk and priced on the snapshot from Manufacturer Rebate Contract Modeling.

The Aggregation Decision #

The grain is a negotiated choice, not a default. A manufacturer contract is written to be validated at a specific level, and the invoice must aggregate to exactly that level — no finer (which over-discloses and inflates line count) and no coarser (which hides the detail a manufacturer needs to accept the bill). The export format is a second, independent decision: what the utilization file carries and, critically, what it omits.

The default is contract + GPI + period, with the per-claim detail retained only in an internal, access-controlled dispute file. That split — coarse aggregate out, tokenized detail in — is what makes a line dispute-ready without disclosing PHI, the disclosure boundary enforced across Security & Compliance Boundaries for Claims Data.

Two subtleties decide the grain in practice. First, the GPI must be the resolved 14-digit therapeutic identifier, not the raw 407-D7 NDC — the rebate is negotiated on therapeutic class, so two package NDCs of the same product roll into one line. That resolution happens upstream in the NDC-to-GPI crosswalk; if it is skipped, the invoice fragments into per-package lines a manufacturer cannot foot against a class-level contract. Second, the period must never straddle a contract-version boundary. If a contract amends mid-quarter, a claim dispensed before the amendment and one dispensed after carry different snapshot versions and different rates, and folding them into a single line is a silent mispricing. The selection step below rejects that skew outright rather than averaging across it — the effective-dating discipline detailed in handling rebate contract effective dating.

The finer grains in the matrix exist for real contract shapes — package-specific terms need the NDC grain, channel differentials need pharmacy in the key — but each finer grain trades disclosure risk for detail. Adding 201-B1 Service Provider ID makes the pharmacy channel visible, which is quasi-identifying for low-volume specialty pharmacies, so a channel-grain export needs a suppression rule for small cells. The per-claim grain never ships externally at all; it is the internal dispute artifact, and it stays behind the access boundary even in tokenized form.

Figure: The eligible claim set is netted and deduplicated, grouped to the contract-GPI-period grain, summed and priced in Decimal, then split into a PHI-stripped manufacturer export and an in-boundary tokenized dispute file.

Step-by-Step Implementation #

1. Select the eligible claims. Keep only accrual-finalized claims in the billing period, net out reversals, and deduplicate on the tokenized reference so no claim is counted twice. PHI is dropped at this gate — the selected rows carry 442-E7 units, the resolved contract_id, GPI, and rate, and an opaque token, never 302-C2 or 310-CA.

from decimal import Decimal, ROUND_HALF_UP
from collections import defaultdict

CENTS = Decimal("0.01")
UNIT_Q = Decimal("0.001")


def select_eligible(raw_claims: list[dict], period: str, snapshot_version: str) -> list[dict]:
    """Keep finalized, in-period, non-reversed claims; strip PHI; dedupe."""
    seen: set[str] = set()
    eligible: list[dict] = []
    for c in raw_claims:
        if c["accrual_state"] != "FINALIZED" or c["period"] != period:
            continue
        if c["snapshot_version"] != snapshot_version:
            continue  # snapshot skew cannot share one bill
        if c.get("reversed"):
            continue  # 407-D7 fill reversed -> excluded, netted next cycle
        token = c["claim_token"]           # tokenized 402-D2, non-PHI
        if token in seen:
            continue                        # duplicate inclusion guard
        seen.add(token)
        eligible.append({
            "claim_token": token,
            "contract_id": c["contract_id"],
            "gpi": c["gpi"],                # 407-D7 resolved to 14-digit GPI
            "period": c["period"],
            "units": Decimal(str(c["units_442e7"])),  # 442-E7 as Decimal, never float
            "rate": Decimal(str(c["rebate_rate"])),
            # NOTE: 302-C2 Cardholder ID and 310-CA Patient Name are absent by design.
        })
    return eligible

2. Aggregate units in Decimal at the chosen grain. Group by (contract_id, gpi, period), sum 442-E7 as Decimal, and capture the rate per group while asserting it is consistent — two rates in one group is a resolution defect, not something to average.

def aggregate(eligible: list[dict]) -> list[dict]:
    units: dict[tuple, Decimal] = defaultdict(lambda: Decimal("0"))
    rates: dict[tuple, Decimal] = {}
    counts: dict[tuple, int] = defaultdict(int)
    for c in eligible:
        key = (c["contract_id"], c["gpi"], c["period"])
        units[key] += c["units"]
        counts[key] += 1
        prior = rates.setdefault(key, c["rate"])
        if prior != c["rate"]:
            raise ValueError(f"rate conflict for {key}")  # effective-dating defect

    lines: list[dict] = []
    for key in sorted(units):               # sort -> deterministic file order
        contract_id, gpi, period = key
        summed = units[key].quantize(UNIT_Q, rounding=ROUND_HALF_UP)
        rate = rates[key]
        amount = (summed * rate).quantize(CENTS, rounding=ROUND_HALF_UP)
        lines.append({
            "contract_id": contract_id, "gpi": gpi, "period": period,
            "units": summed, "rebate_rate": rate, "rebate_amount": amount,
            "claim_count": counts[key],
        })
    return lines

3. Assign an idempotent invoice number. Derive it deterministically so a retried run reproduces the same bill instead of double-billing.

import hashlib


def invoice_number(manufacturer_id: str, period: str, snapshot_version: str) -> str:
    key = f"{manufacturer_id}|{period}|{snapshot_version}".encode("utf-8")
    return f"RBT-{manufacturer_id}-{period}-{hashlib.sha256(key).hexdigest()[:12].upper()}"

4. Strip PHI on the way out. The utilization export is built from aggregate line fields only. The claim_token and every member field are excluded; Decimal values cross the wire as strings so no float coercion reintroduces drift.

def to_utilization_export(inv_no: str, lines: list[dict]) -> list[dict]:
    return [{
        "invoice_number": inv_no,
        "contract_id": ln["contract_id"],
        "gpi": ln["gpi"],
        "period": ln["period"],
        "units": str(ln["units"]),
        "rebate_rate": str(ln["rebate_rate"]),
        "rebate_amount": str(ln["rebate_amount"]),
        "claim_count": ln["claim_count"],
        # no claim_token, no 302-C2, no 310-CA -- aggregate detail only
    } for ln in lines]

Verification and Testing Pattern #

Correctness here is two properties: totals are stable regardless of claim input order, and the export carries no PHI or per-claim field. Pin both explicitly.

import pytest

CLAIMS = [
    {"accrual_state": "FINALIZED", "period": "2026-Q2", "snapshot_version": "v1",
     "claim_token": "t1", "contract_id": "C1", "gpi": "27100010100320",
     "units_442e7": "30", "rebate_rate": "1.250"},
    {"accrual_state": "FINALIZED", "period": "2026-Q2", "snapshot_version": "v1",
     "claim_token": "t2", "contract_id": "C1", "gpi": "27100010100320",
     "units_442e7": "90", "rebate_rate": "1.250"},
]


def test_totals_are_order_independent():
    a = aggregate(select_eligible(CLAIMS, "2026-Q2", "v1"))
    b = aggregate(select_eligible(list(reversed(CLAIMS)), "2026-Q2", "v1"))
    assert a == b
    assert a[0]["units"] == Decimal("120.000")
    assert a[0]["rebate_amount"] == Decimal("150.00")  # 120 * 1.25
    assert a[0]["claim_count"] == 2


def test_duplicate_claim_counted_once():
    eligible = select_eligible(CLAIMS + [CLAIMS[0]], "2026-Q2", "v1")
    assert len(eligible) == 2  # t1 deduped


def test_export_has_no_phi_fields():
    lines = aggregate(select_eligible(CLAIMS, "2026-Q2", "v1"))
    export = to_utilization_export("RBT-C1-2026-Q2-ABC", lines)
    forbidden = {"302-C2", "310-CA", "claim_token", "cardholder_id", "patient_name"}
    for row in export:
        assert forbidden.isdisjoint(row.keys())

The last test is the load-bearing one: it fails the build if any member field ever appears in the manufacturer export. Treat it as a PHI regression gate, not a unit test — a schema change that quietly reintroduces claim_token into the export dict is exactly the class of mistake this assertion exists to catch, and it should run in CI on every commit that touches the export path. The Decimal("120.000") and Decimal("150.00") assertions are equally deliberate: they pin the exact quantization (120 units summed, times a 1.25 per-unit rate, quantized to cents) so a change to rounding mode or quantization exponent breaks a test rather than silently shifting a manufacturer bill.

A second test worth adding drives the reversal path: build an invoice, then feed a reversal for one of its claims into the next period’s selection and assert the original line is unchanged and a negative true-up appears in the following cycle. That test encodes the append-only contract — invoiced lines are immutable — which is the single most common thing a new engineer gets wrong by “just fixing” a sent bill in place.

Gotchas and PHI Guardrails #

• Post-invoice reversals. A fill reversed after the period is cut is never edited into the sent line; it nets in the next cycle via automating rebate true-up adjustments. In-place edits desync the manufacturer’s copy and break the audit trail.
• Duplicate claim inclusion. Unioning overlapping selections double-counts units. Deduplicate on claim_token at selection and let the group-by sum — never append — as the second guard.
• Float units. float(units) * rate rounds differently across platforms. Convert 442-E7 to Decimal at the boundary via Decimal(str(x)), sum in Decimal, and quantize once with ROUND_HALF_UP.
• Rate conflict in a group. Divergent rates for one (contract_id, gpi, period) group signal an effective-dating error from the contract snapshot; raise, don’t average.
• PHI leakage in exports. The utilization file is an external disclosure. It must never carry 302-C2 Cardholder ID, 310-CA Patient Name, or even the internal claim_token. Assert the export’s key set is disjoint from member fields in CI, and see designing secure data pipelines for PHI claims adjudication for the boundary pattern. Field conventions for the file itself follow the NCPDP standards.

Related #

• Rebate Invoice Generation — the parent subsystem this aggregation runs inside, with the full invoice-line schema and builder.
• Manufacturer Rebate Contract Modeling — supplies the versioned snapshot and rate each group is priced on.
• Rebate Accrual Reconciliation — finalizes accruals before selection and absorbs post-invoice reversals.
• NDC-to-GPI Crosswalk Automation — resolves the 407-D7 NDC to the GPI the group-by key depends on.
• Security & Compliance Boundaries for Claims Data — the disclosure boundary the PHI-stripping step enforces.

← Back to Rebate Invoice Generation