Expense Category Taxonomies: A Deterministic Normalization Layer for Automated Auditing

Expense category taxonomies convert unstructured merchant strings into standardized, policy-addressable financial records so that every downstream spend control evaluates a machine-readable category instead of probabilistic OCR text.

Within the broader Core Policy Architecture & Taxonomy Design control plane, this component owns one job: taking a normalized receipt record and assigning it a stable, version-pinned category node before any limit is applied. It does not decide monetary thresholds — that is delegated to Spending Cap Hierarchies — nor does it resolve location-aware allowances, which belong to Per Diem Rate Structuring. It also stops short of the raw merchant-signal parsing handled upstream by Merchant Category Code Routing. This page covers the taxonomy schema, the classification engine, and the audit metadata that make category assignment reproducible at scale. The narrower, GL-mapping walkthrough lives in how to structure expense categories for automated auditing.

Problem Framing & Root Causes

The primary bottleneck in expense automation is not data ingestion but the OCR-to-policy mismatch: raw merchant descriptors such as UBER* TRIP HELP.UBER.COM or AMZN MKTP US*H3K82 reach routing engines with no semantic category, so policy code cannot enforce anything deterministically. Three named failure modes follow directly from this gap:

• Taxonomy pollution — low-confidence OCR tokens are classified anyway, seeding the category layer with garbage that inflates exception queues.
• Alias drift — the same vendor arrives as AMZN MKTP, Amazon Marketplace, and AMAZON.COM*2Z9, so string-equality matching silently misroutes spend.
• Non-deterministic replay — fuzzy or ML classifiers return different nodes across library versions, so an auditor cannot reconstruct why a 2023 expense was approved.

Design Constraints & Prerequisites

The taxonomy layer sits mid-pipeline and inherits hard contracts from the stages around it. It must only activate after Receipt Ingestion & OCR Data Extraction has emitted a normalized record whose OCR confidence clears the gate (typically ≥ 0.85) and whose temporal and monetary fields are already coerced. Introducing classification earlier amplifies OCR noise; delaying it past validation breaks deterministic enforcement.

Concrete prerequisites for the component:

• Upstream data contract: each record carries expense_id, merchant_raw, amount, and ocr_confidence (0.0–1.0). Missing fields fail schema validation rather than defaulting silently.
• Determinism: classification is a pure function of the normalized merchant token and a version-pinned lookup table — no network calls, no runtime training.
• Throughput / memory: batches routinely exceed millions of rows monthly, so the engine streams in bounded chunks and never materializes the full dataset in RAM.
• Compliance precondition: every assignment emits an audit event with the taxonomy version and a deterministic decision hash so decisions are replayable under SOX and IRS Pub 583 retention rules.

Hierarchical schema

Effective taxonomies need a normalized hierarchy that satisfies both granular operational tracking and high-level financial controls. A three-tier model standardizes mapping across finance, travel, and compliance teams and binds cleanly to the chart of accounts:

Explicit node-to-policy bindings replace fuzzy logic: lodging nodes align with Per Diem Rate Structuring to trigger excess-allowance flags, and software or client-entertainment nodes bind to Spending Cap Hierarchies to enforce departmental budgets before GL posting.

Production Python Implementation

The engine below streams CSV input through polars lazy scans, validates each record with pydantic, classifies against a frozen O(1) lookup, and emits a deterministic audit event per decision. It uses a generator so memory stays flat across million-row batches, and it skips low-confidence records at the gate to prevent taxonomy pollution.

from __future__ import annotations

import hashlib
import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterator, Optional

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

logger = structlog.get_logger()

# --- 1. Version-pinned taxonomy table -------------------------------------
# Bumping any node MUST bump TAXONOMY_VERSION so decisions stay auditable.
TAXONOMY_VERSION = "2026.07.0"

TAXONOMY_LOOKUP: dict[str, tuple[str, str, str]] = {
    "UBER": ("Travel", "Ground Transport", "Rideshare Economy"),
    "LYFT": ("Travel", "Ground Transport", "Rideshare Economy"),
    "HILTON": ("Travel", "Lodging", "Standard Hotel"),
    "MARRIOTT": ("Travel", "Lodging", "Standard Hotel"),
    "AMZN MKTP": ("Office Operations", "Supplies", "General Merchandise"),
    "AWS": ("Software & Subscriptions", "Cloud Infrastructure", "Compute Tier"),
}

# Aliases collapse merchant drift into a single canonical key before lookup.
MERCHANT_ALIASES: dict[str, str] = {
    "AMAZON MARKETPLACE": "AMZN MKTP",
    "AMAZON.COM": "AMZN MKTP",
    "AMAZON WEB SERVICES": "AWS",
}

UNCLASSIFIED = ("Unclassified", "Unclassified", "Unclassified")


