Automating Short-Pay Deduction Codes in Python

This page fixes the three ways hand-assigned short-pay deduction codes leak recovered dollars: the same variance gets a different code from every analyst, carriers reject codes they do not recognize, and reversals are never tracked so clawed-back money quietly disappears from the audit’s savings number.

The failure you are hitting

Your team recovers overcharges by short-paying carriers, and the deductions are coded by hand from a shared spreadsheet. It works until volume grows, and then it fails in three ways, none of which throws:

• Inconsistent codes. The same fuel-surcharge overcharge is coded FSC-ADJ by one analyst, D110 by another, and FUEL by a third. The carrier’s AR team sees three unrelated strings for one recurring issue and cannot batch-reconcile any of them, so each deduction is worked by hand or parked in suspense.
• Carrier-rejected codes. A code that is perfectly clear internally — ACC-UNSUP for an unsupported accessorial — means nothing on carrier NEMF’s remittance advice, which expects A200. The deduction is rejected, the invoice is paid in full on the carrier’s side, and the recovery silently reverses without anyone booking it.
• Untracked reversals. When a carrier does rebut a deduction and AP re-issues the withheld amount, nothing writes that reversal back. The audit’s dashboard still counts the original short-pay as recovered, so reported savings drift above reality by exactly the sum of the un-booked reversals.

Together these turn a recovery engine into a slow leak: dollars are withheld, disputed, and re-paid, and the ledger never learns.

Root cause analysis

None of this is a Python bug. It traces to three missing structures that a spreadsheet cannot enforce:

1. No canonical variance-to-code map. Assignment lives in human memory, so it is non-deterministic by construction. Two analysts, or the same analyst on two days, map one variance to two codes.
2. No carrier-specific translation. Internal codes are treated as if they are universal. Each carrier’s AR system has its own deduction vocabulary, and a code that is never translated is a code that is regularly rejected.
3. No reversal lifecycle. A deduction is modeled as a one-way event — booked and forgotten — instead of a record with states. There is nowhere to record that a carrier rebutted it and the money went back.

Reproducible diagnostic

Before changing anything, confirm which failure dominates. This snippet reads a month of booked deductions and quantifies code inconsistency and untracked reversals:

from collections import defaultdict

# rows: dicts loaded from the deduction export / ledger dump
def audit_codes(rows: list[dict]) -> dict:
    codes_per_variance = defaultdict(set)
    reissued_not_reversed = 0
    for r in rows:
        codes_per_variance[r["variance_type"]].add(r["reason_code"])
        if r.get("reissued") and r["state"] != "REVERSED":
            reissued_not_reversed += 1

    inconsistent = {v: sorted(c) for v, c in codes_per_variance.items() if len(c) > 1}
    return {
        "inconsistent_variances": inconsistent,
        "reissued_not_reversed": reissued_not_reversed,
        "total": len(rows),
    }

report = audit_codes(rows)
print("codes per variance (should be 1 each):", report["inconsistent_variances"])
print("re-issued but not booked reversed:", report["reissued_not_reversed"])

Read the output against this table:

If more than one code appears for a single variance_type, stop hand-coding immediately — that inconsistency is what carriers reject and what makes recovered dollars untraceable.

Resolution path

The fix is three small, testable layers: a deterministic canonical mapper, a per-carrier translation table with an explicit refusal path, and a reversal state machine. Pin the dependencies so CI and production agree:

# requirements.txt
pydantic==2.10.6
structlog==24.4.0
pytest==8.3.4

Step 1 — Build a deterministic canonical code mapper

Replace human memory with one frozen table. Given a variance type, it returns exactly one canonical reason code, every time, and raises on anything it does not know rather than guessing:

# codes/canonical.py
from types import MappingProxyType

# Read-only so no caller can mutate the map at runtime.
CANONICAL: MappingProxyType = MappingProxyType({
    "OVERCHARGE_LINEHAUL": "D100",
    "FSC_DRIFT": "D110",
    "WEIGHT_RERATE": "D120",
    "MIN_CHARGE_MISAPPLIED": "D130",
    "ACCESSORIAL_UNSUPPORTED": "D200",
    "DUPLICATE_BILL": "D300",
})


class UnknownVariance(Exception):
    """No canonical code exists for this variance type."""


def to_canonical(variance_type: str) -> str:
    """One variance type -> exactly one canonical reason code, deterministically."""
    try:
        return CANONICAL[variance_type]
    except KeyError as exc:
        # An invented code is exactly the code carriers reject. Fail loudly.
        raise UnknownVariance(f"no canonical code for {variance_type!r}") from exc

