Calculating Cross-Operator Transfer Windows with Python

The task on this page is precise and operational: given two tap events on different operators, decide whether the second tap qualifies as a free or discounted transfer under a shared inter-agency agreement, and produce an auditable record of that decision. When a rider crosses from a municipal bus to regional rail to a third-party microtransit shuttle, each system stamps its own clock, applies its own grace period, and reports to a clearinghouse on its own schedule — so an eligible transfer on one ledger can read as two full fares on another. This is the core of the Transfer Window Logic component inside the broader Fare Rule Validation & Calculation Engines pipeline, and it is written for the transit ops teams, revenue analysts, and Python developers who have to make cross-operator settlement reconcile to the cent. Before any temporal logic runs, both taps are assumed to already be normalized card events from Smart Card Schema Mapping; this page owns only the window decision and the revenue split that follows it.

The Transfer Decision Flow

The evaluator applies ordered guards — sequence, clock drift, delta-versus-window, then the operator boundary — and every path terminates in exactly one status. Ordering matters: an invalid sequence or a drift violation must short-circuit before the window comparison, otherwise a reader whose clock ran backwards would silently manufacture a transfer. The decision flow below traces those guards down to a single terminal status:

Step 1 — Deterministic Timestamp Normalization

AFC logs rarely share identical clock synchronization standards, and a transfer window is only as trustworthy as the two timestamps bounding it. Before applying any temporal logic, normalize every tap to a single UTC reference, reject any timestamp that lacks a timezone, and enforce strict parsing boundaries. Avoid naive string slicing; rely on Python’s standard datetime module for ISO 8601 compliance and explicit timezone resolution. A tap that cannot be parsed deterministically is raised, never guessed — a defaulted timezone is the most common source of phantom transfers across a DST boundary.

import logging
from datetime import datetime, timezone
import re

logger = logging.getLogger("transit_reconciliation")
logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")


class TimestampNormalizationError(Exception):
    """Raised when raw AFC timestamps fail deterministic parsing."""


def normalize_afc_timestamp(raw_ts: str) -> datetime:
    """Parse a raw AFC timestamp to timezone-aware UTC, or raise."""
    # Enforce strict ISO 8601 with an explicit offset or 'Z'
    pattern = r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?([+-]\d{2}:\d{2}|Z)$"
    if not re.match(pattern, raw_ts):
        raise TimestampNormalizationError(f"Malformed timestamp format: {raw_ts}")
    try:
        parsed = datetime.fromisoformat(raw_ts.replace("Z", "+00:00"))
    except ValueError as exc:
        raise TimestampNormalizationError(f"ISO parsing failed: {exc}") from exc
    if parsed.tzinfo is None:
        raise TimestampNormalizationError("Timestamp lacks timezone; refusing to default to UTC.")
    return parsed.astimezone(timezone.utc)

Step 2 — Evaluating the Window

Temporal validation is only the first layer. The implementation below separates the temporal math from the business rules, enforces strict type hints, and emits an immutable audit record for every decision. The effective window is the operator-agreement window plus a grace period, widened by an NTP drift tolerance so that hardware clocks a second or two apart are not penalized. The window and grace values themselves are not hardcoded here — they belong to the per-agreement configuration described in the parent Transfer Window Logic specification and are loaded per operator pair.

from datetime import timedelta
from enum import Enum
from dataclasses import dataclass, field
from typing import List
import hashlib
import uuid


class TransferStatus(Enum):
    ELIGIBLE = "eligible"
    EXPIRED = "expired"
    INVALID_SEQUENCE = "invalid_sequence"
    SAME_OPERATOR = "same_operator"
    CLOCK_DRIFT_EXCEEDED = "clock_drift_exceeded"


@dataclass(frozen=True)
class TapEvent:
    card_id: str
    operator_id: str
    route_id: str
    tap_utc: datetime
    fare_type: str


@dataclass(frozen=True)
class TransferAuditRecord:
    trace_id: str
    first_tap: TapEvent
    second_tap: TapEvent
    delta_seconds: float
    effective_window_seconds: float
    status: TransferStatus
    decision_rationale: str
    evaluated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    audit_hash: str = ""

    def __post_init__(self) -> None:
        # Deterministic idempotency key over the tap pair: reprocessing the same
        # two taps (e.g. on a queue redelivery) yields an identical hash, so
        # downstream ledgers can dedupe without relying on the random trace_id.
        if not self.audit_hash:
            payload = (
                f"{self.first_tap.card_id}|{self.first_tap.operator_id}|"
                f"{self.first_tap.tap_utc.isoformat()}|"
                f"{self.second_tap.operator_id}|{self.second_tap.tap_utc.isoformat()}"
            )
            digest = hashlib.sha256(payload.encode()).hexdigest()
            object.__setattr__(self, "audit_hash", digest)


