GTFS-RT Aggregator

Archives GTFS-RT realtime feeds to Parquet. It fetches each feed on a schedule, groups the fetches into hourly (or other) files, and can compact each day into one sorted file. It can also keep a Parquet copy of every version of a provider's GTFS static feed, expose the compacted days as Iceberg tables, and build a daily TripStopEvent table (one row per trip and stop).

Storage can be a local folder, Google Cloud Storage or MinIO. Everything is configured in one TOML file.

Installation

Python 3.11 or later.

pip install gtfs-rt-aggregator             # realtime feeds only
pip install "gtfs-rt-aggregator[static]"   # also GTFS static feeds and TripStopEvent
pip install "gtfs-rt-aggregator[iceberg]"  # also Iceberg tables
pip install "gtfs-rt-aggregator[punctuality]"  # also final delays per call (DuckDB)

From source:

git clone https://github.com/GaspardMerten/gtfs-rt-aggregator
cd gtfs-rt-aggregator
pip install -e .

Quick start

[storage]
type = "filesystem"
[storage.params]
base_directory = "data"

[[providers]]
name = "ovapi"
timezone = "Europe/Amsterdam"

  [[providers.realtime]]
  url = "https://gtfs.ovapi.nl/nl/vehiclePositions.pb"
  services = ["VehiclePosition"]
  refresh_seconds = 20    # fetch every 20 s
  frequency_minutes = 60  # one aggregated file per hour
  deduplicate = true      # merge identical consecutive rows of a vehicle

  [[providers.realtime]]
  url = "https://gtfs.ovapi.nl/nl/tripUpdates.pb"
  services = ["TripUpdate"]
  refresh_seconds = 20
  [providers.realtime.filter]
  route_types = [2, "100-199"]  # rail only, resolved through the static feed below

  [[providers.static]]
  url = "https://gtfs.ovapi.nl/nl/gtfs-nl.zip"
  check_minutes = 60

[[providers]]
name = "uk"
timezone = "Europe/London"

  [[providers.realtime]]
  url = "https://example.org/uk/gtfsrt.pb"
  services = ["VehiclePosition"]
  [providers.realtime.headers]
  x-api-key = "${UK_API_KEY}"  # read from the environment

gtfs-rt-pipeline configuration.toml

Each job first runs within a minute of startup (see startup_jitter_seconds), then at its interval. The pipeline stops cleanly on Ctrl+C, or SIGTERM sent to its main process. Under systemd, set KillSignal=SIGINT: systemd signals every process of the service, and worker processes end at once on SIGTERM.

