Setting Up Dynamic Routing Rules for High-Value Exceptions

This page solves one precise scenario: the matching cascade has finished, a material break has fallen out as an unmatched exception, and you need routing rules that adapt to the exception’s magnitude, currency exposure, and counterparty risk at evaluation time — not against a hardcoded dollar limit set six months ago. A $4.2M settlement break and a $3 rounding difference must never travel the same path. This implementation compiles a versioned, declarative rule set, evaluates each high-value exception against it with Decimal-safe arithmetic, lands the survivors in a priority review queue, and falls back safely when downstream systems fail. It sits directly under Threshold-Based Routing Logic, which is the deterministic decision stage of the broader Exception Routing & Human-in-the-Loop Workflows pipeline, and every decision below emits a trace_id, source_hash, and match_decision for SOX traceability.

Prerequisites

Dynamic routing never runs against raw feed data. Confirm the upstream pipeline has produced the following state before invoking any code on this page:

The exception has already exhausted exact-key and tolerance matching in Transaction Matching Algorithms & Logic and emerged with a stable UNMATCHED decision.
Each exception payload carries a canonical amount_delta as a Decimal, a 3-letter ISO 4217 currency, a counterparty_tier (1–5), and an immutable source_hash.
A versioned routing matrix exists so that an identical payload plus an identical matrix version always emits an identical destination — the determinism that Threshold-Based Routing Logic requires for replayable audits.
An append-only audit sink is reachable for structured trace_id / source_hash / match_decision events.
pydantic>=2, pyyaml, and Python 3.10+ are available in the runtime.

Step-by-step implementation

Step 1 — Compile and validate the rule set

Hardcoded dollar limits are operational debt. Instead, express each routing rule declaratively in YAML and compile it into validated objects at startup, rejecting malformed boundaries before they ever touch a live exception. Use Decimal for every monetary field to avoid floating-point drift, and a model_validator to guarantee that band edges are coherent.

python

from __future__ import annotations

import logging
from decimal import Decimal
from typing import List, Optional

import yaml
from pydantic import BaseModel, Field, model_validator

audit = logging.getLogger("reconciliation.audit")


class RoutingRule(BaseModel):
    rule_id: str
    min_amount: Decimal = Field(..., ge=Decimal("0.00"))
    max_amount: Optional[Decimal] = None
    currency: str = Field(..., min_length=3, max_length=3)
    counterparty_tier: int = Field(..., ge=1, le=5)
    auto_resolve_conditions: List[str] = Field(default_factory=list)
    escalation_path: str

    @model_validator(mode="after")
    def validate_range(self) -> "RoutingRule":
        if self.max_amount is not None and self.max_amount < self.min_amount:
            raise ValueError(
                f"{self.rule_id}: max_amount must be >= min_amount"
            )
        return self


def compile_rules(yaml_path: str, *, trace_id: str, source_hash: str) -> List[RoutingRule]:
    with open(yaml_path, "r") as f:
        raw = yaml.safe_load(f)
    compiled = [RoutingRule(**r) for r in raw["rules"]]
    audit.info(
        "rules_compiled",
        extra={
            "trace_id": trace_id,
            "source_hash": source_hash,
            "match_decision": "RULES_COMPILED",
            "rule_count": len(compiled),
            "matrix_version": raw.get("version"),
        },
    )
    return compiled

Compilation failures must be logged with a structured payload containing rule_id, the compilation error, and a timestamp, then block the deploy — a routing engine that silently drops a malformed rule is a control deficiency, not a convenience.

Step 2 — Evaluate the exception against the matched band

With a validated rule set in hand, select the single rule whose band contains the exception. Comparisons stay in Decimal space; an open-ended max_amount of None means “no upper bound”, which is exactly what the top high-value band needs.

python

from dataclasses import dataclass


@dataclass(frozen=True)
class Exception_:
    ticket_id: str
    amount_delta: Decimal
    currency: str
    counterparty_tier: int
    source_hash: str


