AFC API Data Extraction

Automated Fare Collection (AFC) systems generate the financial and operational backbone of modern transit networks. Extracting this data reliably requires more than simple HTTP polling; it demands a disciplined pipeline architecture that aligns fare telemetry with service performance. Within the broader Fare Data Ingestion & GTFS-RT Sync framework, API extraction serves as the critical first hop, transforming vendor-specific endpoints into structured, auditable records for revenue reconciliation and ridership analytics.

Memory-Efficient Async Extraction

Transit operators and mobility developers must navigate heterogeneous vendor APIs, each with distinct authentication schemes, pagination models, and rate ceilings. When tap volumes spike during peak hours or special events, synchronous extraction quickly becomes a bottleneck. The production standard is async batching with strict backpressure controls. Buffering raw JSON payloads in-memory without streaming to disk or a message broker will inevitably crash ingestion workers during end-of-day reconciliation runs.

Implement a producer-consumer architecture using bounded asyncio.Queue objects and chunked HTTP responses. Stream payloads directly to NDJSON files or a lightweight broker to maintain a constant memory footprint regardless of transaction volume.

The sequence below shows the producer-consumer flow that keeps memory bounded under peak load:

import asyncio
import aiohttp
from typing import AsyncGenerator

async def stream_tap_batches(
    session: aiohttp.ClientSession, 
    base_url: str, 
    params: dict,
    batch_size: int = 500
) -> AsyncGenerator[dict, None]:
    """Fetches paginated AFC data and yields records without full in-memory buffering."""
    offset = 0
    while True:
        async with session.get(base_url, params={**params, "limit": batch_size, "offset": offset}) as resp:
            resp.raise_for_status()
            # Buffer only one bounded page at a time, then yield records individually
            payload = await resp.json()
            if not payload.get("transactions"):
                break
                
            for tx in payload["transactions"]:
                yield tx
                
            offset += batch_size
            # Yield control to event loop to prevent starvation
            await asyncio.sleep(0)

For large-scale deployments, pair this with Python’s asyncio documentation best practices for task grouping, graceful cancellation, and semaphore-limited concurrency to prevent socket exhaustion.

Rate Limiting & Resilience Patterns

Legacy back-office systems often expose REST or SOAP endpoints that lack native streaming capabilities, forcing engineers to implement Handling Rate Limits on Legacy AFC Vendor APIs through exponential backoff, token rotation, and request windowing. Transit APIs frequently return 429 Too Many Requests or 503 Service Unavailable during farebox sync windows.

Production extractors must implement jittered backoff, circuit breakers, and idempotent request signatures. The tenacity library provides a clean abstraction for retry policies, but you must explicitly handle vendor-specific rate limit headers (X-RateLimit-Remaining, Retry-After).

import httpx
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type

@retry(
    retry=retry_if_exception_type((httpx.HTTPStatusError, httpx.ConnectError)),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    stop=stop_after_attempt(5),
    reraise=True
)
async def resilient_fetch(client: httpx.AsyncClient, endpoint: str, headers: dict) -> dict:
    resp = await client.get(endpoint, headers=headers)
    if resp.status_code == 429:
        # Surface the server's Retry-After hint for observability; tenacity applies
        # exponential backoff between attempts via the wait policy above.
        retry_after = resp.headers.get("Retry-After", "2")
        raise httpx.HTTPStatusError(
            f"Rate limited. Retry-After: {retry_after}s",
            request=resp.request,
            response=resp
        )
    resp.raise_for_status()
    return resp.json()

Strict Schema Validation & Quarantine Routing

Raw fare transactions are meaningless without spatial and temporal context, but they are also dangerous without strict typing. Every extracted payload must pass through a validation gate that enforces data types, required fields (fare_media_type, tap_timestamp, origin_stop_id), and business rules (e.g., no negative balances, sequential tap constraints, valid media types). Pydantic v2 is the industry standard for this layer due to its compiled validation speed and explicit error serialization.

