Per Diem Rate Structuring: Deterministic Validation for High-Volume Expense Pipelines

Per diem rate structuring turns location- and date-scoped allowances into a deterministic pipeline stage that resolves a normalized (location, date, role) tuple to a single authoritative rate — or a structured exception — without loading whole rate tables into memory.

Within the broader Core Policy Architecture & Taxonomy Design framework, this stage owns one narrow contract: given a normalized expense line, return the allowed per diem and a routing decision. It consumes canonical category signals from Expense Category Taxonomies, defers absolute ceiling enforcement to Spending Cap Hierarchies, and delegates the maintenance of the underlying rate matrix to Building Dynamic Per Diem Tables for Global Teams. What follows is the runnable validation engine that sits between those pieces.

Problem Framing & Root Causes

Per diem allowances are inherently deterministic, yet most enterprise pipelines treat them as probabilistic string lookups, which produces three recurring production failures:

• Temporal decoupling. A rate is joined on location alone, ignoring effective_date/expiration_date boundaries. A trip on the day a quarterly rate changes matches two rows (or none), silently inflating or zeroing the allowance.
• Locality ambiguity. Raw itinerary text ("NYC", "New York, NY", "Manhattan") never matches the canonical location_code in the rate table, so the join returns null and the record either defaults to a national average or slips past validation entirely.
• Partial-day drift. Departure and return days carry a reduced allowance (commonly 75%), but this multiplier is applied inconsistently — sometimes in the claim, sometimes in the engine, sometimes not at all — creating reconciliation deltas that look like fraud.

Each failure mode compromises SOX and ISO 27001 traceability because the allowance figure can no longer be reproduced from a versioned input. The fix is to make rate resolution a gated, temporally-correct, memory-bounded join with explicit exception routing.

Design Constraints & Prerequisites

This engine is designed to run as a stateless batch stage with the following contract:

• Upstream data contract. Expenses arrive as a CSV or parquet stream with a normalized schema: expense_id, employee_id, location_code (already resolved to ISO 3166-2 or a GSA/DoD locality id), travel_date, role_tier, claimed_amount, is_partial_day, and policy_version. Locality resolution and date parsing are upstream concerns; see Date Window Validation Logic for the temporal normalization this stage assumes, and the Receipt Ingestion & OCR Data Extraction framework for how those fields are extracted in the first place.
• Rate snapshot. An immutable, versioned rate table with location_code, role_tier, effective_date, expiration_date, and daily_rate. Rows must have non-overlapping validity windows per (location_code, role_tier).
• Memory/throughput. Global travel programs generate millions of line items monthly. Loading the full rate table and expense batch eagerly triggers OOM and unpredictable GC pauses, so the engine must use lazy evaluation and stream to disk. Target: constant memory regardless of batch size, for batches above 500k rows/day.
• Compliance precondition. Every allowance decision must be reproducible from (policy_version, location_code, travel_date, role_tier) and emitted as an immutable audit record.

We use polars for its streaming execution engine — unlike pandas.merge, which materializes intermediate frames and duplicates column buffers, polars defers computation until the final sink, so peak memory tracks chunk size rather than total volume.

Production Python Implementation

The engine below performs a temporally-correct join, applies the partial-day multiplier deterministically, routes exceptions with structured audit metadata, and streams validated records to parquet. It is self-contained and runnable against two CSV inputs.

from __future__ import annotations

import json
import logging
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterator

import polars as pl

# --- Audit-ready structured logging -------------------------------------------
audit_logger = logging.getLogger("per_diem_compliance")
audit_logger.setLevel(logging.INFO)
_handler = logging.FileHandler("per_diem_audit.log", mode="a")
_handler.setFormatter(logging.Formatter("%(message)s"))
audit_logger.addHandler(_handler)