Any string in the file can read an environment variable with ${NAME}. Loading fails if the variable is not set. Write $${ for a literal ${.

Configuration reference

Unknown keys in providers, feeds, [output], [runtime], [raw] and [iceberg] are rejected, so typos fail at load time.

[storage]

type is filesystem, gcs or minio. Parameters go in [storage.params]:

[storage]
type = "minio"
[storage.params]
endpoint = "minio.example.com:9000"
access_key = "${MINIO_ACCESS_KEY}"
secret_key = "${MINIO_SECRET_KEY}"
bucket_name = "gtfs-data"

A provider can have its own [providers.storage] with the same fields.

[[providers]]

A provider needs at least one realtime or static feed.

[[providers.realtime]]

[[providers.apis]] is accepted as an old name for [[providers.realtime]].

A provider can have several feeds of the same service, e.g. one per operator. Their rows go to the same aggregated files, and feedId tells them apart; deduplication is done feed by feed. Feeds of the same service must have the same frequency_minutes and deduplicate.

With deduplicate = true, a vehicle standing still for ten minutes, polled every 30 s, is one row instead of twenty. If the entity changes and later returns to an earlier state, that is a new row. So is an entity that was missing from a fetch in between, or seen again after the feed was not read for longer than 3 × refresh_seconds (at least 10 minutes): an outage. With skip_unchanged = true, a fetch identical to the previous one is not stored: lastSeen is then the last stored fetch that had the entity, which can be earlier than the last time the feed showed it. The times of those unchanged fetches are recorded in the metadata of the next file stored (gtfs_rt_aggregator.unchanged_fetch_times, or in a file with no rows every 10 minutes while the feed does not change), so a frozen feed is not taken for an outage. TripUpdate rows are told apart by entity id, trip id and start date (some feeds reuse an entity id across days).

Filter

[providers.realtime.filter] keeps rows matching any of:

• route_types: GTFS route types, as numbers or ranges, e.g. [2, "100-199"] for rail
• route_ids: route ids
• trip_ids: trip ids
• keep_unmatched_added: also keep ADDED, NEW and DUPLICATED trips whose route is not in the static feed (default false, needs route_types or route_ids)

route_types and route_ids are resolved through the trips and routes of the provider's latest static version, so the provider needs a [[providers.static]] feed. Until a static version can be read (the first one is being downloaded, or storage is down), fetches wait in the spool, however long (spool_max_gb bounds it), never stored unfiltered, with a warning every 30 minutes. The status file counts the entities each fetch dropped (dropped_count) and those whose trip the static version does not know (dropped_unknown, also per hour in the health file and logged once per window): many mean the static version is behind the realtime feed. Trip updates and vehicle positions match by trip or route (vehicles without a trip are dropped). Alerts match by any informed route, route type or trip. Other entity types are always kept.

[[providers.static]]

Needs the static extra. A new version is stored only when a file inside the zip changed (and, with min_change, enough trips changed). A file whose Last-Modified is older than the newest stored version's is not stored, nor the content of an earlier version (a publisher serving a stale copy from a second server would make versions flip). A changed route_types reads the file again even if the server says it did not change.

[[providers.static]]
index_url = "https://data.public.lu/api/1/datasets/horaires-et-arrets-des-transport-publics-gtfs/"
url_pattern = 'https://download\.data\.public\.lu/resources/horaires-et-arrets-des-transport-publics-gtfs/\d{8}-\d{6}/gtfs-[\d-]+\.zip'
check_minutes = 1440

Static conversion writes temporary files (about 1.5 GB for a large national feed). If /tmp is in RAM, set TMPDIR to a folder on disk.

Adapters

For a source that is not a GTFS-RT or GTFS URL, such as a JSON API, write a Python function and name it with adapter instead of url: "path/to/file.py:function" (relative to the configuration file) or "module:function".

[[providers.realtime]]
adapter = "adapters/trafikverket.py:fetch"
services = ["TripUpdate"]
refresh_seconds = 30

[[providers.static]]
adapter = "adapters/trafikverket.py:timetable"
check_minutes = 1440

def fetch(state: dict, env) -> FeedMessage | bytes: ...          # one poll, as GTFS-RT
def timetable(state: dict, env, out_dir: Path) -> Path: ...     # a GTFS zip or folder, written in out_dir

What they return is handled like a download: filter, deduplication, versions, and so on. state is the adapter's own dict (JSON values only). It is kept between calls and across restarts, and saved only after a call that succeeds. Use it for things like a change id. env is os.environ, for API keys. Adapters run in the worker processes, and a file is loaded again when it changes. An adapter that raises counts as a failed fetch, and it is called again at the next refresh. feedId is computed from the adapter text as written. After a restart, a static adapter is not called again before check_minutes if a version is already stored.

[output]

compact_daily and trip_stop_events need a path_template with one folder per day and {service} in a folder name.

[runtime]

Fetches are downloaded to a spool folder on disk, converted by worker processes, then uploaded. Anything in the spool is picked up again after a restart, and waits there while storage is down.

[raw]

Keeps every raw fetch, bundled per feed and hour, so the Parquet files can be rebuilt later.

Fetches skipped as byte-identical to the previous one are not archived.

[iceberg]

Needs the iceberg extra and compact_daily = true. The compacted days are registered as they are (no copy) in one Iceberg table per service type, partitioned by provider and date. The tables use the global storage; providers with their own storage are not registered.

[iceberg]
catalog = "sql"
catalog_uri = "sqlite:////var/lib/gtfs/iceberg.db"
services = ["TripUpdate", "VehiclePosition"]

Command line

gtfs-rt-pipeline configuration.toml [options]

Without options, the pipeline runs until stopped. Each of the other options does one job and exits. The exit status is 1 on error.

From Python:

from gtfs_rt_aggregator import run_pipeline_from_toml

run_pipeline_from_toml("configuration.toml")

Output

Layout

ovapi/VehiclePosition/individual/2026-09-28_14-00-20Z-1a2b3c4d.parquet               # one fetch or block (UTC, feed id)
provider=ovapi/service=VehiclePosition/date=2026-09-28/16-00-00_to_17-00-00.parquet  # aggregated period (local time)
provider=ovapi/service=VehiclePosition/date=2026-09-27/day.parquet                   # compacted day
ovapi/raw/provider=ovapi/feed=VehiclePosition-1a2b3c4d/date=2026-09-28/14-1790000000.tar.zst  # raw archive
ovapi/static/2026-09-28_01-00-00Z/stops.parquet    # one static version, one file per table
ovapi/static/2026-09-28_01-00-00Z/manifest.json
ovapi/static/latest.json                           # manifest of the latest version
ovapi/_status/VehiclePosition-1a2b3c4d.json        # status of one realtime feed
ovapi/_health/VehiclePosition-1a2b3c4d/date=2026-09-28/20260928T080000Z.json  # fetches per local hour (one file per run)
_status/index.json                                 # summary of all feeds (global storage)

A period is aggregated once a file from the next period exists, or 5 minutes after it ended. Files that arrive later are added to the period's file.

Each feed's status file holds the last attempt, last success and last error, the feed's age, and the entity count before and after the filter. URLs are stored without their query string.

Each feed's health file counts, per local hour, the fetches read (ok), those whose entities changed (changed) and those that failed (failed): an hour with reads but no change is a frozen feed, an hour without reads an outage. Filtered feeds also count dropped_unknown (see Filter). Each run of the process writes its own file: add them up for a day.

At INFO, the pipeline logs one summary line every 10 minutes (fetches, changes, failures, skips, spool), plus warnings and errors; per-fetch and per-period lines are DEBUG. A failed request logs the start of the server's answer.

Realtime columns

Each row has the entity's fields plus:

• provider, date: the provider, and the local date of the fetch
• fetchTime: fetch time (timestamp, UTC)
• feedTimestamp: time in the feed header (timestamp, UTC)
• staticVersion: the provider's static version at fetch time, to join with the right timetable
• feedId: 8 characters identifying the realtime feed the row comes from
• contentHash: hash of the entity
• firstSeen, lastSeen: with deduplicate = true, after aggregation (timestamps, UTC)

Times inside entities, such as a vehicle's timestamp, stay Unix seconds as in GTFS-RT. There are no unsigned integer columns.

Reading with DuckDB

SELECT *
FROM read_parquet('data/provider=*/service=VehiclePosition/date=*/*.parquet',
                  hive_partitioning = true, union_by_name = true)
WHERE date = '2026-09-28';

Iceberg tables, without a catalog:

INSTALL iceberg; LOAD iceberg;
SELECT provider, date, count(*) FROM iceberg_scan('s3://my-bucket/iceberg/TripUpdate') GROUP BY ALL;

Or over https, with public_base_url set:

-- If the server needs a token: CREATE SECRET (TYPE http, BEARER_TOKEN '...');
SELECT count(*) FROM iceberg_scan('https://data.example.org/iceberg-public/TripUpdate') WHERE provider = 'be';

TripStopEvent

With trip_stop_events = true, each provider with trip updates and a static feed gets one file per service date, e.g. provider=be/service=TripStopEvent/date=2026-10-24/day.parquet. It has one row per trip and stop:

A service date D is built once D+1 is over and aggregated (and compacted, with compact_daily), from the trip updates of D-1, D and D+1. It is built again when those trip updates change, e.g. late files. Feeds that give only delays or only times both work; the missing value is computed from the schedule. Canceled trips keep their scheduled stops without times. Added trips that are not in the timetable keep their updates without a schedule.

-- Average arrival delay per route, from the last prediction of observed stops
SELECT route_id, avg(last_arrival_delay) / 60 AS minutes
FROM read_parquet('data/provider=be/service=TripStopEvent/date=2026-10-24/day.parquet')
WHERE observed
GROUP BY ALL
ORDER BY minutes DESC;

Final delays (punctuality)

gtfs_rt_aggregator.punctuality.final_calls gives the final delay of every trip at every planned call of a service day, from the stored TripUpdate files and timetable versions (pip install "gtfs-rt-aggregator[punctuality]").

Use it or TripStopEvent:

• TripStopEvent runs in the pipeline, keeps the first and last prediction per stop, and resolves runs from the schedule, so it suits buses and frequency-based feeds that follow the spec.
• final_calls runs on a finished day, keeps only the last value, and is built for rail feeds that bend the spec. It needs a trip_id per trip and does not support frequency-based trips. The service day is trip.startDate; when the feed sends none, it is the local day of the fetch, except that a fetch after midnight goes to the previous day's run of a trip the timetable runs past midnight.

What final_calls does:

• Updates are matched by stop, or by another platform of the same station. When the train calls there twice, the call numbered as the update wins, then the nearest scheduled time.
• The other updates are matched by stop_sequence, read in whichever way fits the live times best: as the timetable's own numbering, as the position in the trip, or as the rank among stops where passengers can board or alight.
• Each trip uses the timetable version its rows name (staticVersion), else default_version. A trip missing from that version uses the newest version that has it. A trip id ending in a date and missing from the timetable takes the only trip with the same id up to that date running on the day.
• Feeds sending only times get delays from the schedule. A time dated a whole day off is moved to the right day.
• Planned stops the feed skipped after its first update take the previous delay (delay_source = "propagated"), up to a stop marked NO_DATA.
• A stop marked SKIPPED keeps no delay or predicted time, and is not a delay later stops take.
• Canceled trips get every planned stop where passengers can board or alight, without delays. Updates of added trips are kept without a schedule.
• timetable_version names the version each train was matched to: resolve its stop and route names in that version, as ids can change meaning between versions.
• When one fetch holds a stop twice, the later row of the files wins, so a day always gives the same result.

from gtfs_rt_aggregator.punctuality import final_calls, storage_timetable

calls, figures = final_calls(
    ["data/provider=be/service=TripUpdate/date=2026-10-24/day.parquet",
     "data/provider=be/service=TripUpdate/date=2026-10-25/day.parquet"],  # trains running past midnight
    "2026-10-24", "Europe/Brussels",
    storage_timetable(storage, "be"),  # or any function (version, table, dest_path) -> bool
    work_dir="/tmp/punctuality",       # one folder per concurrent call
    default_version="2026-10-20_06-00-00Z",  # for rows without staticVersion
    memory_limit="1GB", threads=2,     # optional, DuckDB's defaults otherwise
)

Files must be local paths: download them first from remote storage.

calls (Arrow, sorted by trip and stop_sequence) has one row per call:

• trip_id, stop_sequence, stop_id, service_date (YYYYMMDD) and feed_id
• scheduled_*, predicted_* and observed_at, in Unix seconds
• arrival_delay and departure_delay, in seconds
• delay_source
• the stop's and the trip's schedule relationship, route_id and trip_start_time

figures tells how well the feed matched:

• updates_matched, trips_in_timetable and trips_added: shares from 0 to 1, or None without data
• stop_sequence_mode ("full", "position", "commercial" or None) and stop_sequence_scores
• timetable_version (the default_version) and timetable_versions (the versions used)
• trips_matched_by_id_before_date and propagated_calls: counts

License

MIT