class ExpenseRecord(BaseModel):
    expense_id: str
    merchant_raw: str
    amount: float
    ocr_confidence: float = Field(ge=0.0, le=1.0)
    taxonomy_l1: Optional[str] = None
    taxonomy_l2: Optional[str] = None
    taxonomy_l3: Optional[str] = None
    taxonomy_version: Optional[str] = None
    decision_hash: Optional[str] = None
    policy_status: str = "PENDING"


# --- 2. Deterministic classifier ------------------------------------------
def normalize_merchant(merchant_raw: str) -> str:
    """Uppercase, strip processor noise, and resolve aliases to a canonical key."""
    cleaned = merchant_raw.upper().replace("*", " ").strip()
    # Longest alias match first so "AMAZON WEB SERVICES" beats "AMAZON.COM".
    for alias in sorted(MERCHANT_ALIASES, key=len, reverse=True):
        if cleaned.startswith(alias):
            return MERCHANT_ALIASES[alias]
    # Fall back to the two-token prefix, then the single leading token.
    tokens = cleaned.split()
    if len(tokens) >= 2 and " ".join(tokens[:2]) in TAXONOMY_LOOKUP:
        return " ".join(tokens[:2])
    return tokens[0] if tokens else ""


def classify(merchant_raw: str) -> tuple[str, str, str]:
    """O(1) lookup on the normalized key; deterministic fallback on miss."""
    return TAXONOMY_LOOKUP.get(normalize_merchant(merchant_raw), UNCLASSIFIED)


def decision_hash(expense_id: str, node: tuple[str, str, str]) -> str:
    """SHA-256 over the inputs that determined the assignment, for replay."""
    payload = json.dumps(
        {"expense_id": expense_id, "node": node, "taxonomy_version": TAXONOMY_VERSION},
        separators=(",", ":"),
        sort_keys=True,
    )
    return hashlib.sha256(payload.encode()).hexdigest()


# --- 3. Streaming batch processor -----------------------------------------
def classify_stream(
    input_path: Path,
    chunk_size: int = 50_000,
    confidence_gate: float = 0.85,
) -> Iterator[ExpenseRecord]:
    """Yield classified records lazily so peak memory stays bounded by chunk_size."""
    lf = pl.scan_csv(
        input_path,
        schema_overrides={
            "expense_id": pl.Utf8,
            "merchant_raw": pl.Utf8,
            "amount": pl.Float64,
            "ocr_confidence": pl.Float64,
        },
    )
    total = lf.select(pl.len()).collect().item()

    for offset in range(0, total, chunk_size):
        chunk = lf.slice(offset, chunk_size).collect()
        for row in chunk.iter_rows(named=True):
            try:
                if row["ocr_confidence"] < confidence_gate:
                    logger.warning(
                        "low_confidence_ocr",
                        expense_id=row["expense_id"],
                        confidence=row["ocr_confidence"],
                    )
                    continue

                node = classify(row["merchant_raw"])
                l1, l2, l3 = node
                record = ExpenseRecord(
                    expense_id=row["expense_id"],
                    merchant_raw=row["merchant_raw"],
                    amount=row["amount"],
                    ocr_confidence=row["ocr_confidence"],
                    taxonomy_l1=l1,
                    taxonomy_l2=l2,
                    taxonomy_l3=l3,
                    taxonomy_version=TAXONOMY_VERSION,
                    decision_hash=decision_hash(row["expense_id"], node),
                    policy_status="CLASSIFIED" if node != UNCLASSIFIED else "REVIEW",
                )
                logger.info(
                    "taxonomy_applied",
                    expense_id=record.expense_id,
                    taxonomy_l1=l1,
                    taxonomy_l2=l2,
                    taxonomy_l3=l3,
                    taxonomy_version=TAXONOMY_VERSION,
                    decision_hash=record.decision_hash,
                    policy_status=record.policy_status,
                    ts=datetime.now(timezone.utc).isoformat(),
                )
                yield record
            except ValidationError as exc:
                logger.error(
                    "schema_validation_failed",
                    expense_id=row.get("expense_id"),
                    error=str(exc),
                )


def run(input_path: Path, output_path: Path, chunk_size: int = 50_000) -> None:
    """Sink classified records to parquet in bounded batches (never full in RAM)."""
    buffer: list[dict] = []
    written = False
    for record in classify_stream(input_path, chunk_size=chunk_size):
        buffer.append(record.model_dump())
        if len(buffer) >= chunk_size:
            _flush(buffer, output_path, append=written)
            written = True
            buffer.clear()
    if buffer:
        _flush(buffer, output_path, append=written)


def _flush(rows: list[dict], output_path: Path, append: bool) -> None:
    frame = pl.DataFrame(rows)
    if append and output_path.exists():
        frame = pl.concat([pl.read_parquet(output_path), frame])
    frame.write_parquet(output_path)
    logger.info("chunk_flushed", rows=len(rows), append=append)

