Reconciling GTFS-RT Vehicle Positions with Tap Events

This page solves one attribution problem: given a rider’s tap and a stream of GTFS-Realtime VehiclePosition reports, decide which vehicle and trip the rider was actually on by matching in time and in space. A tap that lands within a tolerance of a vehicle’s reported position is attributed to that vehicle’s trip; a tap with no vehicle anywhere near it in either dimension is flagged a phantom — a stuck reader, a replayed event, or a clock fault — rather than silently charged to the nearest guess. This is the spatiotemporal core of GTFS-Realtime Sync within the Fare Data Ingestion & GTFS-RT Sync pipeline.

Attribution is where fare revenue either gets a defensible trip assignment or leaks into an “unassigned” bucket no month-end close can reconstruct. Downstream fare logic — capping, transfer eligibility, operator settlement — all key on trip_id, so a tap without one is dead weight. The matcher here is deliberately conservative: it would rather flag a tap for review than fabricate a trip assignment that pollutes settlement.

Attribution flow

The matcher takes one tap and the rolling set of recent vehicle positions, filters to candidates that are close in both time and space, scores them, and emits either an attributed trip or a phantom flag. The dual gate is the point: passing only one dimension is not enough, because a vehicle parked at a depot is spatially near a phantom tap but temporally stale.

Step 1 — Modeling taps and positions

Both sides of the match need a typed, timezone-safe shape. A tap carries its own UTC timestamp and coordinates; a GTFS-RT vehicle position carries the same plus the trip_id and vehicle_id we want to copy onto the tap. GTFS-RT timestamps are POSIX epoch seconds by spec, so the model normalizes them to timezone-aware UTC at the boundary rather than trusting device clocks.

import logging
import math
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Optional

logger = logging.getLogger("afc_reconcile")


@dataclass(frozen=True, slots=True)
class TapEvent:
    tap_id: str
    tap_utc: datetime          # timezone-aware UTC
    latitude: float
    longitude: float


@dataclass(frozen=True, slots=True)
class VehiclePosition:
    vehicle_id: str
    trip_id: str
    route_id: str
    reported_utc: datetime     # timezone-aware UTC
    latitude: float
    longitude: float


@dataclass(frozen=True, slots=True)
class Attribution:
    tap_id: str
    trip_id: Optional[str]
    vehicle_id: Optional[str]
    is_phantom: bool
    time_delta_sec: Optional[float] = None
    distance_m: Optional[float] = None


def to_utc(epoch_seconds: int) -> datetime:
    """Normalize a GTFS-RT POSIX timestamp to timezone-aware UTC."""
    return datetime.fromtimestamp(epoch_seconds, tz=timezone.utc)

Using frozen, slotted dataclasses keeps each record immutable and cheap, which matters when the rolling index holds tens of thousands of positions during a peak-hour surge. The Attribution result is explicit about the phantom case: trip_id is None and is_phantom is True, so no downstream stage can mistake an unmatched tap for a real trip.

Step 2 — The nearest-in-time-and-space matcher

The matcher applies both tolerances as hard gates, then picks the surviving candidate with the smallest combined distance. Geography uses the Haversine great-circle distance so the metric is meters, not degrees. Every rejection path is explicit and logged; there is no bare except swallowing a bad coordinate.