def log_audit_event(
    expense_id: str,
    policy_ver: str,
    allowed: float,
    claimed: float,
    code: str,
) -> None:
    """Emit one immutable JSON audit record per non-clean decision.

    The record is reproducible from the policy version and the claim, which is
    what SOX/ISO 27001 evidence requires: an allowance figure that can be
    recomputed from versioned inputs.
    """
    audit_logger.info(
        json.dumps(
            {
                "expense_id": expense_id,
                "policy_version": policy_ver,
                "allowed_amount": round(allowed, 2),
                "claimed_amount": round(claimed, 2),
                "delta": round(claimed - allowed, 2),
                "violation_code": code,
                # UTC for audit timestamps, never local date.
                "timestamp": datetime.now(timezone.utc).isoformat(),
            },
            separators=(",", ":"),
        )
    )


def _load_rates(rate_table_path: Path) -> pl.LazyFrame:
    """Lazily load an immutable rate snapshot with typed date boundaries."""
    return pl.scan_csv(rate_table_path).select(
        "location_code", "role_tier", "effective_date", "expiration_date", "daily_rate"
    ).with_columns(
        pl.col("effective_date").str.to_date("%Y-%m-%d"),
        pl.col("expiration_date").str.to_date("%Y-%m-%d"),
    )


def _load_expenses(expense_path: Path) -> pl.LazyFrame:
    """Lazily load the expense batch under a strict, explicit schema."""
    return pl.scan_csv(expense_path).select(
        "expense_id", "employee_id", "location_code", "travel_date",
        "role_tier", "claimed_amount", "is_partial_day", "policy_version",
    ).with_columns(
        pl.col("travel_date").str.to_date("%Y-%m-%d"),
        pl.col("is_partial_day").cast(pl.Boolean),
    )


def _resolve_allowances(
    expenses: pl.LazyFrame,
    rates: pl.LazyFrame,
    partial_day_multiplier: float,
) -> pl.LazyFrame:
    """Temporally-correct join + deterministic allowance calculation.

    The join keys on (location_code, role_tier); the temporal filter selects the
    single rate row whose validity window contains travel_date. A left join keeps
    unmatched expenses so RATE_NOT_FOUND can be routed rather than dropped.
    """
    joined = expenses.join(
        rates,
        on=["location_code", "role_tier"],
        how="left",
    ).filter(
        # Null-safe: rows with no matching rate survive the filter as nulls.
        pl.col("effective_date").is_null()
        | (
            (pl.col("travel_date") >= pl.col("effective_date"))
            & (pl.col("travel_date") <= pl.col("expiration_date"))
        )
    )
    return joined.with_columns(
        pl.when(pl.col("is_partial_day"))
        .then(pl.col("daily_rate") * partial_day_multiplier)
        .otherwise(pl.col("daily_rate"))
        .round(2)
        .alias("calculated_allowance")
    )


def validate_per_diem_batch(
    expense_path: Path,
    rate_table_path: Path,
    output_path: Path,
    partial_day_multiplier: float = 0.75,
    tolerance: float = 0.0,
) -> dict[str, int]:
    """Validate a per diem batch and stream clean records to parquet.

    Returns a summary count dict for the operational runbook. Exceptions are
    audit-logged, not raised, so a single bad row never stalls the batch.
    """
    rates = _load_rates(rate_table_path)
    expenses = _load_expenses(expense_path)
    resolved = _resolve_allowances(expenses, rates, partial_day_multiplier)

    # Materialize only the exception slices — small relative to the batch.
    missing = resolved.filter(pl.col("daily_rate").is_null()).collect()
    for row in missing.iter_rows(named=True):
        log_audit_event(
            expense_id=row["expense_id"],
            policy_ver=row["policy_version"],
            allowed=0.0,
            claimed=row["claimed_amount"],
            code="RATE_NOT_FOUND",
        )

    over_limit = resolved.filter(
        pl.col("daily_rate").is_not_null()
        & (pl.col("claimed_amount") > pl.col("calculated_allowance") + tolerance)
    ).collect()
    for row in over_limit.iter_rows(named=True):
        log_audit_event(
            expense_id=row["expense_id"],
            policy_ver=row["policy_version"],
            allowed=row["calculated_allowance"],
            claimed=row["claimed_amount"],
            code="OVER_PER_DIEM_LIMIT",
        )

    # Stream only clean records to disk (streaming execution, bounded memory).
    resolved.filter(
        pl.col("daily_rate").is_not_null()
        & (pl.col("claimed_amount") <= pl.col("calculated_allowance") + tolerance)
    ).select(
        "expense_id", "employee_id", "calculated_allowance", "policy_version"
    ).sink_parquet(output_path)

    return {
        "rate_not_found": missing.height,
        "over_limit": over_limit.height,
    }


