Merchant Category Code Routing for Expense Automation Pipelines

Merchant category code routing is the deterministic classification gate that resolves each transaction’s four-digit network code into a single, auditable routing decision before any spend is approved. Within the broader Automated Policy Validation & Anomaly Flagging framework, this component consumes the canonical records produced by Receipt Ingestion & OCR Data Extraction, treats the internal categories defined by Core Policy Architecture & Taxonomy Design as a fixed input contract, and emits an explainable routing path for every line item. It owns code-to-policy resolution and ambiguity handling; it delegates temporal gating to Date Window Validation Logic, repeat-submission checks to Duplicate Receipt Detection, and adaptive spend bands to Dynamic Threshold Tuning. This guide covers the routing engine, its ambiguity-resolution logic, the configuration surface, and the audit trail that makes every decision defensible.

Problem Framing & Root Causes

Category routing looks trivial — map a four-digit code to an approval path — but in production it is where the messiest failures concentrate, because the code is assigned by an acquirer or network long before your pipeline sees it and is frequently wrong, missing, or contested. Four named failure modes account for nearly every routing incident:

• MCC ambiguity — a single merchant legitimately maps to more than one plausible code (a hotel restaurant billed as 5812 “eating places” versus 7011 “lodging”), so a naive first-match lookup routes identical spend down different policies depending on which network stamped the transaction.
• Catch-all drift — acquirers default hard-to-classify merchants into miscellaneous buckets (5999, 7299), and if routing treats those as a valid category rather than an escalation signal, out-of-policy spend hides inside a permissive fallback and never surfaces for review.
• Network divergence — Visa, Mastercard, and Amex assign different codes to the same merchant, so a corporate-card feed and a manual receipt for the same purchase disagree on category, fracturing the audit trail and confusing downstream deduplication.
• OCR string mismatch — the merchant name extracted from a receipt does not match the payment-network descriptor, so any attempt to re-derive a code from text (rather than trust the network code) produces low-confidence guesses that must never be auto-approved.

The design goal is a routing engine that is deterministic (same input, same decision, same audit hash), that treats ambiguity and catch-all codes as explicit escalation states rather than silent passes, and that never invents a category it cannot defend to an auditor.

Design Constraints & Prerequisites

This stage occupies a fixed checkpoint after ingestion and temporal validation and strictly before spend-band scoring. The upstream data contract is a normalized record carrying transaction_id, a four-digit mcc_code (or an explicit null), merchant_name, amount_minor, currency, and an ocr_confidence score in [0, 1]. Category codes follow the ISO 18245 taxonomy, so routing tables remain interoperable across Visa, Mastercard, and Amex feeds; the Expense Category Taxonomies layer owns the mapping from those codes onto internal categories, and this stage consumes it read-only.

Every record must leave with exactly one of five idempotent routing states, logged immutably so exceptions reach the correct review queue without manual reclassification:

Because pipelines routinely ingest millions of line items per close cycle, loading a whole dataset into memory for classification causes OOM crashes in containerized workers. The stage therefore evaluates records in bounded chunks and yields decisions one batch at a time, keeping memory flat regardless of ledger size — the same chunked-queue discipline described in async batch processing. Compliance preconditions: a pinned policy version stamped on every decision so Sarbanes-Oxley Act reviewers can reconstruct the exact routing table active at processing time, and a low-confidence OCR gate so extractions that fall below threshold divert to review rather than auto-routing — the confidence scores originate from Tesseract OCR configuration upstream.

Production Python Implementation

The module below is self-contained and runnable. It uses pydantic v2 for strict input validation, polars for zero-copy batch evaluation, and structlog for JSON audit events. Rules are traversed in explicit priority order (blocklist before allowlist, catch-all detection before both), lookups are pre-compiled into hash sets for O(1) membership, and every decision emits a deterministic decision_hash so re-processing the same record produces the same audit fingerprint. The generator interface keeps memory flat across arbitrarily large batches.

from __future__ import annotations

import hashlib
from datetime import datetime, timezone
from enum import Enum
from pathlib import Path
from typing import Iterator, Optional

import polars as pl
import structlog
from pydantic import BaseModel, ConfigDict, Field, ValidationError

logger = structlog.get_logger("expense.mcc_routing")

POLICY_VERSION = "mcc-routing/2026.07"

# ISO 18245 catch-all / miscellaneous codes: never a valid auto-approve target.
CATCH_ALL_MCCS: frozenset[str] = frozenset({"5999", "7299", "7399", "5964"})


