Accessorial Charge Scoring: Implementation Guide

Accessorial Charge Scoring operates as a deterministic validation layer within the freight audit pipeline. It quantifies the likelihood that an accessorial charge aligns with contracted terms, operational reality, and historical benchmarks. This stage consumes normalized line-item payloads from upstream mapping processes, applies weighted rule matrices, and outputs a confidence metric that dictates downstream routing. The scoring engine functions as a core component of Rule-Based Rate Validation & Accessorial Auditing, translating carrier invoice data into auditable, dispute-ready records without relying on probabilistic models or external API calls.

Pipeline Architecture & Stage Boundaries

Strict stage isolation prevents logic bleed and ensures idempotent execution. The scoring stage begins only after raw EDI 210 segments, carrier portal JSON, or OCR-extracted PDFs have been normalized into a canonical schema. It ends immediately before dispute routing or payment reconciliation.

Inbound Contract: Normalized accessorial records with validated types, standardized units, and enriched lane metadata. Outbound Contract: Scored payload containing confidence_score (0.0–1.0), routing_flag (APPROVE, REVIEW, QUARANTINE), and a structured score_breakdown detailing rule application.

This stage does not perform raw document parsing, carrier master data synchronization, or dispute ticket generation. Those responsibilities belong to upstream ingestion and downstream resolution modules.

Data Contract & Schema Enforcement

The scoring engine expects a rigidly typed accessorial payload. Schema validation must occur at the pipeline boundary using declarative frameworks such as Great Expectations or Pydantic. Records violating type constraints, missing mandatory fields, or containing negative monetary values are quarantined before scoring execution.

Canonical Accessorial Schema:

{
  "shipment_id": "string (UUID)",
  "carrier_scac": "string (4-char)",
  "accessorial_code": "string (carrier or NMFC standard)",
  "accessorial_description": "string",
  "billed_amount": "decimal(10,2)",
  "uom": "string (enum: 'per_stop', 'per_cwt', 'flat', 'per_hour')",
  "trigger_event": "string (e.g., 'detention_hours', 'residential_delivery')",
  "trigger_value": "decimal(8,2)",
  "lane_origin_zip": "string (5-char)",
  "lane_dest_zip": "string (5-char)",
  "actual_weight_lbs": "decimal(8,1)",
  "billing_zone": "string",
  "invoice_date": "date",
  "raw_source_payload": "json"
}

Financial precision is non-negotiable. All monetary and weight calculations must use fixed-point arithmetic to prevent floating-point drift. The Python decimal module is the standard for this requirement (Python Decimal Documentation).

Configuration Schema & Threshold Matrices

Scoring logic is driven by version-controlled YAML profiles. Each accessorial type maps to a scoring configuration containing base validity weights, contractual caps, deviation tolerances, and mandatory operational triggers. This separation allows operations teams to adjust tolerance thresholds without triggering CI/CD deployments.

Accessorial Scoring Profile Schema:

accessorial_profiles:
  DETENTION:
    base_score: 1.0
    contractual_cap_per_hour: 75.00
    deviation_tolerance_pct: 15.0
    required_triggers: ["dock_in_time", "dock_out_time", "free_time_minutes"]
    fallback_score: 0.6
  LIFTGATE:
    base_score: 0.95
    flat_rate_max: 85.00
    residential_override: true
    required_triggers: ["delivery_type"]
    fallback_score: 0.75

Configuration loading must implement schema validation, atomic file replacement, and circuit-breaker fallbacks to prevent pipeline halts during malformed config deployments.

Deterministic Scoring Engine Implementation

The scoring engine evaluates each record against its corresponding profile. It calculates a composite confidence score by applying base weights, checking contractual boundaries, and verifying trigger completeness. Missing triggers reduce the score deterministically rather than throwing exceptions.

import logging
from decimal import Decimal, ROUND_HALF_UP
from typing import Dict, Any, Optional
from pydantic import BaseModel, Field, ValidationError

logger = logging.getLogger(__name__)

class AccessorialRecord(BaseModel):
    shipment_id: str
    carrier_scac: str
    accessorial_code: str
    billed_amount: Decimal
    uom: str
    trigger_event: str
    trigger_value: Optional[Decimal] = None
    lane_origin_zip: str
    lane_dest_zip: str
    billing_zone: str

class ScoredPayload(BaseModel):
    original: AccessorialRecord
    confidence_score: Decimal = Field(ge=0.0, le=1.0)
    routing_flag: str  # APPROVE, REVIEW, QUARANTINE
    score_breakdown: Dict[str, str]
    applied_profile: str

