Transit Fare Revenue Reconciliation Automation

Implementing Pydantic Models for AFC Event Streams

The task is precise: you have a firehose of raw automated fare collection (AFC) tap events — dictionaries decoded from validators, open-loop payment gateways, and mobile-wallet SDKs — and before any of them are allowed to move money, each one must pass a single, deterministic data contract. Anything that fails is quarantined with enough forensic context to replay it later; nothing malformed reaches the ledger. This is a schema-enforcement job, not a pricing job. It is the first gate inside Schema Validation Pipelines, the validation stage of the Fare Data Ingestion & GTFS-RT Sync pipeline, and it exists so that revenue analysts, transit operations engineers, and the Python developers who own the ingestion jobs all trust the same canonical AFCTapEvent record. This guide gives you a runnable Pydantic v2 model and a routing loop that turns noisy vendor JSON into typed, auditable events.

The payloads reaching this code have already been decoded from card ciphers upstream by the smart card schema mapping layer, so the model here concerns itself only with structural correctness — types, bounds, enums, and timestamp normalization — never with cryptography or fare calculation.

The validation decision flow

Legacy firmware rarely emits uniform JSON. It serializes timestamps as naive strings, transmits cents as floats, and appends undocumented telemetry fields. The contract therefore applies selective strictness: noisy enum and timestamp inputs are coerced into canonical forms through mode="before" validators, while the accounting field fare_amount_cents is marked strict=True so a silent float-to-int coercion — a historic source of revenue leakage — is rejected outright. Pairing this with extra="forbid" guarantees that no undocumented vendor key slips into the ledger unnoticed.

Each raw payload follows exactly one path: it is normalized, validated against the strict contract, and routed to a single channel.

Single-path validation flow for one AFC tap event A raw event dict passes through before-validators that normalize the timestamp to UTC and coerce cents through Decimal, then into model_validate, which enforces extra=forbid and strict fields. The gate forks each record along exactly one path: on success it becomes a valid AFCTapEvent bound for settlement; on ValidationError it is turned into an audit record carrying the correlation id, raw payload, and error list, then routed to the rejected quarantine channel with full forensic context. ok ValidationError Raw event dict noisy vendor JSON before-validators normalize UTC · coerce cents model_validate extra=forbid strict fields Valid AFCTapEvent → settlement ledger Build audit record correlation_id · raw · errors Rejected / quarantine forensic context

Defining the strict tap-event schema

The model below normalizes heterogeneous inputs before they enter the ledger. Note that fare_amount_cents is an integer count of minor currency units — never a float — and the coerce_fare validator parses stringy or float-typed inputs through Decimal so no binary floating-point rounding can ever touch a monetary value.

import json
import logging
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from enum import Enum
from typing import Optional
from uuid import uuid4

from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator

# Structured audit logging configuration
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[logging.StreamHandler()],
)
logger = logging.getLogger("afc_reconciliation")


class TapDirection(str, Enum):
    ENTRY = "entry"
    EXIT = "exit"
    TRANSFER = "transfer"


class FareMediaType(str, Enum):
    SMART_CARD = "smart_card"
    MOBILE_WALLET = "mobile_wallet"
    CONTACTLESS_BANK = "contactless_bank"
    CASH = "cash"


class AFCTapEvent(BaseModel):
    model_config = ConfigDict(extra="forbid")

    event_id: str = Field(..., min_length=1, max_length=64)
    device_id: str = Field(..., pattern=r"^[A-Z0-9_-]+$")
    route_id: str = Field(..., pattern=r"^[A-Z0-9_-]+$")
    stop_id: str = Field(..., min_length=1)
    direction: TapDirection
    media_type: FareMediaType
    tap_timestamp: datetime
    fare_amount_cents: int = Field(ge=0, strict=True)
    agency_id: str = Field(..., min_length=2, max_length=8)
    gtfs_trip_id: Optional[str] = None
    validator_skew_ms: Optional[int] = None

    @field_validator("tap_timestamp", mode="before")
    @classmethod
    def normalize_utc(cls, v: "str | datetime") -> datetime:
        """Sanitize legacy timestamps and enforce UTC compliance."""
        dt = datetime.fromisoformat(v.replace("Z", "+00:00")) if isinstance(v, str) else v
        if dt.tzinfo is None:
            # Naive timestamps from field validators are assumed UTC, not local.
            dt = dt.replace(tzinfo=timezone.utc)
        return dt.astimezone(timezone.utc)

    @field_validator("fare_amount_cents", mode="before")
    @classmethod
    def coerce_fare(cls, v: "int | float | str") -> int:
        """Parse firmware-quirk cents (floats/strings) via Decimal, never binary float."""
        if isinstance(v, bool):  # bool is a subclass of int; reject it explicitly
            raise ValueError("boolean is not a valid fare amount")
        if isinstance(v, int):
            return v
        raw = v.replace(",", "").strip() if isinstance(v, str) else v
        try:
            amount = Decimal(str(raw))
        except InvalidOperation as exc:
            raise ValueError(f"non-numeric fare amount: {v!r}") from exc
        if amount != amount.to_integral_value():
            raise ValueError(f"fare_amount_cents must be whole cents, got {amount}")
        return int(amount)

