Expense Audit Guide Python · Pipelines · Compliance

Implementing Tiered Spending Caps in Python

Tiered spending caps produce non-deterministic verdicts whenever overlapping role, category, and per-diem limits are evaluated in implicit order against floating-point money, so the same expense line can pass one run and fail the next. This page is the concrete engine walkthrough for the parent Spending Cap Hierarchies guide: it turns that precedence model into a runnable, type-safe evaluator that resolves the binding ceiling, normalizes currency with Decimal, and routes exceptions through auditable fallback chains before payment execution.

The evaluator sits inside the broader Core Policy Architecture & Taxonomy Design control plane. It assumes the record already carries a stable category node from Expense Category Taxonomies and a location-aware allowance from Per Diem Rate Structuring, and that it cleared confidence gating in the Receipt Ingestion & OCR Data Extraction framework. It owns exactly one job: given a normalized line, compute the tightest applicable limit and emit a defensible verdict.

Why Standard Approaches Fail

Naive cap enforcement fails in production for three named reasons, each producing a silent wrong verdict rather than a visible error:

  1. Float drift on the cent. A $0.01 overage computed with IEEE 754 binary floats rounds a 120.00 limit to 120.0000000001, so a compliant lodging line is hard-rejected — or a 120.01 line slips under. Money arithmetic must never touch float.
  2. Precedence drift. When resolution order lives implicitly in evaluation sequence, a director’s Zurich lodging line is simultaneously subject to a corporate cap, a role cap, a category cap, and a per-diem override, and the code picks whichever branch happens to run first. Refactoring the loop silently re-audits a historical report to a different ceiling — the exact ambiguity the parent Spending Cap Hierarchies model exists to remove.
  3. Verdict without evidence. Boolean-only returns (True/False) give downstream payment gateways and compliance queues nothing to route on and leave auditors no reason for the decision. A batch that scans rules linearly per transaction also stalls once monthly volume crosses a million lines.

Architecture & Algorithm

The evaluator below is self-contained and runnable. It models caps as frozen, tier-tagged rules, pre-indexes them into a Dict[Tuple[TierLevel, str], CapRule] for O(1) resolution, and returns a structured EvaluationResult carrying the applied tier, variance, routing action, and an immutable audit payload. Money is Decimal throughout with ROUND_HALF_UP so rounding is deterministic. Records that miss a rule or fall below the OCR confidence gate divert to a fallback verdict instead of being auto-rejected. Memory and latency notes are inline: @dataclass(slots=True) drops the per-instance __dict__ (roughly a 35% footprint cut in batch workloads), and the pre-built index keeps per-line work constant regardless of rule-set size.

Decision flow of TieredCapEvaluator.evaluate() A top-to-bottom flowchart of one call to evaluate(). The normalized expense line's raw_amount string is parsed to a Decimal with ROUND_HALF_UP, then the binding rule is resolved from the pre-built index keyed on tier and category. A guard checks whether the rule is missing or the receipt confidence is below 0.75; if so the line diverts to the FALLBACK_REQUIRED status routed to AUDIT_REVIEW. Otherwise the amount is compared against two limits: at or under soft_limit the verdict is APPROVED and routed to PAYMENT_GATEWAY; at or under hard_limit it is a SOFT_VIOLATION routed to the MANAGER_REVIEW_QUEUE; above hard_limit it is a HARD_VIOLATION routed to COMPLIANCE_HOLD. Each terminal node is annotated with its routing_action. One deterministic pass through evaluate() Normalized expense line raw_amount : str _parse_amount() → Decimal, quantize .01, ROUND_HALF_UP resolve binding rule (pure lookup) _rule_index[(tier, category)] rule is None, or confidence < 0.75 ? actual ≤ soft_limit ? actual ≤ hard_limit ? yes yes yes no no no FALLBACK_REQUIRED routing_action → AUDIT_REVIEW APPROVED routing_action → PAYMENT_GATEWAY SOFT_VIOLATION routing_action → MANAGER_REVIEW_QUEUE HARD_VIOLATION routing_action → COMPLIANCE_HOLD 
import logging
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional, Dict, Tuple
from decimal import Decimal, ROUND_HALF_UP, InvalidOperation
from datetime import datetime, timezone

# Structured audit logging aligned with compliance retention policies
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s | %(levelname)s | %(name)s | %(message)s',
    datefmt='%Y-%m-%dT%H:%M:%SZ'
)
logger = logging.getLogger("expense_cap_engine")

class TierLevel(Enum):
    STANDARD = 1
    MANAGER = 2
    DIRECTOR = 3
    EXECUTIVE = 4

class EvaluationStatus(Enum):
    APPROVED = "APPROVED"
    SOFT_VIOLATION = "SOFT_VIOLATION"
    HARD_VIOLATION = "HARD_VIOLATION"
    FALLBACK_REQUIRED = "FALLBACK_REQUIRED"

