Date Window Validation Logic for Expense Automation Pipelines

Date window validation is the deterministic temporal gate that decides, before any money moves, whether an expense timestamp legitimately falls inside an authorized travel window. Within the broader Automated Policy Validation & Anomaly Flagging framework, this stage owns one narrow but load-bearing responsibility: resolving receipt_timestamp, travel_start_date, and travel_end_date into a single, auditable validation state. It consumes normalized records from Receipt Ingestion & OCR Data Extraction, treats the travel-window bounds supplied by Core Policy Architecture & Taxonomy Design as a fixed input contract, and delegates everything downstream — repeat-submission checks to Duplicate Receipt Detection, category thresholds to Merchant Category Code Routing, and adaptive limits to Dynamic Threshold Tuning. Getting the temporal decision right first is what keeps those later stages from scoring chronologically impossible claims. The policy-specific mapping of these dates onto corporate travel rules is covered in Validating expense dates against corporate travel policies.

Problem Framing & Root Causes

Temporal defects rarely announce themselves; they surface as false duplicate matches, misapplied caps, and broken audit trails several stages downstream. Four named failure modes account for the overwhelming majority of production incidents:

• Temporal decoupling — a receipt arrives without its authorizing travel window (or the window arrives late from the ERP), so a legitimate trip gets flagged out-of-window purely because bounds were absent at evaluation time.
• Timezone drift — a receipt stored in UTC crosses a calendar boundary when re-localized to the traveler’s zone (a 11 PM local dinner becomes the next fiscal day in UTC), flipping window membership and fracturing the audit period.
• Grace-period ambiguity — a submission lands hours after the trip ends; whether it is VALID, buffered, or a violation is undefined unless the buffer is an explicit, versioned policy value rather than reviewer discretion.
• Future-dated skew — clock skew or an OCR year typo (2062 for 2026) produces a timestamp far in the future that silently passes naive >= checks.

Design Constraints & Prerequisites

This stage must occupy a fixed checkpoint immediately after ingestion and parsing, and strictly before any financial routing. Every record must leave with exactly one of five idempotent states, logged immutably so exceptions route to the correct review queue without manual timestamp reconciliation:

The upstream data contract is a normalized payload carrying expense_id, receipt_timestamp, travel_start_date, and travel_end_date. Because pipelines routinely ingest millions of line items per close cycle, loading whole datasets into memory for temporal checks causes OOM crashes in containerized workers. The stage therefore streams records in bounded chunks and yields validated results one at a time, keeping memory flat regardless of batch size — the same chunked-queue discipline described in async batch processing. Compliance preconditions: IANA timezones only (never OS-level offsets), and a pinned policy version attached to every decision so Sarbanes-Oxley Act reviewers can reconstruct the exact rules active at processing time.

Production Python Implementation

The module below is self-contained and runnable on Python 3.9+. It uses zoneinfo for IANA resolution, dateutil for ambiguous-format parsing, and structlog for JSON audit events. Every decision emits a deterministic decision_hash so re-processing the same record produces the same audit fingerprint, and the generator interface keeps memory flat across arbitrarily large batches.

from __future__ import annotations

import hashlib
import json
from dataclasses import asdict, dataclass
from datetime import datetime, timedelta, timezone
from enum import Enum
from typing import Any, Dict, Iterable, Iterator, Optional, Tuple

import structlog
from dateutil import parser as dateutil_parser
from zoneinfo import ZoneInfo

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

POLICY_VERSION = "date-window/2026.07"


class ValidationState(str, Enum):
    VALID = "VALID"
    GRACE_PERIOD_APPLIED = "GRACE_PERIOD_APPLIED"
    OUTSIDE_WINDOW = "OUTSIDE_WINDOW"
    MISSING_DATE = "MISSING_DATE"
    TIMEZONE_DRIFT = "TIMEZONE_DRIFT"


@dataclass(frozen=True)
class WindowPolicy:
    """Versioned, immutable configuration snapshot pinned per batch."""

    default_timezone: str = "America/New_York"
    submission_grace_hours: int = 72
    strict_travel_window: bool = True
    max_future_skew_hours: int = 24
    policy_version: str = POLICY_VERSION

    @property
    def tz(self) -> ZoneInfo:
        return ZoneInfo(self.default_timezone)


@dataclass(frozen=True)
class ValidationResult:
    expense_id: str
    state: ValidationState
    detail: str
    receipt_utc: Optional[str]
    travel_start_utc: Optional[str]
    travel_end_utc: Optional[str]
    grace_end_utc: Optional[str]
    policy_version: str
    decision_hash: str
    audit_timestamp: str


