Handling EDI 844 Rebate Claims with Python

When a distributor buys product at wholesale, resells it to a contracted end customer at a lower negotiated price, and then transmits an X12 844 (Product Transfer Account Adjustment) to claim back the deviation, the claim is only as trustworthy as the agreement it is bound to — an 844 line that references a contract number your parser cannot resolve to a specific, effective-dated agreement version is not a claim, it is an unadjudicated liability. This page documents the exact procedure for turning a raw 844 transaction set into typed, validated claim facts: walking the envelope and LIN detail loop, extracting claimed quantities and deviated prices as decimal.Decimal, binding every claim line to one agreement version, emitting a fact with a deterministic record_key, and reconciling against the AMT and CTT control totals. It is the implementation-level companion to the CSV & EDI parsing workflows topic area, which governs how vendor submissions enter the broader data ingestion normalization pipelines.

Prerequisites

Before parsing a single segment, confirm the following are in place:

• A trading-partner data contract declaring the X12 version (commonly 004010 or 005010), the segment terminator, and the element/sub-element separators. Read them from ISA16 and ISA11/ISA105 rather than hardcoding ~ and *, so a partner shipping | separators does not silently corrupt your split.
• A resolvable agreement store. The 844 carries a contract number, never your internal agreement version. You need effective-dated agreement records to resolve contract_id + claim date to exactly one version — the responsibility of agreement schema design and its versioning rebate agreement schemas guidance.
• A running validation surface. Bound claim lines are adjudicated by the claim validation rule engine; this parser produces the typed facts that engine consumes, so its field names must match the engine’s expected input contract.
• Python packages: pydantic>=2.6 for schema validation and aiofiles>=23 for async I/O. Every monetary and quantity field uses decimal.Decimal from the standard library — never float.
• A sample corpus of real 844 files per partner, including at least one multi-line claim, one carrying a BGN01 purpose code of 05 (replace) or 01 (cancel), and one referencing a contract that has been re-versioned mid-quarter, so the resolver is tested against the variants that actually break it.

Step-by-Step Implementation

Step 1 — Isolate the 844 transaction sets on the real terminator

X12 segments are delimited by the terminator declared in ISA16, not by newlines. Partners insert \r\n for human readability; those bytes are cosmetic and must be stripped before splitting. A single interchange can carry 844s alongside other transaction sets, so filter on the ST*844 header and yield one ST/SE set at a time. Read the file asynchronously so a slow transmission never blocks the worker pool.

import aiofiles
from pathlib import Path
from typing import AsyncIterator, List

async def stream_844_sets(file_path: Path) -> AsyncIterator[List[str]]:
    """Yield the segments of each ST*844 transaction set in the interchange."""
    async with aiofiles.open(file_path, mode="r", encoding="utf-8") as f:
        raw = await f.read()

    terminator = raw[105] if raw.startswith("ISA") and len(raw) > 105 else "~"
    raw = raw.replace("\r", "").replace("\n", "")
    segments = [s.strip() for s in raw.split(terminator) if s.strip()]

    tx: List[str] = []
    keep = False
    for seg in segments:
        if seg.startswith("ST*"):
            keep = seg.startswith("ST*844")
            tx = [seg] if keep else []
        elif seg.startswith("SE*") and keep:
            tx.append(seg)
            yield tx
            keep, tx = False, []
        elif keep:
            tx.append(seg)

Validation check: every yielded set starts with ST*844 and ends with SE*, and the SE01 segment count equals len(tx). A mismatch means a dropped or duplicated segment — route the whole transaction set to the dead-letter queue with the raw payload retained for audit.

Step 2 — Parse the claim header from BGN, N1, and REF

The BGN segment carries the transaction purpose (BGN01: 00 original, 05 replace, 01 cancel), the distributor’s claim reference number (BGN02), and the claim date (BGN03). The N1 loop names the parties — N1*SF (ship-from), N1*ST (ship-to), and N1*SU (supplier/manufacturer being claimed against) — and REF*CT carries the contract number that links the claim back to a rebate agreement and to payout structure modeling.