class TransferWindowEvaluator:
    def __init__(
        self,
        max_window_minutes: int = 90,
        grace_period_seconds: int = 15,
        ntp_tolerance_seconds: float = 2.0,
    ) -> None:
        self.max_window_minutes = max_window_minutes
        self.grace_period_seconds = grace_period_seconds
        self.ntp_tolerance_seconds = ntp_tolerance_seconds
        self.audit_trail: List[TransferAuditRecord] = []

    def evaluate(self, first: TapEvent, second: TapEvent) -> TransferAuditRecord:
        trace_id = str(uuid.uuid4())

        # 1. Sequence validation
        if second.tap_utc <= first.tap_utc:
            return self._log(trace_id, first, second, 0.0, 0.0,
                             TransferStatus.INVALID_SEQUENCE, "Second tap precedes first tap.")

        # 2. Delta and effective window
        delta = (second.tap_utc - first.tap_utc).total_seconds()
        effective_window = (
            timedelta(minutes=self.max_window_minutes)
            + timedelta(seconds=self.grace_period_seconds)
        ).total_seconds()

        # 3. Clock-drift guard (drift below zero or beyond window + tolerance)
        if delta < -self.ntp_tolerance_seconds or delta > effective_window + self.ntp_tolerance_seconds:
            status = TransferStatus.CLOCK_DRIFT_EXCEEDED if delta < 0 else TransferStatus.EXPIRED
            return self._log(trace_id, first, second, delta, effective_window, status,
                             f"Delta {delta:.2f}s outside tolerance/window.")

        # 4. Operator eligibility boundary
        if first.operator_id == second.operator_id:
            return self._log(trace_id, first, second, delta, effective_window,
                             TransferStatus.SAME_OPERATOR, "Intra-operator tap; cross-transfer rules N/A.")

        # 5. Final eligibility
        is_eligible = 0 <= delta <= effective_window
        status = TransferStatus.ELIGIBLE if is_eligible else TransferStatus.EXPIRED
        rationale = ("Within transfer window and grace period." if is_eligible
                     else f"Delta {delta:.2f}s exceeds {effective_window}s window.")
        return self._log(trace_id, first, second, delta, effective_window, status, rationale)

    def _log(self, trace_id: str, first: TapEvent, second: TapEvent, delta: float,
             window: float, status: TransferStatus, rationale: str) -> TransferAuditRecord:
        record = TransferAuditRecord(
            trace_id=trace_id,
            first_tap=first,
            second_tap=second,
            delta_seconds=delta,
            effective_window_seconds=window,
            status=status,
            decision_rationale=rationale,
        )
        self.audit_trail.append(record)
        logger.info("[%s] %s | delta=%.1fs | %s", trace_id, status.value, delta, rationale)
        return record

Revenue reconciliation fails when edge cases are swallowed by silent pass statements. This evaluator instead enforces explicit status enumeration, immutable records, and a deterministic audit_hash idempotency key, so a queue redelivery of the same tap pair dedupes cleanly instead of double-settling.

Step 3 — Prorating the Settlement

An ELIGIBLE result is not the end of the task — a cross-operator transfer means the single collected fare must be split between the originating and accepting operators. The clearinghouse matches the pair, confirms eligibility, and prorates revenue across the two operators:

Given a fare $F$ (in whole cents) and agreed weights $w_i$ per operator leg — distance, agreed flat split, or vehicle-hours — each operator’s raw share is

$$s_i = F \cdot \frac{w_i}{\sum_j w_j}$$

Because integer cents rarely divide evenly, the naive per-operator rounding either loses or invents a cent at settlement. Compute each share with Decimal — never float for money — and reconcile the rounding remainder with the largest-remainder method so the parts always sum back to $F$ exactly.

from decimal import Decimal, ROUND_DOWN
from typing import Dict


def prorate_fare(total_cents: int, weights: Dict[str, Decimal]) -> Dict[str, int]:
    """Split a fare across operators so shares sum exactly to total_cents."""
    if total_cents < 0:
        raise ValueError("Fare cannot be negative.")
    weight_sum = sum(weights.values())
    if weight_sum <= 0:
        raise ValueError("Weights must sum to a positive value.")

    total = Decimal(total_cents)
    floors: Dict[str, int] = {}
    remainders: Dict[str, Decimal] = {}
    for op, w in weights.items():
        exact = total * (w / weight_sum)
        whole = exact.quantize(Decimal("1"), rounding=ROUND_DOWN)
        floors[op] = int(whole)
        remainders[op] = exact - whole

    # Distribute the leftover cents to the largest fractional remainders first.
    leftover = total_cents - sum(floors.values())
    for op in sorted(remainders, key=lambda k: remainders[k], reverse=True)[:leftover]:
        floors[op] += 1
    return floors

