Freight Contract Architecture & Rate Mapping

When a freight audit program has no deterministic contract architecture underneath it, every other stage of the pipeline inherits the ambiguity. An EDI 210 invoice arrives referencing a lane and a service date, and the audit engine has to answer a deceptively hard question: which rate was in force when this shipment was tendered? Without versioned tariffs, point-in-time resolution, and an immutable archive of superseded agreements, that question is answered by guesswork — and guesswork at scale means silent overpayment, unrecoverable disputes, and SOX-relevant audit gaps that surface months later during a carrier reconciliation. This page is the architectural backbone for the rest of the freight audit pipeline: it defines how carrier contracts, negotiated rate schedules, fuel indices, and accessorial tariffs are modelled, versioned, stored, and resolved into the auditable rate matrices that downstream ingestion and validation depend on.

The audience here is Python ETL developers, freight auditors, and logistics analysts who own the rate store and the resolution logic that sits between raw carrier submissions and the AP ledger. The data flows in one direction — carrier submission → ingestion → rate resolution → validation → dispute → audit-ready record — and the contract architecture described below is what every other stage reads from. Get this layer right and the rest of the system becomes deterministic; get it wrong and no amount of downstream tuning will recover the lost variance.

Pipeline architecture overview

The contract architecture does not run in isolation. It is the system of record that the ingestion and validation tiers query on every invoice. Carrier submissions enter through Automated Invoice Parsing & EDI/XML Ingestion, where each payload is normalized into a canonical ShipmentInvoice and stamped with a contract_version_id that pins it to an exact tariff snapshot. From there the invoice flows into Rule-Based Rate Validation & Accessorial Auditing, which reads the resolved rate matrix to compute expected charges and classify variance. The contract store sits at the centre of that loop, queried point-in-time on every lookup.

The transport topology decouples carrier submission endpoints from the workers that resolve and validate. Invoices land via SFTP drops, AS2 transmissions, or REST webhooks, are normalized, and are published onto a message broker (Kafka or RabbitMQ) as discrete events. Resolution workers consume those events, query the contract store for the active version, and either approve the invoice or emit a dispute payload. The broker gives the pipeline back-pressure handling during carrier volume spikes and a natural quarantine path: any payload that cannot be resolved is routed to a dead-letter topic rather than silently dropped, preserving a complete audit trail for later reconciliation.

The canonical data flow is summarised below; the architecture stages map directly onto the system-of-record responsibilities this page defines.

Format and protocol coverage

Carrier rate data and the invoices that reference it arrive in incompatible shapes, and the contract architecture has to normalize all of them into a single rate model before resolution is possible. The ingestion routing logic inspects each payload’s transport envelope and content signature, then dispatches it to a format-specific parser. EDI 210 Motor Carrier Freight Details and Invoices and EDI 810 Invoices are parsed segment-by-segment through EDI 210/810 Processing; carrier-specific XML billing feeds route through XML Freight Bill Ingestion; and scanned or digital contract PDFs and rate sheets are handled by PDF Invoice Parsing with Python before their tariffs can be loaded into the rate store.

EDI 210 parsing in particular requires strict segment mapping so that the resolved invoice carries every field the contract store needs to find a match. The mapping below is the minimum contract for motor-carrier freight invoices.

Because freight class, weight breaks, and lane geometry differ by mode, the contract store keys each rate matrix by mode and resolves the matching tariff family. Less-than-truckload agreements expand into class-based rate tables and discount/minimum-charge structures via LTL Rate Sheet Digitization, while full-truckload agreements resolve to lane-flat or mileage-band rates through FTL Base Rate Extraction. The routing logic is mode-aware so that an LTL invoice is never validated against an FTL rate family.

Contract and rate configuration

A resilient rate model begins with a normalized contract schema that decouples carrier identifiers, lane definitions, and pricing tiers. The core data model must support effective dating, geographic zoning, weight and quantity breakpoints, and mode-specific pricing rules. Without deterministic versioning, overlapping contract periods and mid-cycle renegotiations introduce silent audit failures that compound across thousands of monthly invoices.

Production systems implement contract version control by stamping each record with effective_start, effective_end, a version_hash, and an amendment_type such as GENERAL_RATE_INCREASE or LANE_SPECIFIC_ADJUSTMENT. This is what makes the contract_version_id on every ingested invoice meaningful: it enables point-in-time rate resolution during historical invoice reconciliation and prevents temporal bleed, where a newer rate incorrectly overwrites the pricing of a shipment that was tendered weeks earlier.