The strict=True flag matters: after coerce_fare returns an int, Pydantic still refuses to silently accept, say, a stray Decimal or a bare float that bypassed the before-validator, so the field is guarded on both sides. The normalize_utc validator also fixes the common firmware bug of emitting naive timestamps — it stamps them UTC rather than letting Pydantic interpret them in the server’s local zone, which would shift every off-peak reconciliation by hours.

Routing valid and rejected events

Reconciliation needs deterministic routing: valid events proceed to settlement, invalid payloads are quarantined with full forensic context. The loop below never raises on bad data — it captures the structured ValidationError, builds an immutable audit record keyed by a correlation ID, and hands both channels back to the caller for downstream ledger writes and dead-letter replay.

def ingest_afc_stream(
    raw_events: list[dict],
) -> tuple[list[AFCTapEvent], list[dict]]:
    """Validate AFC tap events, routing each to the valid or rejected channel.

    Returns a tuple of (validated_models, audit_records).
    """
    valid_events: list[AFCTapEvent] = []
    rejected_audits: list[dict] = []

    for idx, payload in enumerate(raw_events):
        try:
            # Raises ValidationError on type mismatch, missing field, or extra key.
            event = AFCTapEvent.model_validate(payload)
            valid_events.append(event)
        except ValidationError as exc:
            error_details = [
                {
                    "field": ".".join(str(p) for p in err["loc"]),
                    "message": err["msg"],
                    "input": err.get("input"),
                }
                for err in exc.errors()
            ]
            audit_record = {
                "ingestion_correlation_id": str(uuid4()),
                "raw_index": idx,
                "raw_payload": payload,
                "validation_errors": error_details,
                "rejection_timestamp_utc": datetime.now(timezone.utc).isoformat(),
            }
            logger.warning("AFC payload quarantined: %s", json.dumps(audit_record, default=str))
            rejected_audits.append(audit_record)

    logger.info(
        "Stream ingestion complete. Valid: %d | Rejected: %d",
        len(valid_events),
        len(rejected_audits),
    )
    return valid_events, rejected_audits

Validation & test cases

The behavior worth pinning down is the boundary between “coerced and accepted” and “rejected and audited.” The batch below feeds the loop one clean event, one that needs coercion, and two that must fail — a float that would leak revenue and an undocumented vendor key.

batch = [
    # 1. Clean canonical event — passes untouched.
    {
        "event_id": "evt-0001",
        "device_id": "VAL-7A",
        "route_id": "R-42",
        "stop_id": "8801",
        "direction": "entry",
        "media_type": "contactless_bank",
        "tap_timestamp": "2026-07-03T08:14:05Z",
        "fare_amount_cents": 275,
        "agency_id": "METRO",
    },
    # 2. Legacy firmware quirks — stringy cents and a naive local-looking timestamp.
    {
        "event_id": "evt-0002",
        "device_id": "VAL-7A",
        "route_id": "R-42",
        "stop_id": "8802",
        "direction": "exit",
        "media_type": "smart_card",
        "tap_timestamp": "2026-07-03T08:41:00",
        "fare_amount_cents": "3.00",
        "agency_id": "METRO",
    },
    # 3. Fractional cents — a float that would silently round; must be rejected.
    {
        "event_id": "evt-0003",
        "device_id": "VAL-7A",
        "route_id": "R-42",
        "stop_id": "8803",
        "direction": "entry",
        "media_type": "mobile_wallet",
        "tap_timestamp": "2026-07-03T09:02:11Z",
        "fare_amount_cents": 199.5,
        "agency_id": "METRO",
    },
    # 4. Undocumented vendor telemetry key — extra="forbid" rejects it.
    {
        "event_id": "evt-0004",
        "device_id": "VAL-7A",
        "route_id": "R-42",
        "stop_id": "8804",
        "direction": "entry",
        "media_type": "smart_card",
        "tap_timestamp": "2026-07-03T09:15:47Z",
        "fare_amount_cents": 275,
        "agency_id": "METRO",
        "battery_pct": 85,
    },
]