class RoutingState(str, Enum):
    AUTO_APPROVE = "AUTO_APPROVE"
    THRESHOLD_EXCEEDED = "THRESHOLD_EXCEEDED"
    BLOCKED = "BLOCKED"
    AMBIGUOUS = "AMBIGUOUS"
    MISSING_MCC = "MISSING_MCC"


class TransactionSchema(BaseModel):
    """Normalized record consumed by the router. Money is an integer count of
    minor units (cents) so ceiling comparisons never accumulate float error."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    transaction_id: str
    merchant_name: str
    mcc_code: Optional[str] = Field(default=None, pattern=r"^\d{4}$")
    amount_minor: int = Field(ge=0)
    currency: str = Field(min_length=3, max_length=3)
    ocr_confidence: float = Field(ge=0.0, le=1.0)


class RoutingRule(BaseModel):
    """One allow/block rule. `priority` orders traversal; lower runs first."""

    model_config = ConfigDict(frozen=True, extra="forbid")

    rule_id: str
    priority: int = 100
    mcc_allowlist: frozenset[str] = frozenset()
    mcc_blocklist: frozenset[str] = frozenset()
    max_amount_minor: Optional[int] = None
    routing_path: str


class RoutingDecision(BaseModel):
    transaction_id: str
    state: RoutingState
    routing_path: str
    matched_rule: Optional[str]
    mcc_code: Optional[str]
    detail: str
    policy_version: str
    decision_hash: str
    routed_at_utc: str


class MCCRoutingEngine:
    """Deterministic category router. One record in, one RoutingDecision out."""

    def __init__(
        self,
        rules: list[RoutingRule],
        *,
        ocr_confidence_floor: float = 0.85,
        fallback_path: str = "manual_review",
        policy_version: str = POLICY_VERSION,
    ) -> None:
        # Freeze traversal order once so routing is deterministic across runs.
        self._rules = sorted(rules, key=lambda r: (r.priority, r.rule_id))
        self._ocr_floor = ocr_confidence_floor
        self._fallback_path = fallback_path
        self._policy_version = policy_version

    def _decision(
        self,
        txn: TransactionSchema,
        state: RoutingState,
        routing_path: str,
        matched_rule: Optional[str],
        detail: str,
    ) -> RoutingDecision:
        fingerprint = "|".join([
            txn.transaction_id, state.value, routing_path,
            str(matched_rule), str(txn.mcc_code), self._policy_version,
        ])
        decision_hash = hashlib.sha256(fingerprint.encode("utf-8")).hexdigest()
        decision = RoutingDecision(
            transaction_id=txn.transaction_id,
            state=state,
            routing_path=routing_path,
            matched_rule=matched_rule,
            mcc_code=txn.mcc_code,
            detail=detail,
            policy_version=self._policy_version,
            decision_hash=decision_hash,
            routed_at_utc=datetime.now(timezone.utc).isoformat(),
        )
        logger.info("mcc_routed", **decision.model_dump())
        return decision

    def resolve(self, txn: TransactionSchema) -> RoutingDecision:
        # 1. Missing code cannot be routed — divert with the raw payload intact.
        if txn.mcc_code is None:
            return self._decision(txn, RoutingState.MISSING_MCC,
                                  self._fallback_path, None,
                                  "mcc_code null or malformed")

        # 2. Low-confidence OCR must never auto-approve; classification is unsafe.
        if txn.ocr_confidence < self._ocr_floor:
            return self._decision(txn, RoutingState.AMBIGUOUS,
                                  self._fallback_path, None,
                                  f"ocr_confidence {txn.ocr_confidence:.2f} below floor")

        # 3. Catch-all codes are an escalation signal, not a valid category.
        if txn.mcc_code in CATCH_ALL_MCCS:
            return self._decision(txn, RoutingState.AMBIGUOUS,
                                  self._fallback_path, None,
                                  f"catch-all mcc {txn.mcc_code} requires review")

        # 4. Priority-ordered traversal: blocklist wins over allowlist.
        for rule in self._rules:
            if txn.mcc_code in rule.mcc_blocklist:
                return self._decision(txn, RoutingState.BLOCKED,
                                      "ap_review", rule.rule_id,
                                      f"mcc {txn.mcc_code} blocked by {rule.rule_id}")
            if txn.mcc_code in rule.mcc_allowlist:
                if (rule.max_amount_minor is not None
                        and txn.amount_minor > rule.max_amount_minor):
                    return self._decision(txn, RoutingState.THRESHOLD_EXCEEDED,
                                          "ceiling_review", rule.rule_id,
                                          "amount over rule ceiling")
                return self._decision(txn, RoutingState.AUTO_APPROVE,
                                      rule.routing_path, rule.rule_id,
                                      "matched allowlisted rule")

        # 5. Explicit fallback — never guess a permissive category.
        return self._decision(txn, RoutingState.AMBIGUOUS,
                              self._fallback_path, None,
                              "no matching rule; routed to fallback")

    def route_batch(self, batch: pl.DataFrame) -> pl.DataFrame:
        """Validate then route a chunk, appending decision columns."""
        if batch.is_empty():
            return batch
        try:
            validated = [TransactionSchema(**row) for row in batch.to_dicts()]
        except ValidationError as exc:
            logger.error("schema_validation_failed",
                         batch_size=batch.height, error=str(exc))
            raise
        decisions = [self.resolve(txn) for txn in validated]
        return batch.with_columns(
            pl.Series("routing_state", [d.state.value for d in decisions]),
            pl.Series("routing_path", [d.routing_path for d in decisions]),
            pl.Series("decision_hash", [d.decision_hash for d in decisions]),
        )


def stream_ledger(input_path: Path, batch_size: int = 100_000) -> Iterator[pl.DataFrame]:
    """Generator over a CSV ledger. Memory stays flat regardless of file size."""
    reader = pl.read_csv_batched(source=input_path, batch_size=batch_size)
    while (batches := reader.next_batches(1)) is not None:
        yield batches[0]


if __name__ == "__main__":
    structlog.configure(processors=[structlog.processors.JSONRenderer()])
    engine = MCCRoutingEngine(rules=[
        RoutingRule(rule_id="blocked-gambling", priority=10,
                    mcc_blocklist=frozenset({"7995"}), routing_path="ap_review"),
        RoutingRule(rule_id="travel-lodging", priority=50,
                    mcc_allowlist=frozenset({"7011", "3501"}),
                    max_amount_minor=50_000, routing_path="travel_approver"),
        RoutingRule(rule_id="meals", priority=60,
                    mcc_allowlist=frozenset({"5812", "5814"}),
                    max_amount_minor=7_500, routing_path="manager_approver"),
    ])
    fixtures = [
        {"transaction_id": "T1", "merchant_name": "Grand Hotel", "mcc_code": "7011",
         "amount_minor": 42000, "currency": "USD", "ocr_confidence": 0.97},
        {"transaction_id": "T2", "merchant_name": "Casino", "mcc_code": "7995",
         "amount_minor": 20000, "currency": "USD", "ocr_confidence": 0.99},
        {"transaction_id": "T3", "merchant_name": "Corner Store", "mcc_code": "5999",
         "amount_minor": 1500, "currency": "USD", "ocr_confidence": 0.95},
        {"transaction_id": "T4", "merchant_name": "Faded Diner", "mcc_code": "5812",
         "amount_minor": 3000, "currency": "USD", "ocr_confidence": 0.40},
    ]
    for decision in (engine.resolve(TransactionSchema(**f)) for f in fixtures):
        print(decision.transaction_id, decision.state.value, decision.routing_path)

T1 resolves AUTO_APPROVE down the travel path, T2 is BLOCKED by the priority-10 gambling rule before any allowlist runs, T3 is caught as AMBIGUOUS because 5999 is a catch-all code, and T4 is AMBIGUOUS because its OCR confidence sits below the floor — none of which a naive first-match table would separate. Modeling money as amount_minor integers keeps the ceiling comparison exact, and freezing both the schema and the rule set makes re-processing idempotent.

Configuration Reference

Every routing decision is judged by exactly one immutable configuration. Pin the policy_version in your config store and bump it whenever the routing table or any threshold changes, so audit reconstruction resolves to a single rule set.

Version-pin polars, pydantic (v2), and structlog in your lockfile; a pydantic major bump changes validation semantics, and mixing v1/v2 schemas across the pipeline silently drops the extra="forbid" guard.

Validation & Testing

Routing fails at the boundaries between states, so the suite pins the exact transitions rather than sampling the interior, and asserts that decision_hash is stable across repeated runs to prove idempotency.

import pytest

from mcc_routing import (MCCRoutingEngine, RoutingRule, RoutingState,
                         TransactionSchema)


def _txn(**overrides: object) -> TransactionSchema:
    base = dict(transaction_id="T", merchant_name="M", mcc_code="5812",
                amount_minor=3000, currency="USD", ocr_confidence=0.95)
    base.update(overrides)
    return TransactionSchema(**base)


@pytest.fixture
def engine() -> MCCRoutingEngine:
    return MCCRoutingEngine(rules=[
        RoutingRule(rule_id="block", priority=10,
                    mcc_blocklist=frozenset({"7995"}), routing_path="ap_review"),
        RoutingRule(rule_id="meals", priority=60,
                    mcc_allowlist=frozenset({"5812"}),
                    max_amount_minor=7_500, routing_path="manager"),
    ])


def test_allowlisted_under_ceiling_auto_approves(engine: MCCRoutingEngine) -> None:
    assert engine.resolve(_txn(amount_minor=5000)).state is RoutingState.AUTO_APPROVE


def test_over_ceiling_is_threshold_exceeded(engine: MCCRoutingEngine) -> None:
    assert engine.resolve(_txn(amount_minor=9000)).state is RoutingState.THRESHOLD_EXCEEDED


def test_blocklist_beats_allowlist(engine: MCCRoutingEngine) -> None:
    assert engine.resolve(_txn(mcc_code="7995")).state is RoutingState.BLOCKED


def test_catch_all_is_ambiguous(engine: MCCRoutingEngine) -> None:
    assert engine.resolve(_txn(mcc_code="5999")).state is RoutingState.AMBIGUOUS


def test_low_confidence_never_auto_approves(engine: MCCRoutingEngine) -> None:
    assert engine.resolve(_txn(ocr_confidence=0.40)).state is RoutingState.AMBIGUOUS


def test_missing_code_diverts(engine: MCCRoutingEngine) -> None:
    assert engine.resolve(_txn(mcc_code=None)).state is RoutingState.MISSING_MCC


def test_decision_hash_is_idempotent(engine: MCCRoutingEngine) -> None:
    txn = _txn()
    assert engine.resolve(txn).decision_hash == engine.resolve(txn).decision_hash

The confidence gate on this component is the priority-ordering and catch-all tests: if test_blocklist_beats_allowlist or test_catch_all_is_ambiguous regresses, the build must fail closed, because those are exactly the cases that leak blocked or unclassifiable spend into Dynamic Threshold Tuning as if it were a clean category. Fixtures should also cover split transactions (one receipt, two codes) and network-divergent duplicates where a card feed and a manual receipt disagree on mcc_code — those belong to Duplicate Receipt Detection but must not auto-approve here.

Operational Runbook

1. Deploy behind a version gate. Ship the new policy_version alongside the routing table and process each in-flight batch under the manifest active when it started, never mid-flight. Roll back by re-pinning the previous version — no code revert required.
2. Wire the audit stream. Route the structlog JSON events to your SIEM or compliance warehouse and confirm decision_hash, matched_rule, and policy_version appear on every event before enabling downstream routing.
3. Tune the chunk size. Start batch_size at 100,000 rows and watch worker memory during month-end close; lower it if the resident set approaches the container budget.
4. Baseline the state distribution. Emit a counter per RoutingState over a clean week, then alert on deviation.
5. Alert thresholds. Page when AMBIGUOUS exceeds ~3% of a batch (a routing-table gap or a rise in catch-all codes from an acquirer), when MISSING_MCC exceeds its baseline (an upstream feed or OCR contract regression — see receipt error categorization), or when BLOCKED spikes (a policy change or a genuine compliance event).
6. Cross-check ceilings. Confirm every max_amount_minor matches the fixed limits owned by Spending Cap Hierarchies; routing ceilings are a first-pass gate, not the authoritative cap.
7. Roll forward, not back, on data. Because routing is idempotent, reprocessing a corrected batch under the same policy_version yields identical decision_hash values, so audit continuity survives any replay.

Merchant category code routing is a deterministic compliance control, not a heuristic classifier. Priority-ordered traversal, explicit catch-all and low-confidence escalation, and immutable audit logging give AP teams a component that scales predictably, satisfies audit requirements, and turns a manual classification bottleneck into a reproducible enforcement gate.

Related

• Automated Policy Validation & Anomaly Flagging — the parent framework this component plugs into
• Date Window Validation Logic — temporal gate that runs before routing
• Duplicate Receipt Detection — reconciles network-divergent duplicates that disagree on code
• Dynamic Threshold Tuning — governs the adaptive spend band after routing
• Expense Category Taxonomies — owns the code-to-category mapping this stage consumes
• Receipt Ingestion & OCR Data Extraction — the upstream stage that produces canonical records