To support regulatory compliance and retrospective dispute resolution, the architecture integrates immutable rate-sheet archiving as an append-only data-lake layer. Archived rate sheets are stored in partitioned Parquet or Delta Lake format, indexed by carrier_scac, contract_id, and effective_date, so that when an EDI 210 invoice references a rate from a superseded agreement, the engine can reconstruct the exact pricing environment at the time of shipment tender.

The configuration store also holds the moving parts that change more often than base rates: fuel surcharge (FSC) indices and discount tiers. FSC indices are versioned alongside the contract because the formula and the pegged index can change independently of the base tariff; the validation tier reads them through Fuel Surcharge Formula Implementation. Discount tiers are stored as ordered breakpoints so that a single lookup resolves the applicable percentage for a given shipment weight or annual volume commitment.

The schema below is the authoritative contract record. The model_validator computes a stable version_hash from the fields that define a contract’s pricing identity, so two records that price identically hash identically — which is what lets the archive deduplicate amendments that change metadata but not rates.

# models/contract_schema.py
from pydantic import BaseModel, Field, field_validator, model_validator
from datetime import date
from typing import Optional, Dict, Any
from enum import Enum
import hashlib
import json


class ContractStatus(str, Enum):
    ACTIVE = "ACTIVE"
    EXPIRED = "EXPIRED"
    SUSPENDED = "SUSPENDED"


class AmendmentType(str, Enum):
    GRI = "GENERAL_RATE_INCREASE"
    LANE_ADJ = "LANE_SPECIFIC_ADJUSTMENT"
    ACCESSORIAL_UPDATE = "ACCESSORIAL_UPDATE"


class RateContract(BaseModel):
    contract_id: str
    carrier_scac: str
    mode: str  # LTL, FTL, INTERMODAL
    effective_start: date
    effective_end: Optional[date] = None
    status: ContractStatus = ContractStatus.ACTIVE
    rate_matrix: Dict[str, Any] = Field(default_factory=dict)
    amendment_type: Optional[AmendmentType] = None
    version_hash: str = ""

    @field_validator("effective_end")
    @classmethod
    def validate_date_range(cls, v: Optional[date], info) -> Optional[date]:
        # Reject zero-width or inverted effective windows — these silently
        # break point-in-time resolution by matching nothing or everything.
        if v and v <= info.data.get("effective_start"):
            raise ValueError("effective_end must be strictly after effective_start")
        return v

    @model_validator(mode="before")
    @classmethod
    def compute_version_hash(cls, data: Any) -> Any:
        # Hash only the fields that define pricing identity so that a pure
        # metadata amendment does not produce a spurious new version.
        if isinstance(data, dict):
            payload = json.dumps({
                "contract_id": data.get("contract_id"),
                "carrier_scac": data.get("carrier_scac"),
                "effective_start": str(data.get("effective_start")),
                "rate_matrix_keys": sorted(data.get("rate_matrix", {}).keys()),
            }, sort_keys=True)
            data["version_hash"] = hashlib.sha256(payload.encode()).hexdigest()
        return data

Core ETL implementation

The primary loop the contract architecture exposes to the rest of the pipeline is resolve, then validate. Resolution selects the single tariff record that was in force at the shipment date; validation compares the billed charge against the contracted charge within tolerance. The two are deliberately separated so that a resolution failure (no active contract) and a validation failure (a real overcharge) route to different dispute paths.

Point-in-time resolution is the operation most likely to be implemented incorrectly. The common production bug is to select the latest contract for a carrier rather than the one active at ship_date, which silently misprices every historical or back-dated invoice. The engine below filters to records whose effective window contains the shipment date, then takes the most recent qualifying effective_start so that mid-cycle amendments win over the original agreement.

# validation/rate_engine.py
import logging
from datetime import date
from decimal import Decimal
from typing import Tuple

logger = logging.getLogger(__name__)