from pydantic import BaseModel
from typing import Optional, List
from datetime import date, datetime

class ClaimHeader(BaseModel):
    purpose_code: str            # 00 original, 05 replace, 01 cancel
    claim_reference: str
    claim_date: date
    supplier_id: str
    ship_from: Optional[str] = None
    ship_to: Optional[str] = None
    contract_id: str

def parse_header(segments: List[str]) -> ClaimHeader:
    bgn = next((s for s in segments if s.startswith("BGN*")), None)
    if not bgn:
        raise ValueError("Missing mandatory BGN segment")
    b = bgn.split("*")

    def n1(qual: str) -> Optional[str]:
        seg = next((s for s in segments if s.startswith(f"N1*{qual}*")), "").split("*")
        return seg[4] if len(seg) > 4 else (seg[2] if len(seg) > 2 else None)

    ref_ct = next((s for s in segments if s.startswith("REF*CT*")), None)
    if not ref_ct:
        raise ValueError("Missing REF*CT contract number; claim cannot be bound")

    return ClaimHeader(
        purpose_code=b[1],
        claim_reference=b[2],
        claim_date=datetime.strptime(b[3], "%Y%m%d").date(),
        supplier_id=n1("SU") or "",
        ship_from=n1("SF"),
        ship_to=n1("ST"),
        contract_id=ref_ct.split("*")[2],
    )

Validation check: assert purpose_code is a known code and claim_reference, supplier_id, and contract_id are non-empty. A 05 replace or 01 cancel must reference a prior claim_reference you have already ingested; if you have not, hold the transaction rather than booking a floating adjustment.

Step 3 — Extract claimed quantities and deviated amounts from the LIN loop

Line granularity is what makes a rebate claim auditable. Each LIN loop carries item identifiers (VN vendor part, UP UPC), a QTY with the claimed quantity and unit of measure, and two CTP prices — the wholesale acquisition cost (CTP02 == "WHL") and the contracted resale price (CTP02 == "CON"), both in CTP03. The per-unit rebate is the deviation between them, and the line claim is that deviation times the claimed quantity. Every figure stays in Decimal.

from decimal import Decimal, ROUND_HALF_UP

class ClaimLine(BaseModel):
    line_number: str
    item_id: str
    claimed_quantity: Decimal
    uom: str
    wholesale_cost: Decimal
    contract_price: Decimal
    claim_amount: Decimal

def parse_lines(segments: List[str]) -> List[ClaimLine]:
    lines: List[ClaimLine] = []
    cur: dict = {}
    for seg in segments:
        p = seg.split("*")
        if seg.startswith("LIN*"):
            if cur:
                lines.append(_finalize(cur))
            cur = {"line_number": p[1], "item_id": p[4] if len(p) > 4 else p[2]}
        elif seg.startswith("QTY*") and cur:
            cur["claimed_quantity"] = Decimal(p[2])
            cur["uom"] = p[3] if len(p) > 3 else "EA"
        elif seg.startswith("CTP*") and cur and len(p) > 3:
            if p[2] == "WHL":
                cur["wholesale_cost"] = Decimal(p[3])
            elif p[2] == "CON":
                cur["contract_price"] = Decimal(p[3])
    if cur:
        lines.append(_finalize(cur))
    return lines

def _finalize(cur: dict) -> ClaimLine:
    qty = cur["claimed_quantity"]
    unit_rebate = cur["wholesale_cost"] - cur["contract_price"]
    cur["claim_amount"] = (unit_rebate * qty).quantize(Decimal("0.01"), ROUND_HALF_UP)
    return ClaimLine(**cur)

Validation check: assert wholesale_cost >= contract_price (a negative deviation is not a rebate) and claim_amount == ((wholesale_cost - contract_price) * claimed_quantity).quantize(Decimal("0.01")). A line whose recomputed amount disagrees with any partner-supplied line total is quarantined before it distorts an accrual.

Step 4 — Bind each claim line to one agreement version