class DateWindowValidator:
    """Deterministic temporal gate. One record in, one ValidationResult out."""

    def __init__(self, policy: WindowPolicy) -> None:
        self.policy = policy
        self._tz = policy.tz

    def _parse(self, raw: Optional[Any]) -> Optional[Tuple[datetime, bool]]:
        """Parse a raw timestamp to UTC. Returns (utc_dt, had_explicit_tz)."""
        if raw is None:
            return None
        text = str(raw).strip()
        if not text or text.lower() in {"null", "none", "nan"}:
            return None
        try:
            dt = dateutil_parser.parse(text)
        except (ValueError, OverflowError, TypeError):
            return None
        had_tz = dt.tzinfo is not None
        if not had_tz:  # naive inputs are localized to the policy travel zone
            dt = dt.replace(tzinfo=self._tz)
        return dt.astimezone(timezone.utc), had_tz

    def _window_offset_seconds(self, ref_utc: datetime) -> float:
        """UTC offset of the travel zone at a given instant (DST-aware)."""
        offset = ref_utc.astimezone(self._tz).utcoffset()
        return abs(offset.total_seconds()) if offset else 0.0

    def _result(
        self,
        expense_id: str,
        state: ValidationState,
        detail: str,
        receipt: Optional[datetime],
        start: Optional[datetime],
        end: Optional[datetime],
        grace_end: Optional[datetime],
    ) -> ValidationResult:
        def iso(dt: Optional[datetime]) -> Optional[str]:
            return dt.isoformat() if dt else None

        fingerprint = "|".join([
            expense_id, state.value, str(iso(receipt)),
            str(iso(start)), str(iso(end)), self.policy.policy_version,
        ])
        decision_hash = hashlib.sha256(fingerprint.encode("utf-8")).hexdigest()
        result = ValidationResult(
            expense_id=expense_id,
            state=state,
            detail=detail,
            receipt_utc=iso(receipt),
            travel_start_utc=iso(start),
            travel_end_utc=iso(end),
            grace_end_utc=iso(grace_end),
            policy_version=self.policy.policy_version,
            decision_hash=decision_hash,
            audit_timestamp=datetime.now(timezone.utc).isoformat(),
        )
        logger.info("date_window_validated", **asdict(result))
        return result

    def validate_row(self, row: Dict[str, Any]) -> ValidationResult:
        expense_id = str(row.get("expense_id", "UNKNOWN"))
        receipt = self._parse(row.get("receipt_timestamp"))
        start = self._parse(row.get("travel_start_date"))
        end = self._parse(row.get("travel_end_date"))

        if receipt is None:
            return self._result(expense_id, ValidationState.MISSING_DATE,
                                "receipt timestamp null or unparseable",
                                None, None, None, None)
        if start is None or end is None:
            return self._result(expense_id, ValidationState.MISSING_DATE,
                                "travel window bounds missing",
                                receipt[0], None, None, None)

        receipt_utc, had_tz = receipt
        start_utc, end_utc = start[0], end[0]
        grace_end = end_utc + timedelta(hours=self.policy.submission_grace_hours)

        # Reject impossible future receipts (clock skew or OCR year typo).
        now = datetime.now(timezone.utc)
        if receipt_utc > now + timedelta(hours=self.policy.max_future_skew_hours):
            return self._result(expense_id, ValidationState.OUTSIDE_WINDOW,
                                "receipt dated beyond future-skew tolerance",
                                receipt_utc, start_utc, end_utc, grace_end)

        if start_utc <= receipt_utc <= end_utc:
            state, detail = ValidationState.VALID, "within authorized travel window"
        elif end_utc < receipt_utc <= grace_end:
            state = ValidationState.GRACE_PERIOD_APPLIED
            detail = f"within {self.policy.submission_grace_hours}h post-trip grace buffer"
        else:
            # Distinguish genuine out-of-window from a boundary that only fails
            # under UTC because the travel-zone offset was not applied.
            nearest_gap = min(abs((receipt_utc - start_utc).total_seconds()),
                              abs((receipt_utc - grace_end).total_seconds()))
            drift_budget = self._window_offset_seconds(receipt_utc)
            if had_tz and nearest_gap <= drift_budget:
                state = ValidationState.TIMEZONE_DRIFT
                detail = "fails window only under UTC; travel-zone offset may reconcile"
            else:
                state = ValidationState.OUTSIDE_WINDOW
                detail = "precedes travel start or exceeds grace buffer"

        return self._result(expense_id, state, detail,
                            receipt_utc, start_utc, end_utc, grace_end)

    def validate_stream(
        self, rows: Iterable[Dict[str, Any]]
    ) -> Iterator[ValidationResult]:
        """Lazily validate an arbitrarily large iterable of records.

        Memory stays flat: one row is materialized at a time, so this pattern
        wraps a pandas `chunksize` reader or a polars lazy frame without ever
        loading the full batch into RAM.
        """
        for row in rows:
            yield self.validate_row(row)