For files too large to hold even a single chunk-sized frame comfortably, replace the collected slices with a fully lazy transform and sink_parquet() on the LazyFrame; the generator contract above keeps the classification logic identical either way.

Audit-ready logging

Regulatory frameworks (SOX, IRS Pub 583) require immutable, queryable trails for automated decisions, so print() or unstructured logs are insufficient. Configure structlog to render JSON with ISO timestamps and a filtering level, matching the audit stream consumed by the rest of the control plane:

import structlog

structlog.configure(
    processors=[
        structlog.contextvars.merge_contextvars,
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.format_exc_info,
        structlog.processors.JSONRenderer(),
    ],
    wrapper_class=structlog.make_filtering_bound_logger(20),  # INFO
    cache_logger_on_first_use=True,
)

Each taxonomy_applied event carries the taxonomy_version and decision_hash, so a reviewer can replay a historical batch and confirm the same node was assigned — the property that makes classification defensible in an audit.

Configuration Reference

Version-pin polars, pydantic, and structlog in a lockfile. Because the taxonomy table is data, ship it as a versioned artifact (or config-as-code) rather than editing constants in place, so a rollback is a version re-pin, not a code revert.

Validation & Testing

Test the classifier as a pure function first, then the streaming contract:

import polars as pl
import pytest
from mymodule import classify, normalize_merchant, run, UNCLASSIFIED


@pytest.mark.parametrize("raw,expected", [
    ("UBER* TRIP HELP.UBER.COM", ("Travel", "Ground Transport", "Rideshare Economy")),
    ("AMZN MKTP US*H3K82", ("Office Operations", "Supplies", "General Merchandise")),
    ("Amazon Web Services", ("Software & Subscriptions", "Cloud Infrastructure", "Compute Tier")),
    ("UNKNOWN VENDOR 4821", UNCLASSIFIED),
])
def test_classification_is_deterministic(raw, expected):
    assert classify(raw) == expected
    assert classify(raw) == classify(raw)  # idempotent


def test_alias_drift_collapses_to_one_node():
    keys = {normalize_merchant(x) for x in ("AMZN MKTP", "Amazon Marketplace", "AMAZON.COM*2Z9")}
    assert keys == {"AMZN MKTP"}


def test_confidence_gate_drops_low_ocr(tmp_path):
    src = tmp_path / "in.csv"
    src.write_text(
        "expense_id,merchant_raw,amount,ocr_confidence\n"
        "E1,UBER TRIP,42.0,0.99\n"
        "E2,UBER TRIP,42.0,0.40\n"
    )
    out = tmp_path / "out.parquet"
    run(src, out, chunk_size=10)
    df = pl.read_parquet(out)
    assert df["expense_id"].to_list() == ["E1"]  # E2 dropped at the gate

Key edge-case fixtures to keep in the suite: faded receipts that arrive just under the confidence gate, split transactions where one physical receipt yields two merchant_raw rows, and alias variants of the same vendor. Assert that the decision_hash for a given expense_id/node pair is stable across runs — a changed hash without a TAXONOMY_VERSION bump is a regression.

Operational Runbook

1. Ship the taxonomy behind a version gate. Deploy the new TAXONOMY_VERSION alongside the code; process an in-flight batch under the version active when it started, never mid-flight. Roll back by re-pinning the previous table version — no code revert required.
2. Wire the audit stream. Route the structlog JSON events to your SIEM or compliance warehouse. Confirm decision_hash and taxonomy_version appear on every taxonomy_applied event before enabling downstream routing.
3. Baseline the node distribution. Emit a counter per L1 node and per policy_status. Record the ratios during a clean week.
4. Alert thresholds. Page when policy_status="REVIEW" (the UNCLASSIFIED share) exceeds ~3% of a batch — usually new merchants or an alias-table gap — and when low_confidence_ocr exceeds its baseline, which points at an upstream regression best triaged through receipt error categorization.
5. Triage the review queue. UNCLASSIFIED records carry the raw merchant string, so reviewers add an alias or a node rather than investigating from scratch; feed confirmed mappings back into MERCHANT_ALIASES or the table and bump the version.
6. Roll forward, not back, on data. Reprocessing a corrected batch under the same TAXONOMY_VERSION yields identical decision_hash values, so audit continuity is preserved through any replay.

Once records are classified, enforcement becomes a deterministic routing problem: threshold breaches, restricted L3 nodes, and temporal anomalies each map to a review queue defined declaratively rather than in hardcoded conditionals, which is exactly the handoff consumed by Automated Policy Validation & Anomaly Flagging.

Related

• Core Policy Architecture & Taxonomy Design
• How to structure expense categories for automated auditing
• Per Diem Rate Structuring
• Spending Cap Hierarchies
• Security & Compliance Boundaries
• Merchant Category Code Routing