The variance vocabulary this map keys on is produced upstream in Charge Variance Classification; keep the two in lockstep so a new variance type never reaches assignment without a code.

Step 2 — Translate the canonical code per carrier

The canonical code is internal. Each carrier gets its own translation so it only ever sees a code its AR system recognizes. A carrier that has no mapping for a code is not sent a guess — it is flagged as a translation gap so the recovery routes to a formal claim instead of a rejection:

# codes/translate.py
from codes.canonical import to_canonical

# canonical code -> carrier's own deduction vocabulary, per SCAC
CARRIER_CODES: dict[str, dict[str, str]] = {
    "NEMF": {"D100": "R100", "D110": "F110", "D200": "A200", "D300": "DUP01"},
    "CTII": {"D100": "RATE", "D110": "FUEL", "D120": "WGT", "D200": "ACC"},
    "WARD": {"D100": "D100", "D110": "D110", "D200": "D200", "D300": "D300"},
}


class CarrierCodeGap(Exception):
    """Carrier has no accepted code for this canonical code; route to a claim."""


def carrier_code(scac: str, variance_type: str) -> str:
    """Resolve the code THIS carrier will accept, or raise a routable gap."""
    canonical = to_canonical(variance_type)
    table = CARRIER_CODES.get(scac.upper())
    if table is None:
        raise CarrierCodeGap(f"no translation table for carrier {scac!r}")
    try:
        return table[canonical]
    except KeyError as exc:
        raise CarrierCodeGap(
            f"{scac} does not accept {canonical} for {variance_type}"
        ) from exc

Edge case: carrier CTII has no D300 mapping because it disputes every duplicate manually rather than accepting a deduction. Translation surfaces that as a CarrierCodeGap, which is the correct signal to raise the recovery through Reconciling Short-Pays Against Carrier Remittance rather than short-paying a code the carrier will bounce.

Step 3 — Track reversals with a state machine

A deduction is a record with a lifecycle, not a fire-and-forget event. The state machine allows only legal transitions, and a reversal is booked atomically so the ledger can never show a re-issued payment against a still-open short-pay:

# codes/lifecycle.py
import enum
import structlog

logger = structlog.get_logger()


class State(str, enum.Enum):
    OPEN = "OPEN"
    ACCEPTED = "ACCEPTED"
    REVERSED = "REVERSED"
    ESCALATED = "ESCALATED"


# Only these transitions are legal; anything else is a bug, not a state.
ALLOWED = {
    State.OPEN: {State.ACCEPTED, State.REVERSED, State.ESCALATED},
    State.ACCEPTED: set(),          # final
    State.REVERSED: set(),          # final
    State.ESCALATED: {State.REVERSED, State.ACCEPTED},
}


class IllegalTransition(Exception):
    """Attempted a transition the lifecycle does not permit."""


def apply_response(record: dict, response: str, ledger) -> dict:
    """Advance a deduction on a carrier response and book any reversal atomically."""
    current = State(record["state"])
    reversible = record["reason_code"] != "D300"  # duplicates are not reversible

    if response == "accepted":
        target = State.ACCEPTED
    elif response == "rejected" and reversible:
        target = State.REVERSED
    else:                            # rejected on a non-reversible code
        target = State.ESCALATED

    if target not in ALLOWED[current]:
        raise IllegalTransition(f"{current} -> {target} not allowed")

    if target is State.REVERSED:
        # Re-issue and reversal entry are one unit of work — never split them.
        ledger.book_reversal(record["deduction_id"], record["deducted_amount"])
        logger.info("deduction_reversed", deduction_id=record["deduction_id"])

    record["state"] = target.value
    return record

Edge case: a rejected response on a D300 duplicate cannot reverse — a duplicate that was truly paid twice is not up for negotiation — so it routes to ESCALATED for a human, and only then can it reach REVERSED if the carrier proves the second invoice was legitimate.

Verification

Prove each failure is closed, not hidden. These assertions belong in the suite that runs on every code-table change:

import pytest
from codes.canonical import to_canonical, UnknownVariance
from codes.translate import carrier_code, CarrierCodeGap
from codes.lifecycle import apply_response, State, IllegalTransition


class FakeLedger:
    def __init__(self):
        self.reversals = []
    def book_reversal(self, ded_id, amount):
        self.reversals.append((ded_id, amount))


def test_canonical_is_deterministic():
    assert to_canonical("FSC_DRIFT") == to_canonical("FSC_DRIFT") == "D110"
    with pytest.raises(UnknownVariance):
        to_canonical("NOT_A_TYPE")