class TapReconciler:
    def __init__(
        self,
        time_tolerance_sec: float = 60.0,
        distance_tolerance_m: float = 150.0,
    ) -> None:
        self.time_tolerance_sec = time_tolerance_sec
        self.distance_tolerance_m = distance_tolerance_m

    @staticmethod
    def _haversine_m(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
        """Great-circle distance in meters between two WGS-84 points."""
        r = 6_371_000.0  # Earth radius, meters
        p1, p2 = math.radians(lat1), math.radians(lat2)
        dphi = math.radians(lat2 - lat1)
        dlmb = math.radians(lon2 - lon1)
        a = math.sin(dphi / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(dlmb / 2) ** 2
        return 2 * r * math.asin(math.sqrt(a))

    def match(self, tap: TapEvent, positions: list[VehiclePosition]) -> Attribution:
        """Attribute a tap to the nearest vehicle within both tolerances, or flag phantom."""
        best: Optional[VehiclePosition] = None
        best_distance = float("inf")
        best_dt = 0.0

        for pos in positions:
            dt = abs((tap.tap_utc - pos.reported_utc).total_seconds())
            if dt > self.time_tolerance_sec:
                continue  # temporally stale: parked or lagged report

            try:
                dist = self._haversine_m(tap.latitude, tap.longitude,
                                         pos.latitude, pos.longitude)
            except ValueError as exc:
                # Out-of-range coordinate (e.g. NaN sentinel); skip, do not crash.
                logger.warning("Bad coordinate on vehicle %s: %s", pos.vehicle_id, exc)
                continue

            if dist > self.distance_tolerance_m:
                continue  # spatially too far to be the boarded vehicle

            if dist < best_distance:
                best, best_distance, best_dt = pos, dist, dt

        if best is None:
            logger.info("Phantom tap %s: no vehicle within %.0fs / %.0fm",
                        tap.tap_id, self.time_tolerance_sec, self.distance_tolerance_m)
            return Attribution(tap.tap_id, None, None, is_phantom=True)

        logger.info("Tap %s -> trip %s (dt=%.1fs, d=%.0fm)",
                    tap.tap_id, best.trip_id, best_dt, best_distance)
        return Attribution(
            tap_id=tap.tap_id,
            trip_id=best.trip_id,
            vehicle_id=best.vehicle_id,
            is_phantom=False,
            time_delta_sec=best_dt,
            distance_m=best_distance,
        )

Both gates are continue guards, so a candidate has to clear time and space to compete; the winner is the spatially nearest among those survivors. Ordering the time gate first is a small optimization — the timestamp comparison is cheaper than a Haversine call, so temporally stale positions are discarded before any trig runs. In production the positions list is not the whole feed but a rolling index pruned to the time tolerance, keyed for fast spatial lookup; the parent sync component maintains that index and can back the spatial predicate with a PostGIS ST_DWithin query at scale.

Validation & Test Cases

The two behaviors that matter: a tap next to its vehicle at the right time attributes to the correct trip, and a tap with no vehicle nearby is flagged phantom instead of forced onto a wrong trip.

reconciler = TapReconciler(time_tolerance_sec=60.0, distance_tolerance_m=150.0)

# A vehicle on trip T-100 reports near a downtown stop at 08:00:00Z.
positions = [
    VehiclePosition("BUS-7", "T-100", "R-42", to_utc(1783065600), 40.7128, -74.0060),
    # A different vehicle, far away and stale (a depot report).
    VehiclePosition("BUS-9", "T-200", "R-99", to_utc(1783065000), 40.8000, -74.2000),
]

# 1. Normal case: tap ~20 m away, 15 s later -> attributed to trip T-100.
good_tap = TapEvent("tap-1", to_utc(1783065615), 40.71282, -74.00615)
result = reconciler.match(good_tap, positions)
assert result.is_phantom is False
assert result.trip_id == "T-100" and result.vehicle_id == "BUS-7"
assert result.distance_m < 150 and result.time_delta_sec <= 60

# 2. Phantom case: tap in the harbor, no vehicle within tolerance -> flagged.
phantom_tap = TapEvent("tap-2", to_utc(1783065600), 40.6000, -74.0500)
flagged = reconciler.match(phantom_tap, positions)
assert flagged.is_phantom is True
assert flagged.trip_id is None and flagged.vehicle_id is None

# 3. Edge: a spatially-close but temporally-stale vehicle must NOT match.
stale_only = [VehiclePosition("BUS-3", "T-300", "R-1", to_utc(1783060000),
                              40.71282, -74.00615)]   # same spot, ~93 min earlier
assert reconciler.match(good_tap, stale_only).is_phantom is True

The first case attributes cleanly because BUS-7 clears both gates and is the nearest survivor; BUS-9 is excluded on both time and distance. The phantom tap in case 2 has no candidate within either tolerance, so it returns is_phantom=True with no trip rather than being snapped to the closest far-away vehicle. Case 3 is the dual-gate proof: a vehicle sitting on the exact coordinates but reporting 93 minutes earlier still fails the time gate, so proximity alone never manufactures a match.

Edge Cases & Gotchas

• GPS drift near terminals. Vehicles bunch together and coordinates jitter at depots and layovers; tighten the distance tolerance or add a route-consistency check so a tap is not attributed to a bus idling on the next platform.
• Future-dated taps. A tap timestamped ahead of every vehicle report is a clock fault, not a valid event; the time gate flags it phantom, which is the correct conservative outcome — investigate the reader’s NTP sync.
• Multiple candidates within tolerance. On dense corridors two vehicles can both qualify; nearest-in-space is a reasonable default, but break ties with the most recent report and a matching route_id before trusting the assignment.
• Feed gaps. During a GTFS-RT outage there are no positions to match against; route those taps to the static-schedule fallback rather than flagging a phantom storm, so a feed outage becomes delayed attribution, not lost revenue.
• Tolerances are agency-specific. ±60 s / ±150 m suits urban bus; rail with sparse position reports or rural routes with long headways need wider windows. Load them per mode, never hardcode one pair network-wide.

Integration Note

This matcher is the attribution engine inside GTFS-Realtime Sync: it consumes the validated positions that stage decodes and produces the trip_id that anchors a tap to real service. Its output feeds Transfer Window Logic directly — a cross-operator transfer decision is only trustworthy once each leg is attributed to a specific trip, so the phantom flag here prevents a stuck reader from fabricating a second leg the transfer evaluator would otherwise price. Taps flagged phantom follow the same dead-letter and fallback paths the sync stage already defines, so an unmatched tap is quarantined for review rather than dropped. For authoritative field semantics on every entity decoded upstream, consult the GTFS Realtime Specification.

FAQ

Related

• GTFS-Realtime Sync — the parent component that decodes and indexes the positions this matcher consumes
• Transfer Window Logic — downstream logic that needs the attributed trip before pricing a transfer
• Calculating Cross-Operator Transfer Windows with Python — the transfer decision that a phantom flag protects from a fabricated leg
• Schema Validation Pipelines — the stage that guarantees a clean tap before it reaches this matcher
• Mapping Multi-Modal Fare Zones to PostGIS Polygons — the spatial index that backs the distance predicate at scale

Part of GTFS-Realtime Sync, within Fare Data Ingestion & GTFS-RT Sync.