valid, rejected = ingest_afc_stream(batch)
print(len(valid), len(rejected))          # -> 2 2
print(valid[1].fare_amount_cents)         # -> 3   ("3.00" parsed via Decimal)
print(valid[1].tap_timestamp.tzinfo)      # -> UTC (naive input stamped UTC)
print(rejected[0]["validation_errors"][0]["field"])  # -> fare_amount_cents
print(rejected[1]["validation_errors"][0]["field"])  # -> battery_pct

Two events settle and two are quarantined. Event 3 fails because 199.5 is not whole cents — the coerce_fare validator raises before the strict integer check, so the fraction is caught rather than truncated to 199 or 200. Event 4 fails on the unexpected battery_pct key, and because the audit record carries the full raw payload plus the offending field name, an operator can trace it back to a specific firmware build and strip the key upstream. A completely empty dict {} would produce one audit record whose validation_errors lists every required field as missing, which is the signal that a producer is emitting the wrong message shape entirely.

Transit-specific debugging steps

When reconciliation discrepancies surface, isolate the failure vector with these targeted diagnostics:

  1. Clock skew and timezone drift. Field validators often run on isolated industrial networks with no NTP sync. If tap_timestamp normalization produces off-by-hours settlement, check the device’s clock, then use the validator_skew_ms field to apply a deterministic offset during batch reconciliation rather than mutating the raw timestamp in place.
  2. Legacy float leakage. Older validators transmit fares as 12.0 or "12,00". The Decimal-based coerce_fare handles both, but if you see ValueError: non-numeric fare amount, look for a currency symbol or a localized decimal separator the replace step did not strip.
  3. GTFS trip-ID mismatches. When gtfs_trip_id is None but downstream capping needs it, cross-reference device_id and tap_timestamp against your live position feed from GTFS-RT Realtime Sync; a missing trip assignment usually means schedule drift or unlinked vehicle telemetry rather than a bad tap.
  4. Blanket rejection. If the pipeline suddenly drops every event, a new firmware key is almost always the cause. Temporarily switch to extra="ignore" in staging to surface the rogue key from the audit records, strip it upstream, then restore extra="forbid" before promoting to production — never ship ignore to a ledger-facing service.

Integration note

This model is the entry contract for the Schema Validation Pipelines component: the AFCTapEvent records it emits are the same canonical shape the streaming validator hands to the reconciliation engine, and the audit records it quarantines are exactly the payloads a dead-letter replay job re-drives once the upstream fix lands. It shares an ingestion boundary with its cousins in the same pipeline — the batch feeds parsed by Optimizing pandas chunksize for 10M-row fare files and the paged pulls in Handling rate limits on legacy AFC vendor APIs both feed dictionaries into this same model_validate gate, so every source route settles into one typed schema regardless of how the bytes arrived.

Frequently Asked Questions

Why store fares as integer cents instead of a Decimal field?

Integer minor units are the canonical ledger representation across this pipeline: they are exact, they sum without drift, and they serialize to JSON and SQL without ambiguity. Decimal is used only transiently inside coerce_fare to parse stringy or float inputs safely, then converted to int. The field itself stays a strict int so that a stray float can never reach settlement. Fare math (proration, capping) is done downstream in Decimal, not here.

Should I use `model_validate` or construct the model directly?

Always model_validate for untrusted input. Direct construction (AFCTapEvent(**payload)) also validates in Pydantic v2, but model_validate is the explicit contract-enforcement call, accepts a plain dict, and keeps the intent obvious to reviewers. Reserve model_construct for trusted, already-validated data where you deliberately want to skip validation for speed — never for raw vendor payloads.

How do I keep `extra="forbid"` from rejecting every new firmware field?

Treat unknown keys as a producer contract change, not a model bug. The audit record names the offending field, so triage it: if it is genuine telemetry you want, add a typed optional field to the model; if it is noise, strip it in the upstream adapter. The one thing not to do is leave extra="ignore" in production — it lets undocumented data flow silently past the gate, which is exactly the corruption this stage exists to stop.

What happens to a naive timestamp with no timezone?

normalize_utc stamps it UTC rather than letting Python interpret it in the server’s local zone. Assuming local time is the classic reconciliation bug: a validator in a UTC+2 region would shift every tap two hours and smear the peak/off-peak boundary. If a specific vendor genuinely emits local time, convert it in that vendor’s adapter before it reaches this model, so the contract stays unambiguous.

↑ Part of Schema Validation Pipelines