if __name__ == "__main__":
    structlog.configure(processors=[structlog.processors.JSONRenderer()])
    policy = WindowPolicy()
    validator = DateWindowValidator(policy)

    fixtures = [
        {"expense_id": "EXP-1", "receipt_timestamp": "2026-05-15 19:30",
         "travel_start_date": "2026-05-14", "travel_end_date": "2026-05-18"},
        {"expense_id": "EXP-2", "receipt_timestamp": "2026-05-19T20:00:00-04:00",
         "travel_start_date": "2026-05-14", "travel_end_date": "2026-05-18"},
        {"expense_id": "EXP-3", "receipt_timestamp": None,
         "travel_start_date": "2026-05-14", "travel_end_date": "2026-05-18"},
        {"expense_id": "EXP-4", "receipt_timestamp": "2026-05-14T03:00:00+00:00",
         "travel_start_date": "2026-05-14", "travel_end_date": "2026-05-18"},
    ]
    for result in validator.validate_stream(fixtures):
        print(json.dumps({"expense_id": result.expense_id,
                          "state": result.state.value,
                          "detail": result.detail}))

EXP-1 resolves VALID, EXP-2 lands 30 hours past trip end and is tagged GRACE_PERIOD_APPLIED, EXP-3 is MISSING_DATE, and EXP-4 — a UTC-stored 3 AM receipt that is actually 11 PM the prior evening in America/New_York — is caught as TIMEZONE_DRIFT rather than silently rejected.

Configuration Reference

Every field is a WindowPolicy attribute; the whole object is frozen and version-pinned so a single batch is judged by exactly one configuration. Pin the policy_version in your config store and bump it whenever any temporal rule changes (they typically change quarterly).

Validation & Testing

Temporal logic fails at boundaries, so the test suite pins the exact edges rather than sampling the interior. Use frozen fixtures and assert on state, and assert that decision_hash is stable across repeated runs to prove idempotency.

import pytest

from date_window import DateWindowValidator, ValidationState, WindowPolicy


@pytest.fixture
def validator() -> DateWindowValidator:
    return DateWindowValidator(WindowPolicy(submission_grace_hours=72))


def _row(receipt: object) -> dict:
    return {"expense_id": "T", "receipt_timestamp": receipt,
            "travel_start_date": "2026-05-14", "travel_end_date": "2026-05-18"}


def test_inclusive_start_boundary(validator: DateWindowValidator) -> None:
    assert validator.validate_row(_row("2026-05-14 00:00")).state is ValidationState.VALID


def test_exact_grace_edge_is_still_grace(validator: DateWindowValidator) -> None:
    # end 2026-05-18T00:00 local -> +72h == 2026-05-21T00:00 local, inclusive.
    assert validator.validate_row(_row("2026-05-21 00:00")).state is ValidationState.GRACE_PERIOD_APPLIED


def test_one_second_past_grace_is_violation(validator: DateWindowValidator) -> None:
    assert validator.validate_row(_row("2026-05-21 00:00:01")).state is ValidationState.OUTSIDE_WINDOW


def test_utc_midnight_crossing_flags_drift(validator: DateWindowValidator) -> None:
    # 03:00Z on the start date is 23:00 the previous evening in America/New_York.
    assert validator.validate_row(_row("2026-05-14T03:00:00+00:00")).state is ValidationState.TIMEZONE_DRIFT


def test_missing_receipt(validator: DateWindowValidator) -> None:
    assert validator.validate_row(_row(None)).state is ValidationState.MISSING_DATE


def test_decision_hash_is_idempotent(validator: DateWindowValidator) -> None:
    row = _row("2026-05-15 12:00")
    assert validator.validate_row(row).decision_hash == validator.validate_row(row).decision_hash

The confidence gate on this component is the drift and grace boundaries: if either the timezone-crossing test or the exact-grace-edge test regresses, the build must fail closed, because those are precisely the cases that leak false approvals into Duplicate Receipt Detection across adjacent fiscal periods.

Operational Runbook

1. Deploy behind a version gate. Ship the new policy_version alongside the code; process an 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 a compliance data warehouse. Confirm decision_hash and policy_version appear on every event before enabling downstream routing.
3. Monitor state distribution. Emit a counter per ValidationState. Baseline the ratios during a clean week, then alert on deviations.
4. Alert thresholds. Page when GRACE_PERIOD_APPLIED exceeds ~5% of a batch (submission-process breakdown or a misaligned buffer), when TIMEZONE_DRIFT exceeds ~1% (an upstream feed is emitting UTC without localization), or when MISSING_DATE exceeds its baseline (an OCR or ERP contract regression — see receipt error categorization).
5. Triage the review queues. MISSING_DATE and TIMEZONE_DRIFT records carry both interpretations and the raw payload, so reviewers reclassify rather than investigate.
6. Roll forward, not back, on data. Because validation is idempotent, reprocessing a corrected batch under the same policy_version yields identical decision_hash values, so audit continuity is preserved through any replay.

Related

• Automated Policy Validation & Anomaly Flagging
• Validating expense dates against corporate travel policies
• Duplicate Receipt Detection
• Merchant Category Code Routing
• Dynamic Threshold Tuning