@dataclass(slots=True)
class ExpenseRecord:
    expense_id: str
    category: str
    raw_amount: str  # String input prevents float drift during ingestion
    currency: str
    employee_tier: TierLevel
    receipt_confidence: float = 1.0
    policy_version: str = "v1.0"

@dataclass(slots=True)
class CapRule:
    tier: TierLevel
    category: str
    hard_limit: Decimal
    soft_limit: Decimal
    currency: str = "USD"
    rule_id: str = ""

@dataclass(slots=True)
class EvaluationResult:
    expense_id: str
    status: EvaluationStatus
    applied_tier: TierLevel
    cap_limit: Decimal
    actual_amount: Decimal
    variance_pct: Decimal
    routing_action: str
    audit_payload: Dict = field(default_factory=dict)

class TieredCapEvaluator:
    def __init__(self, rules: list[CapRule]):
        # O(1) lookup index: (tier, category) -> CapRule
        self._rule_index: Dict[Tuple[TierLevel, str], CapRule] = {
            (r.tier, r.category): r for r in rules
        }
        self._currency_precision = Decimal("0.01")

    def _parse_amount(self, raw: str) -> Decimal:
        try:
            amt = Decimal(raw)
            return amt.quantize(self._currency_precision, rounding=ROUND_HALF_UP)
        except InvalidOperation as e:
            logger.error("Invalid amount format: %s", raw)
            raise ValueError(f"Malformed currency string: {raw}") from e

    def _resolve_rule(self, tier: TierLevel, category: str) -> Optional[CapRule]:
        return self._rule_index.get((tier, category))

    def evaluate(self, record: ExpenseRecord) -> EvaluationResult:
        actual = self._parse_amount(record.raw_amount)
        rule = self._resolve_rule(record.employee_tier, record.category)

        # Fallback triggers for missing rules or low OCR confidence
        if rule is None:
            return self._build_fallback(record, actual, "UNMAPPED_RULE")
        if record.receipt_confidence < 0.75:
            return self._build_fallback(record, actual, "LOW_OCR_CONFIDENCE")

        # Deterministic tier evaluation
        if actual <= rule.soft_limit:
            status, routing = EvaluationStatus.APPROVED, "PAYMENT_GATEWAY"
        elif actual <= rule.hard_limit:
            status, routing = EvaluationStatus.SOFT_VIOLATION, "MANAGER_REVIEW_QUEUE"
        else:
            status, routing = EvaluationStatus.HARD_VIOLATION, "COMPLIANCE_HOLD"

        # Variance calculation against soft threshold
        variance = ((actual - rule.soft_limit) / rule.soft_limit * 100) if rule.soft_limit else Decimal("0.00")
        variance = variance.quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)

        return EvaluationResult(
            expense_id=record.expense_id,
            status=status,
            applied_tier=record.employee_tier,
            cap_limit=rule.hard_limit,
            actual_amount=actual,
            variance_pct=variance,
            routing_action=routing,
            audit_payload={
                "rule_id": rule.rule_id,
                "policy_version": record.policy_version,
                "evaluated_at": datetime.now(timezone.utc).isoformat(),
                "confidence_score": record.receipt_confidence
            }
        )

    def _build_fallback(self, record: ExpenseRecord, actual: Decimal, reason: str) -> EvaluationResult:
        logger.warning("Fallback triggered for %s: %s", record.expense_id, reason)
        return EvaluationResult(
            expense_id=record.expense_id,
            status=EvaluationStatus.FALLBACK_REQUIRED,
            applied_tier=record.employee_tier,
            cap_limit=Decimal("0.00"),
            actual_amount=actual,
            variance_pct=Decimal("0.00"),
            routing_action="AUDIT_REVIEW",
            audit_payload={
                "fallback_reason": reason,
                "policy_version": record.policy_version,
                "evaluated_at": datetime.now(timezone.utc).isoformat()
            }
        )

Two limits per rule give the engine three verdicts instead of a binary pass/fail: at or under soft_limit the line is APPROVED and released to the payment gateway; between the soft and hard limits it becomes a SOFT_VIOLATION routed to manager review; above hard_limit it is a HARD_VIOLATION held for compliance. The variance_pct field quantifies how far over the soft threshold a line landed, which lets Dynamic Threshold Tuning recalibrate soft limits from real breach distributions rather than guesswork. Because _resolve_rule keys on (tier, category), the winning rule is a pure lookup — no ordered scan whose sequence could drift under refactoring.