if __name__ == "__main__":
    summary = validate_per_diem_batch(
        expense_path=Path("expenses.csv"),
        rate_table_path=Path("rates.csv"),
        output_path=Path("validated.parquet"),
    )
    print(json.dumps(summary))

The exception slices are collected eagerly because they are expected to be small relative to the batch; the clean-record path stays lazy and sinks straight to parquet, so peak memory tracks the exception count, not the full volume. Rows carrying RATE_NOT_FOUND are exactly the locality-ambiguity failures from the problem framing — route them to a manual review queue rather than defaulting to a national average.

Streaming very large batches

When even the exception slices risk exceeding memory, replace the eager .collect() calls with a chunked generator over the input so audit emission also stays bounded:

def iter_audit_exceptions(
    resolved: pl.LazyFrame,
    chunk_size: int = 50_000,
) -> Iterator[dict[str, object]]:
    """Yield exception rows in chunks so audit logging never materializes the
    full exception set at once. Use with .slice() over a sorted, stable order."""
    offset = 0
    while True:
        chunk = (
            resolved.filter(
                pl.col("daily_rate").is_null()
                | (pl.col("claimed_amount") > pl.col("calculated_allowance"))
            )
            .slice(offset, chunk_size)
            .collect()
        )
        if chunk.height == 0:
            break
        yield from chunk.iter_rows(named=True)
        offset += chunk_size

Configuration Reference

All tunable behavior is exposed as explicit parameters or environment-pinned constants — never hardcoded inside the join logic, which would break change-management controls.

Pin polars to a minor version in requirements.txt (e.g. polars==1.9.*): str.to_date and streaming-sink semantics have shifted across releases, and an unpinned upgrade can silently change join or null handling.

Validation & Testing

Because the engine is deterministic, tests assert exact allowances and exact routing rather than tolerances. Cover the three named failure modes plus the clean path.

import polars as pl
from pathlib import Path


def _write_csv(path: Path, df: pl.DataFrame) -> None:
    df.write_csv(path)


def test_partial_day_multiplier(tmp_path: Path) -> None:
    """A partial day at a $100 rate must allow exactly $75.00."""
    _write_csv(tmp_path / "rates.csv", pl.DataFrame({
        "location_code": ["US-NY"], "role_tier": ["IC"],
        "effective_date": ["2026-01-01"], "expiration_date": ["2026-12-31"],
        "daily_rate": [100.0],
    }))
    _write_csv(tmp_path / "exp.csv", pl.DataFrame({
        "expense_id": ["E1"], "employee_id": ["U1"], "location_code": ["US-NY"],
        "travel_date": ["2026-06-15"], "role_tier": ["IC"],
        "claimed_amount": [75.0], "is_partial_day": [True], "policy_version": ["v3"],
    }))
    summary = validate_per_diem_batch(
        tmp_path / "exp.csv", tmp_path / "rates.csv", tmp_path / "out.parquet",
    )
    assert summary == {"rate_not_found": 0, "over_limit": 0}
    out = pl.read_parquet(tmp_path / "out.parquet")
    assert out["calculated_allowance"][0] == 75.0