Invalid records must be quarantined for manual review rather than poisoning downstream revenue models. Implement a dual-sink routing pattern that separates clean data from malformed payloads.

The dual-sink validation gate below routes each payload to exactly one destination:

from pydantic import BaseModel, ValidationError, field_validator
from datetime import datetime
import logging
from typing import Protocol

class TapSink(Protocol):
    async def __call__(self, record: dict) -> None: ...

class TapRecord(BaseModel):
    transaction_id: str
    tap_timestamp: datetime
    device_id: str
    origin_stop_id: str
    fare_media_type: str
    amount: float

    @field_validator("amount")
    @classmethod
    def enforce_non_negative(cls, v: float) -> float:
        if v < 0:
            raise ValueError("Negative fare amount violates revenue integrity rules")
        return v

async def validate_and_route(
    raw: dict, 
    valid_sink: TapSink, 
    quarantine_sink: TapSink
) -> None:
    try:
        parsed = TapRecord.model_validate(raw)
        await valid_sink(parsed.model_dump(mode="json"))
    except ValidationError as e:
        logging.warning(f"Quarantined tx {raw.get('transaction_id', 'UNKNOWN')}: {e}")
        await quarantine_sink({**raw, "validation_error": str(e)})

Refer to the Pydantic v2 Documentation for advanced validators, custom error formatting, and integration with message brokers like Kafka or Redis Streams.

GTFS Alignment & Reconciliation Logic

Extracted tap records must be cross-referenced against scheduled and real-time vehicle positions. The GTFS-RT Realtime Sync component provides the trip_id, vehicle_id, and timestamp anchors needed to map fare events to actual service delivery. However, AFC APIs rarely return GTFS-aligned identifiers natively. Engineers must implement deterministic mapping tables or fuzzy matching logic to reconcile farebox device IDs with GTFS stop_id and route_id values.

Scalable reconciliation requires:

1. Timezone Normalization: Convert all tap_timestamp values to UTC immediately upon ingestion. DST shifts frequently cause duplicate or missing revenue attribution.
2. Spatial Proximity Matching: When GPS drift occurs, match taps to the nearest valid stop_id within a configurable radius (e.g., 50m) using Haversine distance or PostGIS ST_DWithin.
3. Temporal Windowing: Join taps to GTFS-RT vehicle positions using a sliding window (tap_timestamp ± 90s) to account for farebox sync latency and cellular network delays.

For authoritative GTFS-RT implementation details, consult the official GTFS Realtime Specification.

Legacy Fallbacks & Batch Processing

Not all vendors support modern JSON APIs. Many legacy fare systems still rely on flat-file exports or batched CSV dumps delivered via SFTP or secure web portals. Integrating these sources requires robust CSV Batch Parsing Workflows that handle encoding anomalies, malformed delimiters, and partial row writes.

When processing multi-gigabyte daily dumps, avoid pandas.read_csv() on the entire file. Use chunked iterators or polars.scan_csv() to maintain O(1) memory complexity. Always validate row counts against vendor-provided manifests before committing to the reconciliation database.

Production Deployment Checklist

• Implement bounded async queues and NDJSON streaming to prevent OOM during peak tap surges
• Configure exponential backoff with jitter and explicit Retry-After Configure exponential backoff with jitter and explicit `Retry-After` header parsing
• Enforce Pydantic schema validation at the ingestion boundary; route failures to a dedicated quarantine table
• Normalize all timestamps to UTC and apply timezone-aware business rules before reconciliation
• Maintain a versioned device_id → stop_id Maintain a versioned `device_id` → `stop_id` mapping table with audit trails for hardware swaps
• Run daily reconciliation diffs: SUM(validated_amounts) vs Run daily reconciliation diffs: `SUM(validated_amounts)` vs `vendor_settlement_report`
• Instrument extraction latency, 429/5xx rates, and quarantine volume with Prometheus/Grafana dashboards