Validation & Test Cases

Exercise the evaluator and the proration split against concrete pairs. The normal case is a genuine cross-operator transfer inside the window; the edge cases cover a reversed clock and an expired window.

ev = TransferWindowEvaluator(max_window_minutes=90, grace_period_seconds=15)

bus = TapEvent("CARD001", "MUNI_BUS", "R12",
               normalize_afc_timestamp("2026-07-03T08:00:00Z"), "adult")
rail = TapEvent("CARD001", "REGIONAL_RAIL", "L4",
                normalize_afc_timestamp("2026-07-03T08:40:00Z"), "adult")

# Normal case: 40 min < 90 min window, different operators -> ELIGIBLE
r = ev.evaluate(bus, rail)
assert r.status is TransferStatus.ELIGIBLE
assert r.delta_seconds == 2400.0

# Edge case: second tap before first (reader clock ran backwards) -> INVALID_SEQUENCE
r2 = ev.evaluate(rail, bus)
assert r2.status is TransferStatus.INVALID_SEQUENCE

# Edge case: taps 2 hours apart -> EXPIRED, no transfer granted
late = TapEvent("CARD001", "REGIONAL_RAIL", "L4",
                normalize_afc_timestamp("2026-07-03T10:00:00Z"), "adult")
assert ev.evaluate(bus, late).status is TransferStatus.EXPIRED

# Proration: a 275c fare split 60/40 sums back to 275 exactly (165 + 110)
shares = prorate_fare(275, {"MUNI_BUS": Decimal("60"), "REGIONAL_RAIL": Decimal("40")})
assert shares == {"MUNI_BUS": 165, "REGIONAL_RAIL": 110}
assert sum(shares.values()) == 275

The proration assertion is the one that protects settlement: Decimal("275") * (Decimal("60") / Decimal("100")) is 165.0 and the 40 leg is 110.0, which already sum to 275. On a fare that does not divide cleanly — say 275c split three ways — the largest-remainder pass hands the stray cent to the leg with the biggest fractional part, so the shares never drift a cent above or below the collected fare across millions of settled transfers.

Edge Cases & Debugging for Transit Ops

Temporal eligibility is necessary but not sufficient, and most production incidents live in the gaps between the guards above:

1. Clock skew spikes. If CLOCK_DRIFT_EXCEEDED climbs, validate reader NTP sync against a central time server before touching the tolerance. Widening ntp_tolerance_seconds to mask hardware drift also widens the window an evader can exploit.
2. Duplicate taps. Sub-60-second intra-operator re-taps are reader noise, not transfers; the SAME_OPERATOR guard already excludes them, but log delta < 60 intra-operator pairs so a miswired gate does not masquerade as a transfer.
3. Route topology mismatches. Temporal eligibility does not imply geographic validity. Cross-reference tap coordinates against GTFS-Realtime feeds — kept fresh by the GTFS-RT Realtime Sync pipeline — to reject false-positive transfers on parallel corridors or closed-loop shuttles.
4. Grace-period boundaries. Taps within ±1s of the window edge trip downstream fare-capping bugs. Apply a deterministic rounding policy (floor to the whole second) before evaluation, and keep every boundary decision in the audit_trail for replay.

For high-throughput reconciliation, wrap evaluate() in a batch processor and export audit_trail to immutable columnar storage (Parquet or S3) keyed on audit_hash; the SHA-256 key makes the export idempotent under redelivery.

Integration Note

This task is one leaf of the parent Transfer Window Logic component: the evaluator here produces the ELIGIBLE verdict that capping and concession rules downstream depend on. On the ingestion side it consumes normalized events validated by Implementing Pydantic Models for AFC Event Streams, so the TapEvent fields are guaranteed present before a window is ever computed. Its closest sibling is Building Graceful Degradation for Offline Fare Readers: a validator that loses backhaul cannot see the other operator’s tap, so it approximates this exact rule from a local rolling window and defers the authoritative cross-operator decision — the one computed here — until the clearinghouse reconciles after sync.

FAQ

Related

• Transfer Window Logic — the parent component this window decision belongs to
• Building Graceful Degradation for Offline Fare Readers — how a validator approximates this rule when it cannot see the other operator’s tap
• Implementing Pydantic Models for AFC Event Streams — the schema layer that guarantees a clean TapEvent before evaluation
• Smart Card Schema Mapping — where raw card taps become the normalized events this page consumes
• Dynamic Peak Pricing Threshold Adjustment Scripts — the peak multipliers a transferred fare is re-priced against

Part of Transfer Window Logic, within Fare Rule Validation & Calculation Engines.