def load_scoring_profiles(config_path: str) -> Dict[str, Any]:
    """Load and validate YAML configuration with atomic fallback."""
    # Implementation omitted for brevity; use ruamel.yaml + schema validation
    pass

def calculate_accessorial_score(
    record: AccessorialRecord, 
    profile: Dict[str, Any]
) -> ScoredPayload:
    """Deterministic scoring with strict boundary enforcement."""
    breakdown = {}
    score = Decimal(profile.get("base_score", 0.5))
    
    # 1. Contractual Cap Validation
    cap = Decimal(str(profile.get("contractual_cap_per_hour", 0) or profile.get("flat_rate_max", 0)))
    if cap > 0 and record.billed_amount > cap:
        overage = record.billed_amount - cap
        tolerance = Decimal(str(profile.get("deviation_tolerance_pct", 0))) / 100
        allowed_overage = cap * tolerance
        
        if overage > allowed_overage:
            score -= Decimal("0.4")
            breakdown["cap_breach"] = f"Exceeds cap by {overage} (tolerance: {tolerance})"
        else:
            breakdown["cap_breach"] = "Within tolerance"
    else:
        breakdown["cap_breach"] = "No cap or within limit"

    # 2. Trigger Completeness Check
    required = profile.get("required_triggers", [])
    # In production, trigger presence is resolved upstream via event enrichment
    missing_triggers = [t for t in required if not _is_trigger_met(record, t)]
    
    if missing_triggers:
        penalty = Decimal("0.15") * len(missing_triggers)
        score -= penalty
        breakdown["triggers"] = f"Missing: {', '.join(missing_triggers)}"
    else:
        breakdown["triggers"] = "All required triggers present"

    # 3. Clamp & Route
    final_score = max(Decimal("0.0"), min(Decimal("1.0"), score.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)))
    
    if final_score >= Decimal("0.85"):
        routing = "APPROVE"
    elif final_score >= Decimal("0.60"):
        routing = "REVIEW"
    else:
        routing = "QUARANTINE"

    return ScoredPayload(
        original=record,
        confidence_score=final_score,
        routing_flag=routing,
        score_breakdown=breakdown,
        applied_profile=record.accessorial_code
    )

def _is_trigger_met(record: AccessorialRecord, trigger: str) -> bool:
    """Deterministic trigger resolution. Extend per operational reality."""
    trigger_map = {
        "dock_in_time": record.trigger_value is not None and record.trigger_value > 0,
        "delivery_type": record.trigger_event in ("residential_delivery", "limited_access"),
        "free_time_minutes": record.trigger_value is not None
    }
    return trigger_map.get(trigger, False)

Error Handling & Deterministic Fallbacks

Production pipelines must gracefully degrade under partial data conditions. The scoring engine implements a three-tier error strategy:

  1. Schema Violations: Caught at the pipeline boundary. Records are routed to a dead-letter queue (DLQ) with explicit field-level error messages. No scoring is attempted.
  2. Missing Configuration: If an accessorial_code lacks a matching profile, the engine applies a global fallback profile (UNKNOWN_ACCESSORIAL) with a conservative base score of 0.5 and forces REVIEW routing.
  3. Trigger Data Gaps: When upstream enrichment fails to populate required triggers (e.g., missing dock_in_time), the engine applies a deterministic penalty rather than halting. This prevents false positives during carrier system outages.

All exceptions are wrapped in structured logging payloads containing shipment_id, carrier_scac, and error_context. Silent defaults are strictly prohibited.

Operational Deployment & Observability

The scoring stage must integrate with existing pipeline orchestration and monitoring frameworks. Key operational requirements include:

  • Idempotency: Scoring functions must produce identical outputs for identical inputs, regardless of execution order or retry attempts.
  • Config Versioning: Threshold matrices are deployed via GitOps. The engine reads from a read-only snapshot directory to prevent mid-batch configuration drift.
  • Metric Emission: Emit counters for records_scored, routing_distribution (APPROVE/REVIEW/QUARANTINE), and config_fallback_invocations. Track scoring latency percentiles (p50, p95) to detect upstream enrichment bottlenecks.
  • Cross-Validation Dependencies: Scoring assumes prerequisite data integrity. Lane-based rate adjustments are resolved upstream via Lane Matching Algorithms, while dimensional and zonal discrepancies are normalized through Weight & Zone Cross-Validation before entering this stage.

By enforcing strict boundaries, deterministic fallbacks, and auditable configuration matrices, the scoring engine delivers consistent, production-grade accessorial validation at scale.