def test_rate_not_found_is_routed(tmp_path: Path) -> None:
    """An unresolved locality must route to review, not settle silently."""
    _write_csv(tmp_path / "rates.csv", pl.DataFrame({
        "location_code": ["US-NY"], "role_tier": ["IC"],
        "effective_date": ["2026-01-01"], "expiration_date": ["2026-12-31"],
        "daily_rate": [100.0],
    }))
    _write_csv(tmp_path / "exp.csv", pl.DataFrame({
        "expense_id": ["E2"], "employee_id": ["U2"], "location_code": ["XX-ZZ"],
        "travel_date": ["2026-06-15"], "role_tier": ["IC"],
        "claimed_amount": [90.0], "is_partial_day": [False], "policy_version": ["v3"],
    }))
    summary = validate_per_diem_batch(
        tmp_path / "exp.csv", tmp_path / "rates.csv", tmp_path / "out.parquet",
    )
    assert summary["rate_not_found"] == 1
    assert pl.read_parquet(tmp_path / "out.parquet").height == 0


def test_expired_rate_does_not_match(tmp_path: Path) -> None:
    """A travel_date outside every validity window is RATE_NOT_FOUND."""
    _write_csv(tmp_path / "rates.csv", pl.DataFrame({
        "location_code": ["US-NY"], "role_tier": ["IC"],
        "effective_date": ["2025-01-01"], "expiration_date": ["2025-12-31"],
        "daily_rate": [100.0],
    }))
    _write_csv(tmp_path / "exp.csv", pl.DataFrame({
        "expense_id": ["E3"], "employee_id": ["U3"], "location_code": ["US-NY"],
        "travel_date": ["2026-06-15"], "role_tier": ["IC"],
        "claimed_amount": [90.0], "is_partial_day": [False], "policy_version": ["v3"],
    }))
    summary = validate_per_diem_batch(
        tmp_path / "exp.csv", tmp_path / "rates.csv", tmp_path / "out.parquet",
    )
    assert summary["rate_not_found"] == 1

Fixtures worth adding beyond these: timezone drift on travel_date (a UTC-vs-local boundary that flips the effective-date match), split trips spanning a quarterly rate change, and overlapping rate rows that must be rejected at snapshot-build time rather than resolved here.

Operational Runbook

Deploy and monitor this stage as a stateless, idempotent job. Re-running the same batch against the same rate snapshot must produce byte-identical output.

1. Pre-flight the snapshot. Verify the rate table’s cryptographic checksum against the published manifest before ingestion; assert no overlapping (location_code, role_tier) validity windows. Reject the run on mismatch — never proceed on a mutated snapshot.
2. Cross-check the source. Reconcile the snapshot against the authoritative publication such as the GSA Per Diem Rates or the equivalent national tax-authority table before promoting a new policy_version.
3. Run the batch. Execute validate_per_diem_batch against the pinned snapshot; capture the returned summary dict.
4. Alert on exception ratio. If rate_not_found / total > 2%, halt downstream settlement — this signals a locality-resolution regression upstream, not individual bad claims. Tune the overage threshold that separates auto-approval from a compliance hold via Dynamic Threshold Tuning.
5. Ship audit logs. Stream per_diem_audit.log to your SIEM or compliance dashboard; each JSON line is queryable by policy_version and violation_code. Keep the handling of the underlying PII within the limits set by Security & Compliance Boundaries.
6. Roll back. Because output is a fresh parquet keyed by policy_version, rollback is repointing settlement to the prior snapshot’s output and re-running — no in-place mutation, no partial state.

Related

• Core Policy Architecture & Taxonomy Design — the parent framework this stage plugs into.
• Building Dynamic Per Diem Tables for Global Teams — how to build and version the rate snapshot this engine consumes.
• Expense Category Taxonomies — the normalization layer that isolates per diem from other reimbursables.
• Spending Cap Hierarchies — absolute ceiling enforcement that runs alongside per diem limits.
• Date Window Validation Logic — the temporal normalization this stage assumes upstream.