class RateValidationEngine:
    def __init__(self, tolerance_pct: float = 0.015):
        # Tolerance absorbs rounding drift between carrier billing systems
        # and the internal ERP; it is NOT a license to ignore real variance.
        self.tolerance = Decimal(str(tolerance_pct))

    def resolve_point_in_time(self, rate_matrix: dict, shipment_date: date) -> dict:
        """Return the single tariff record active on shipment_date."""
        active = []
        for _lane_id, rate in rate_matrix.items():
            if rate["effective_start"] <= shipment_date:
                end = rate.get("effective_end")
                if end is None or end >= shipment_date:
                    active.append(rate)
        if not active:
            # Resolution failure — distinct from a pricing dispute. Route to
            # the fallback cascade, not the overcharge queue.
            raise ValueError(f"No active contract for shipment date {shipment_date}")
        # Most recent effective_start wins so amendments override the original.
        return max(active, key=lambda r: r["effective_start"])

    def validate_linehaul(
        self, contracted_rate: Decimal, billed_rate: Decimal
    ) -> Tuple[bool, Decimal]:
        if contracted_rate == 0:
            return False, Decimal("0")
        variance = abs(billed_rate - contracted_rate) / contracted_rate
        is_compliant = variance <= self.tolerance
        if not is_compliant:
            logger.warning(
                "Rate variance %s exceeds tolerance. Contracted: %s, Billed: %s",
                f"{variance:.4%}", contracted_rate, billed_rate,
            )
        return is_compliant, variance

Once linehaul is resolved, the validation tier layers on accessorial and fuel checks. Carrier-specific accessorial codes are mapped to a canonical taxonomy through Accessorial Charge Taxonomy Mapping so that “detention”, “DET”, and “DETENT” all resolve to one schedule entry; charges that fail the schedule check are forwarded to Accessorial Charge Scoring for weighted penalty assignment. Lane geometry is confirmed against the contracted origin/destination pairs by Lane Matching Algorithms, and billed weight and zone are reconciled by Weight & Zone Cross-Validation before the variance verdict is finalised.

Idempotency and deduplication

Carrier feeds redeliver. The same invoice will arrive twice over AS2 after a transmission retry, appear again in a nightly portal export, and sometimes be re-sent by the carrier with a corrected total. If the pipeline processes each arrival blindly, a single shipment can be approved and paid more than once. Idempotency is therefore a first-class property of the contract architecture, not an afterthought.

The deduplication key is derived deterministically from the fields that uniquely identify a freight movement — never from an auto-increment id or a receipt timestamp. A stable composite of carrier, invoice number, and PRO survives retransmission and re-export unchanged.

# pipeline/idempotency.py
import hashlib
from typing import Optional


def ingestion_key(carrier_scac: str, invoice_number: str, pro_number: str) -> str:
    """Deterministic key — identical for every redelivery of one shipment."""
    raw = f"{carrier_scac.strip().upper()}|{invoice_number.strip()}|{pro_number.strip()}"
    return hashlib.sha256(raw.encode()).hexdigest()


def claim_or_skip(store, key: str, payload_hash: str) -> Optional[str]:
    """
    Atomically claim an ingestion key.
    Returns the disposition: 'new', 'duplicate', or 'amended'.
    """
    existing = store.get(key)
    if existing is None:
        store.put(key, payload_hash)        # first sighting
        return "new"
    if existing == payload_hash:
        return "duplicate"                  # byte-identical redelivery, skip
    store.put(key, payload_hash)
    return "amended"                        # same shipment, changed charges

The disposition matters: a byte-identical duplicate is skipped silently, but an amended redelivery (same key, different payload hash) is a real correction that must re-enter resolution. A common mistake is to treat any repeated key as a duplicate, which causes corrected invoices to be dropped and the original — possibly wrong — charge to stand. The retry safety guarantee is that processing the same payload any number of times yields exactly one settlement record.

Observability and error handling

A freight audit pipeline that cannot explain why an invoice was approved, disputed, or quarantined is unauditable, which defeats its purpose. Every resolution decision emits a structured log line carrying the invoice id, the resolved contract_version_id, the computed variance, and the disposition. Logs are structured (JSON) so that they are queryable, because “grep the worker output” does not scale to a SOX audit request six months later.

Three routing destinations cover every outcome. Invoices that resolve and validate flow to settlement. Invoices that fail resolution — no active contract, malformed mandatory segment — go to a quarantine table with the raw payload preserved for reprocessing once the missing contract is digitized. Invoices that exhaust retries on transient infrastructure errors go to a dead-letter queue with the failure context attached.

# pipeline/observability.py
import logging

logger = logging.getLogger("freight_audit")