def select_rule(
    exc: Exception_, rules: List[RoutingRule], *, trace_id: str
) -> RoutingRule:
    for rule in rules:
        within_floor = exc.amount_delta >= rule.min_amount
        within_ceiling = rule.max_amount is None or exc.amount_delta <= rule.max_amount
        currency_ok = rule.currency in ("*", exc.currency)
        tier_ok = exc.counterparty_tier <= rule.counterparty_tier
        if within_floor and within_ceiling and currency_ok and tier_ok:
            audit.info(
                "rule_matched",
                extra={
                    "trace_id": trace_id,
                    "source_hash": exc.source_hash,
                    "match_decision": rule.escalation_path,
                    "rule_id": rule.rule_id,
                },
            )
            return rule
    raise LookupError(f"NO_RULE_MATCH for ticket {exc.ticket_id}")

Step 3 — Score and rank for the review queue

High-value exceptions compete for finite reviewer attention, so rank them by a composite of risk and magnitude rather than arrival order. A priority heap gives O(log n) insertion and extraction, and an idempotency guard prevents the same break from being enqueued twice when a feed is replayed.

python

import heapq
import time
from dataclasses import field


@dataclass(order=True)
class ExceptionTicket:
    priority: float
    ticket_id: str = field(compare=False)
    amount_delta: Decimal = field(compare=False)
    counterparty: str = field(compare=False)
    source_hash: str = field(compare=False, default="")
    created_at: float = field(compare=False, default_factory=time.time)
    state: str = field(compare=False, default="PENDING")


class ReviewQueue:
    def __init__(self) -> None:
        self._heap: list[ExceptionTicket] = []
        self._seen: set[str] = set()

    def push(self, ticket: ExceptionTicket, *, trace_id: str) -> None:
        if ticket.ticket_id in self._seen:
            return  # idempotent guard
        self._seen.add(ticket.ticket_id)
        heapq.heappush(self._heap, ticket)
        audit.info(
            "ticket_enqueued",
            extra={
                "trace_id": trace_id,
                "source_hash": ticket.source_hash,
                "match_decision": "ENQUEUED",
                "priority": ticket.priority,
            },
        )

    def pop(self) -> ExceptionTicket:
        ticket = heapq.heappop(self._heap)
        ticket.state = "IN_REVIEW"
        return ticket

    def __len__(self) -> int:
        return len(self._heap)

Because heapq is a min-heap, push the negated product -(risk_score * float(amount_delta)) as priority so the largest, riskiest break surfaces first. When several exceptions reference the same ledger account or counterparty, batch them into one review context to cut operational latency — that batched sign-off is the subject of Automating Batch Reconciliation Sign-Offs.

Step 4 — Configure the fallback chain

No routing engine survives production without fault tolerance. When primary evaluation fails, a downstream ledger API returns a non-recoverable error, or confidence collapses, the exception must degrade to a safe, fully-logged path rather than vanish.

python

class CircuitBreaker:
    def __init__(self, *, threshold: int = 5, window_s: float = 60.0) -> None:
        self.threshold = threshold
        self.window_s = window_s
        self._failures: list[float] = []
        self.open = False

    def record_failure(self, *, trace_id: str, source_hash: str) -> None:
        now = time.monotonic()
        self._failures = [t for t in self._failures if now - t < self.window_s]
        self._failures.append(now)
        if len(self._failures) >= self.threshold:
            self.open = True
            audit.warning(
                "circuit_open",
                extra={
                    "trace_id": trace_id,
                    "source_hash": source_hash,
                    "match_decision": "DLQ_ROUTED",
                    "failures": len(self._failures),
                },
            )


def route_with_fallback(exc: Exception_, rules, breaker, queue, *, trace_id: str) -> str:
    if breaker.open:
        return "DEAD_LETTER_QUEUE"
    try:
        rule = select_rule(exc, rules, trace_id=trace_id)
    except LookupError:
        breaker.record_failure(trace_id=trace_id, source_hash=exc.source_hash)
        return "DEAD_LETTER_QUEUE"
    queue.push(
        ExceptionTicket(
            priority=-(exc.counterparty_tier * float(exc.amount_delta)),
            ticket_id=exc.ticket_id,
            amount_delta=exc.amount_delta,
            counterparty=str(exc.counterparty_tier),
            source_hash=exc.source_hash,
        ),
        trace_id=trace_id,
    )
    return rule.escalation_path

The chain has three rungs: a circuit breaker that opens after 5 failures in a 60-second window and diverts to a dead-letter queue; jittered exponential-backoff retries capped at 3 attempts for transient network or database errors; and a manual-override fallback that force-routes to a senior reviewer with mandatory justification when automated resolution confidence drops below 0.85. Dispute and SLA handling downstream of this fallback are covered in Tracking Dispute Resolution SLAs in Python. Every fallback transition preserves the original payload so the audit trail stays complete.