A contract_id is not an agreement version. The same contract is re-versioned when tiers, effective dates, or eligible items change, and a claim dated 2026-07-01 must resolve to the version in force on that date. Resolve contract_id + claim_date against your effective-dated agreement store; a line that resolves to zero or more than one version is not adjudicable and is escalated through fallback routing logic rather than guessed.

class BoundClaimLine(ClaimLine):
    agreement_version: str

def resolve_version(contract_id: str, claim_date: date, store) -> str:
    """store.versions(contract_id) -> list of (version, valid_from, valid_to)."""
    hits = [
        v for (v, start, end) in store.versions(contract_id)
        if start <= claim_date <= end
    ]
    if len(hits) != 1:
        raise LookupError(
            f"{contract_id} resolves to {len(hits)} versions on {claim_date}"
        )
    return hits[0]

def bind_lines(header: ClaimHeader, lines: List[ClaimLine], store) -> List[BoundClaimLine]:
    version = resolve_version(header.contract_id, header.claim_date, store)
    return [BoundClaimLine(**l.model_dump(), agreement_version=version) for l in lines]

Validation check: assert every emitted line has a non-empty agreement_version and that the resolver returned exactly one hit. Overlapping validity windows in the store are a data-quality defect in the agreement itself — surface it as an exception, never silently take the first match.

Step 5 — Emit typed claim facts with a deterministic record_key

Downstream settlement must be able to replay a file without double-booking. Give every claim line a record_key derived only from stable identity — the contract, agreement version, claim reference, line number, and item — so the same logical claim always hashes to the same key regardless of retransmission or ordering.

import hashlib

class RebateClaimFact(BaseModel):
    record_key: str
    claim_reference: str
    supplier_id: str
    contract_id: str
    agreement_version: str
    line_number: str
    item_id: str
    claimed_quantity: Decimal
    claim_amount: Decimal
    claim_date: date
    purpose_code: str

def to_facts(header: ClaimHeader, bound: List[BoundClaimLine]) -> List[RebateClaimFact]:
    facts = []
    for l in bound:
        seed = "|".join([
            header.contract_id, l.agreement_version,
            header.claim_reference, l.line_number, l.item_id,
        ])
        key = hashlib.sha256(seed.encode()).hexdigest()[:32]
        facts.append(RebateClaimFact(
            record_key=key, claim_reference=header.claim_reference,
            supplier_id=header.supplier_id, contract_id=header.contract_id,
            agreement_version=l.agreement_version, line_number=l.line_number,
            item_id=l.item_id, claimed_quantity=l.claimed_quantity,
            claim_amount=l.claim_amount, claim_date=header.claim_date,
            purpose_code=header.purpose_code,
        ))
    return facts

Validation check: the settlement upsert is keyed on record_key, so a replayed original (purpose_code == "00") is a no-op; assert that re-running the same file produces zero net change. A 05 replace supersedes the prior fact by record_key; a 01 cancel tombstones it.

Step 6 — Reconcile against the AMT and CTT control totals

The 844 closes with AMT (the supplier-visible total claimed amount) and CTT (CTT01 line count, CTT02 quantity hash total). Reconcile the assembled facts against both before releasing them to settlement, deduction and financial close; a control break means a line was dropped, duplicated, or misparsed.

def reconcile(facts: List[RebateClaimFact], segments: List[str]) -> None:
    ctt = next((s.split("*") for s in segments if s.startswith("CTT*")), None)
    amt = next((s.split("*") for s in segments if s.startswith("AMT*")), None)

    if ctt and int(ctt[1]) != len(facts):
        raise ValueError(f"CTT line count {ctt[1]} != {len(facts)} parsed facts")

    total = sum((f.claim_amount for f in facts), Decimal("0")).quantize(Decimal("0.01"))
    if amt and Decimal(amt[2]).quantize(Decimal("0.01")) != total:
        raise ValueError(f"AMT total {amt[2]} != reconciled {total}")

Validation check: assert the reconciled claim_amount sum equals AMT02 to the cent and the fact count equals CTT01. A divergence is a business-logic error routed to the reconciliation exception queue with the raw payload retained — not a parse failure and never a silent truncation.