def test_translation_is_carrier_specific():
    assert carrier_code("NEMF", "ACCESSORIAL_UNSUPPORTED") == "A200"
    assert carrier_code("CTII", "OVERCHARGE_LINEHAUL") == "RATE"
    with pytest.raises(CarrierCodeGap):
        carrier_code("CTII", "DUPLICATE_BILL")   # CTII has no D300 mapping


def test_reversal_is_booked_once():
    ledger = FakeLedger()
    rec = {"state": "OPEN", "reason_code": "D100",
           "deduction_id": "abc123", "deducted_amount": "200.00"}
    out = apply_response(rec, "rejected", ledger)
    assert out["state"] == State.REVERSED.value
    assert ledger.reversals == [("abc123", "200.00")]


def test_duplicate_rejection_escalates_not_reverses():
    ledger = FakeLedger()
    rec = {"state": "OPEN", "reason_code": "D300",
           "deduction_id": "dup777", "deducted_amount": "540.00"}
    out = apply_response(rec, "rejected", ledger)
    assert out["state"] == State.ESCALATED.value
    assert ledger.reversals == []                 # nothing re-issued yet


def test_accepted_is_final():
    rec = {"state": "ACCEPTED", "reason_code": "D100",
           "deduction_id": "x", "deducted_amount": "10.00"}
    with pytest.raises(IllegalTransition):
        apply_response(rec, "rejected", FakeLedger())

In production the proof is telemetry: exactly one reason_code per variance_type, a rejection rate near zero per carrier, and one deduction_reversed log line for every re-issued payment. A rejection-rate spike on a single SCAC means that carrier changed its accepted codes — update the translation table, do not disable the check.

Preventive configuration

Encode the guarantees as configuration so the regression cannot return:

deduction_codes:
  enforce_canonical_map: true        # reject any variance_type without a code
  reject_untranslated: true          # never short-pay a code the carrier lacks
  non_reversible_codes: [D300]       # duplicates escalate, never auto-reverse
  book_reversal_atomically: true     # re-issue and ledger entry are one unit
  alert_rejection_rate_per_carrier: 0.02
  fail_ci_on_multi_code_variance: true

• Single-code CI gate. Assert every variance_type in the canonical map resolves to exactly one code; a duplicate mapping fails the build.
• Translation coverage test. For each active carrier, assert every canonical code that carrier receives has a translation, or is explicitly listed as a claim-only gap.
• Reversal reconciliation. Nightly, assert there are zero re-issued payments without a matching REVERSED entry — the guard against silent savings drift.
• Shared vocabulary. The canonical map is imported, never copied, so the same codes back the assignment in Short-Pay & Deduction Management and the reconciliation on carrier remittance.

FAQ

Why do carriers keep rejecting my deduction codes even though they look correct internally?

Because “correct internally” and “recognized by the carrier’s AR system” are different things. Each carrier reconciles deductions against its own vocabulary, so an internal D200 has to be translated to whatever that carrier accepts — A200 for one, ACC for another. Add a per-carrier translation table (Step 2) and stop sending untranslated internal codes.

How do I stop three analysts from coding the same variance three different ways?

Take the decision away from analysts. A single frozen map from variance_type to one canonical code (Step 1) makes assignment deterministic, and a CI gate that fails when any variance type resolves to more than one code keeps it that way. Assignment becomes a lookup, not a judgment call.

A carrier rebutted a deduction and we re-paid — why is our savings number still too high?

Because the reversal was never booked. If a deduction is modeled as a one-way event, re-issuing the withheld amount does not update anything, so the dashboard still counts the original short-pay as recovered. Model the deduction as a state machine and book the reversal atomically with the re-issue (Step 3); then re-issued dollars leave the savings total automatically.

Should a rejected duplicate-billing deduction reverse automatically?

No. A duplicate that was genuinely paid twice is not negotiable, so a rejection on a D300 code escalates to a human instead of auto-reversing. Only if the carrier proves the second invoice was legitimate does it move to reversed. Auto-reversing duplicates would hand back money the carrier was never owed.

Related

• Short-Pay & Deduction Management — the parent stage this code layer implements.
• Reconciling Short-Pays Against Carrier Remittance — where translation gaps and reversals are matched against what the carrier actually paid.
• Charge Variance Classification — the source of the variance types the canonical map is keyed on.
• Reconciliation Ledger Schema — the append-only ledger every reversal is booked into.
• Carrier Dispute API Integration — where claim-only recoveries and escalated duplicates are filed.

Up one level: Short-Pay & Deduction Management · Section: Discrepancy Resolution & Dispute Routing