Step 5 — Enforce segregation of duties on commit

The user who created an exception can never approve its resolution. Enforce dual control for any break above the configured ceiling or involving a sanctioned counterparty, and record a cryptographic signature hash alongside each approval.

python

def authorize(*, maker: str, checker: str, amount: Decimal,
              dual_threshold: Decimal, trace_id: str, source_hash: str) -> bool:
    if maker == checker:
        audit.warning(
            "sod_violation",
            extra={"trace_id": trace_id, "source_hash": source_hash,
                   "match_decision": "REJECTED_SOD"},
        )
        return False
    requires_dual = amount >= dual_threshold
    audit.info(
        "authorization",
        extra={"trace_id": trace_id, "source_hash": source_hash,
               "match_decision": "DUAL_REQUIRED" if requires_dual else "SINGLE_OK"},
    )
    return True

Configuration boundary table

Parameter	Default	Valid range	Notes
`high_value_floor`	`Decimal("10000.00")`	`> 0`	Minimum `amount_delta` that classifies an exception as high-value
`dual_control_threshold`	`Decimal("500000.00")`	`>= high_value_floor`	Above this, two distinct approvers are mandatory
`auto_resolve_confidence`	`0.85`	`0.50`–`0.99`	Below this, force-route to a senior reviewer
`breaker_failure_threshold`	`5`	`1`–`50`	Consecutive failures before the circuit opens
`breaker_window_s`	`60.0`	`5`–`600`	Rolling window for counting failures
`max_retries`	`3`	`0`–`10`	Attempts before dead-letter handoff
`queue_alert_pct`	`150`	`100`–`300`	Percent of SLA capacity that triggers a depth alert
`matrix_version`	`required`	semver string	Pins the rule set for replayable routing

Verification and testing

Confirm the implementation against a sample ledger fixture before promotion:

Shadow run. Replay 30 days of historical ledger deltas through compile_rules then route_with_fallback, asserting that no exception raises NO_RULE_MATCH and that the destination distribution matches the expected control mix.
Determinism check. Route the same fixed payload twice under one matrix_version and assert byte-identical escalation_path and match_decision — the property Threshold-Based Routing Logic depends on.
Idempotency check. Push the same ticket_id twice and assert len(queue) increments only once.
Breaker check. Inject 5 consecutive LookupErrors inside the window and assert the sixth call returns DEAD_LETTER_QUEUE.

python

def test_band_routing() -> None:
    rules = compile_rules("rules.yaml", trace_id="t-1", source_hash="h-1")
    breaker, queue = CircuitBreaker(), ReviewQueue()
    exc = Exception_("EX-900", Decimal("4200000.00"), "USD", 1, "h-900")
    dest = route_with_fallback(exc, rules, breaker, queue, trace_id="t-1")
    assert dest != "DEAD_LETTER_QUEUE"
    assert len(queue) == 1

Troubleshooting

NO_RULE_MATCH — the exception’s amount_delta, currency, or counterparty_tier falls outside every compiled band. Root cause is usually a gap between the top band’s max_amount and the next floor, or a missing wildcard * currency rule. Fix by adding an open-ended catch-all band (max_amount: null) that force-routes to a senior reviewer.
DECIMAL_DRIFT / off-by-a-cent routing — a float leaked into amount_delta upstream, so a 9999.999999 value skips the high-value floor. Normalise every monetary field to Decimal at ingestion and quantize to the currency’s minor unit.
DLQ_FLOOD — the circuit breaker keeps opening. Root cause is a flapping downstream ledger API, not the rules; inspect retry logs, widen breaker_window_s, and confirm jittered backoff is actually applied before blaming the matrix.
SOD_VIOLATION on legitimate approvals — maker and checker resolve to the same identity because a shared service account signs both steps. Bind approvals to individual operator identities, never a pooled credential.
STALE_MATRIX — two replicas route the same payload differently. They are pinned to different matrix_version values; enforce a single pinned version per deployment and reject any exception whose payload references an unknown version.

Part of Threshold-Based Routing Logic — within Exception Routing & Human-in-the-Loop Workflows.