Common Failure Modes and Fixes

1. Claim line not bound to an agreement version. Emitting a fact keyed only on contract_id lets two different versions of the same contract collapse into one accrual bucket. Always resolve contract_id + claim_date to exactly one version (Step 4); a zero- or multi-hit resolution is an exception, not a default to the newest version.
2. Float amounts and prices. Parsing CTP03 as float accumulates rounding drift across thousands of lines and pushes claims into the wrong tier. Cast prices and quantities to Decimal at the segment boundary and quantize the computed claim_amount to two places.
3. Duplicate claim re-submission. Partners routinely retransmit an original 844 (BGN01 == "00") after a timeout, or resend a whole day’s file. Without a deterministic record_key (Step 5) each retransmission books the rebate again. Derive the key from stable identity and make the settlement upsert idempotent; hand borderline near-duplicates to duplicate claim detection.
4. Unit-of-measure mismatch. A claim priced per case (CA) but quantified in eaches (EA), or vice versa, silently multiplies or divides the rebate by the pack size. Assert the QTY UOM matches the agreement’s contracted UOM before computing claim_amount, and convert through the pack factor when they differ rather than trusting the raw number.
5. Control break swallowed by the parser. If a LIN loop is dropped, the facts still look internally consistent — only the AMT/CTT reconciliation catches it. Never emit facts before Step 6 passes; a CTT01 or AMT02 mismatch quarantines the entire transaction set.

Operational Checklist

• Segment terminator and element separators read from the ISA header, never hardcoded.
• Only ST*844 sets processed; SE01 segment count reconciled before parsing.
• BGN purpose code validated; 05 replace and 01 cancel reference a known prior claim.
• REF*CT contract number extracted; missing contract number rejects the claim.
• Claimed quantity and both CTP prices carried as decimal.Decimal; claim_amount recomputed, not trusted.
• Every line bound to exactly one effective-dated agreement version; zero/multi hits escalated.
• record_key derived from stable identity; settlement upsert idempotent on replay.
• Claim UOM reconciled against the agreement’s contracted UOM before amount math.
• AMT total and CTT line count reconciled against assembled facts before release.
• Syntax, binding, and control-break errors routed to DLQ, fallback routing, and the exception queue respectively, with raw payload retained.

Frequently Asked Questions

What does an 844 claim actually represent? It is a distributor’s request for a Product Transfer Account Adjustment: the distributor bought product at wholesale, resold it to a contracted end customer below that cost, and is claiming the deviation back from the manufacturer. The manufacturer’s typical reply is an 849 (Response to Product Transfer Account Adjustment) approving, reducing, or denying each line.

Why resolve the agreement version by date instead of trusting the contract number on the 844? The 844 carries only the contract number the partner knows; it has no visibility into your internal versioning. When tiers or effective dates change mid-quarter you keep a new version of the same contract, so a claim must resolve to the version in force on its claim date. Binding on the bare contract number would mix incompatible tier math into one accrual.

How do I stop a retransmitted 844 from paying twice? Give each claim line a deterministic record_key built from the contract, agreement version, claim reference, line number, and item, then make the settlement upsert idempotent on that key. A resent original becomes a no-op, a replace supersedes the prior fact, and a cancel tombstones it — so retransmission never inflates the payable.

Where does the rebate amount come from if the 844 has no single amount field per line? Compute it from the line’s two CTP prices: the wholesale acquisition cost minus the contracted resale price, times the claimed quantity, quantized to the cent as a Decimal. Reconcile the sum of those computed line amounts against the AMT total in the summary before releasing any fact.

Related

• Parent topic: CSV & EDI Parsing Workflows
• Parsing EDI 810 Invoices with Python — extracting SAC off-invoice allowances from the invoice side of the same trade
• Parsing EDI 852 Product Activity Data — sell-through that substantiates the quantities claimed on an 844
• Claim Validation & Rule Engine Configuration — adjudicating the bound claim facts this parser emits
• Agreement Schema Design — the effective-dated versions each claim line binds to

Up one level: CSV & EDI Parsing Workflows