def emit_decision(invoice_id: str, version_id: str, variance, disposition: str) -> None:
    logger.info(
        "audit_decision",
        extra={
            "invoice_id": invoice_id,
            "contract_version_id": version_id,
            "variance_pct": f"{variance:.4%}" if variance is not None else None,
            "disposition": disposition,  # approved | disputed | quarantined
        },
    )


def quarantine(store, invoice_id: str, raw_payload: bytes, reason: str) -> None:
    # Preserve the raw bytes so the invoice can be replayed unchanged once
    # the missing contract or fix lands. Never discard the original payload.
    store.write_quarantine(invoice_id=invoice_id, payload=raw_payload, reason=reason)
    logger.warning("quarantined", extra={"invoice_id": invoice_id, "reason": reason})

Alert thresholds are derived from the variance and quarantine rates, not from raw error counts. A quarantine rate above a configured ceiling for a single carrier usually means a contract lapsed or an amendment was never digitized — the alert points procurement at the specific carrier_scac. Threshold derivation and alert routing are owned by Threshold Tuning & Alerting, which the contract architecture feeds with its decision stream.

Performance and scaling

Freight audit volume is bursty: month-end and carrier batch drops can deliver hundreds of thousands of invoices in a window. The resolution and validation work is I/O-bound (contract-store lookups) with a CPU-bound core (Decimal arithmetic over rate matrices), so the scaling model is a bounded-concurrency worker pool that caps in-flight lookups while keeping the CPU saturated. Large carrier drops are chunked into micro-batches through Async Batch Processing Workflows rather than loaded into memory whole.

# pipeline/concurrency.py
import asyncio
from typing import Awaitable, Callable, List


async def run_pool(
    invoices: List[dict],
    handler: Callable[[dict], Awaitable[dict]],
    max_concurrency: int = 32,
) -> List[dict]:
    """Bounded fan-out: cap concurrent contract-store lookups."""
    sem = asyncio.Semaphore(max_concurrency)

    async def _guarded(inv: dict) -> dict:
        async with sem:               # back-pressure against the rate store
            return await handler(inv)

    # return_exceptions keeps one poison invoice from killing the batch.
    return await asyncio.gather(*(_guarded(i) for i in invoices), return_exceptions=True)

The tuning guidance below is a starting point; profile against the real contract store before committing. Resolution caching helps disproportionately because most invoices in a batch hit the same handful of active contracts.

As a rough benchmark, a single worker resolving and validating cached-tariff LTL invoices sustains the low thousands per second; uncached resolution against a cold Parquet archive is one to two orders of magnitude slower, which is why the active-contract cache is load-bearing rather than optional.

Failure modes and troubleshooting

Most production incidents in this layer trace back to a small set of recurring causes. The table maps the symptom an operator sees to the root cause and the resolution path.

When an invoice quarantines, the first diagnostic is always to re-resolve it in isolation against the archive at its exact ship_date — that immediately separates a genuine missing-contract case from a segment-drift parsing problem upstream in ingestion.

Rate-mapping modules

The contract architecture is implemented across four focused modules. Each digitizes or validates one slice of the tariff and feeds the resolved rate matrix this page describes.

• LTL Rate Sheet Digitization — Converts class-based PDF and Excel rate sheets into version-controlled, lookup-ready tables keyed by ZIP3, freight class, and weight break.
• FTL Base Rate Extraction — Resolves full-truckload lane-flat and mileage-band rates with equipment-type modifiers from negotiated FTL agreements.
• Accessorial Charge Taxonomy Mapping — Normalizes carrier-specific accessorial codes to a canonical taxonomy and validates them against the contracted schedule.
• Fuel Surcharge Formula Implementation — Implements DOE-indexed and contract-pegged FSC formulas, versioned independently of the base tariff.

Related

• Automated Invoice Parsing & EDI/XML Ingestion — the ingestion tier that stamps each invoice with the contract_version_id this architecture resolves.
• Rule-Based Rate Validation & Accessorial Auditing — the validation tier that reads the resolved rate matrix to classify variance and route disputes.
• EDI 210/810 Processing — segment-level parsing that produces the fields point-in-time resolution depends on.
• Lane Matching Algorithms — confirms billed origin/destination pairs against contracted lanes during validation.
• Threshold Tuning & Alerting — turns the decision stream emitted here into actionable alerts.

---

Up: Freight Bill Auditing & Carrier Rate Contract Automation — the freight audit pipeline home.