Step-by-Step Integration

  1. Load the cap set once and freeze it. Build the evaluator at batch start from an immutable policy snapshot so every line in the run audits against one policy_version; never read a cap from a mutable store mid-batch.

    rules = [
    CapRule(TierLevel.DIRECTOR, "LODGING", Decimal("450.00"), Decimal("400.00"), "USD", "R-DIR-LODG"),
    CapRule(TierLevel.STANDARD, "MEALS", Decimal("75.00"), Decimal("60.00"), "USD", "R-STD-MEAL"),
]
evaluator = TieredCapEvaluator(rules)

  2. Feed string amounts, never floats. Coerce upstream amount fields to str before constructing the record so parsing owns the Decimal conversion. Verify the cent boundary holds:

    ok = ExpenseRecord("E1", "MEALS", "60.00", "USD", TierLevel.STANDARD, receipt_confidence=0.98)
assert evaluator.evaluate(ok).status is EvaluationStatus.APPROVED
over = ExpenseRecord("E2", "MEALS", "75.01", "USD", TierLevel.STANDARD, receipt_confidence=0.98)
assert evaluator.evaluate(over).status is EvaluationStatus.HARD_VIOLATION

  3. Divert low-confidence and unmapped lines, don’t reject them. Confirm the fallback path routes to review rather than auto-blocking so degraded scans reach a human via the receipt error categorization path:

    blurry = ExpenseRecord("E3", "MEALS", "50.00", "USD", TierLevel.STANDARD, receipt_confidence=0.4)
assert evaluator.evaluate(blurry).routing_action == "AUDIT_REVIEW"

  4. Stream, don’t materialize. Iterate records through a generator and yield each EvaluationResult straight to the downstream queue (Kafka, SQS) so memory stays flat regardless of batch size.

  5. Hot-reload atomically. For live policy edits, rebuild _rule_index off to the side and swap it under a threading.Lock/asyncio.Lock so no line ever sees a half-updated rule set.

  6. Wire the audit stream first. Route the structured logger to your SIEM and confirm rule_id, policy_version, and evaluated_at appear on every verdict before any COMPLIANCE_HOLD action can gate a payment.

Edge Cases & Gotchas

Edge condition Failure it causes Mitigation
Float amount reaches the evaluator $0.01 overage misjudged by binary drift Accept raw_amount as str, cast to Decimal, apply ROUND_HALF_UP per the Python decimal documentation
Linear rule scan per transaction Pipeline stalls on million-line batches Pre-index rules into Dict[Tuple[TierLevel, str], CapRule] for O(1) resolution
Boolean-only return Routing and audit have no reason attached Return EvaluationResult with an immutable audit_payload (rule id, timestamp, confidence)
Receipt confidence below 0.75 Rigid enforcement hard-blocks a degraded scan Route to FALLBACK_REQUIRED / AUDIT_REVIEW instead of auto-rejection
Cross-currency line (EUR vs USD rule) Amount compared against a limit in another currency Normalize with cached, TTL-bounded FX rates before evaluation; never compute rates inline
Non-deterministic call in the path Identical inputs yield different audit hashes Keep evaluate a pure function — no random, no unseeded uuid, no wall-clock branching

Every fallback carries an immutable audit_payload with UTC timestamps, policy versions, and the fallback reason, satisfying NIST SP 800-53 AU-2 (Audit Events) and AU-3 (Content of Audit Records). Where that payload could contain sensitive fields, scope its retention through Security & Compliance Boundaries.

FAQ

Why use two limits (soft and hard) per rule instead of one?

A single limit forces a binary pass/fail and either floods manager queues or lets marginal overages through. The soft limit marks “over policy but plausibly justified” and routes to review; the hard limit marks “block and hold.” The gap between them is where a SOFT_VIOLATION verdict and its variance_pct let you triage rather than reject.

Why Decimal instead of float for money?

Binary floating point cannot represent most decimal cent values exactly, so 0.1 + 0.2 != 0.3 and a limit comparison at the cent boundary becomes non-deterministic. Decimal with ROUND_HALF_UP makes every arithmetic step reproducible, which is a hard requirement for a verdict that must replay identically years later under audit.

How does the O(1) rule index stay correct when policy changes mid-run?

It doesn’t change mid-run — that is the point. The snapshot is frozen at batch start so the whole run audits against one policy_version. For live systems, rebuild the index off to the side and swap the reference atomically under a lock; in-flight lines finish against the old snapshot and new lines pick up the new one cleanly.

What happens to an expense that has no matching cap rule?

_resolve_rule returns None, and the record becomes FALLBACK_REQUIRED with reason UNMAPPED_RULE routed to AUDIT_REVIEW — never a silent pass. This surfaces taxonomy gaps back to Expense Category Taxonomies so the missing (tier, category) pair can be added deliberately.

How do I extend this to multiple currencies?

Normalize each line’s amount to the rule’s currency before comparison, using FX rates cached with a TTL and fetched outside the evaluation path. Keep the conversion deterministic per snapshot — pin the rate set alongside the policy_version so a re-audit reproduces the same converted amount and verdict.