A production-structured MVP that continuously ingests the Toronto Transit Commission
GTFS-Realtime VehiclePositions feed, persists
fetch metadata and normalized relational rows append-only in PostgreSQL
(with PostGIS for geom), exposes query APIs via FastAPI, and visualizes feed health +
vehicle movement in a Streamlit dashboard. An apps/analytics/ pipeline joins the live
feed against static GTFS to produce per-trip upsampled trajectories.
The layout is intentionally structured so that later additions (VehiclePositions, Alerts, static GTFS joins, map views, analytics) drop in without touching the existing data model.
┌─────────────┐ every 30 s (HTTP GET) ┌──────────────┐
│ Collector │ ─────── protobuf ─────────▶│ TTC feed │
│ (asyncio) │◀──────── bytes ─────── └──────────────┘
└──────┬──────┘
│ single txn per fetch:
│ 1. feed_fetch_logs (always; success or failure)
│ 2. raw_gtfsrt_snapshots (on success)
│ 3. trip_updates (normalized)
│ 4. trip_update_stop_times
▼
┌───────────────┐ ┌──────────────┐ ┌─────────────────┐
│ PostgreSQL │◀── SQL ─│ FastAPI │◀─ HTTP ─│ Streamlit │
│ (TIMESTAMPTZ │ │ (SQLAlchemy │ JSON │ dashboard │
│ + BYTEA) │ │ 2.x) │ │ (4 pages) │
└───────────────┘ └──────────────┘ └─────────────────┘
Three long-running processes plus Postgres, orchestrated by docker-compose.
- Append-only — every poll writes new rows; "latest" is a query, not a mutation.
- Normalized + provenance — each
FeedEntity.vehicleis normalized intovehicle_positionswith its decoded JSON kept inraw_entity, so the feed can be re-normalized later if the parser changes. Araw_gtfsrt_snapshotsmetadata row records each fetch'scontent_sha256; the raw protobuf bytes are not persisted. - Clean module seams —
db/,core/,apps/api,apps/collector,apps/dashboard. Eachapps/*entry depends oncore+dbbut not on each other. - Sync SQLAlchemy 2.x with the typed
Mapped[X]style. Async adds transaction complexity with no latency benefit at 2 polls / minute.
cp .env.example .env
make up # builds images, starts postgres + migrator + api + collector + dashboardThen:
- API http://localhost:8000/health
- Dashboard http://localhost:8501
- Postgres
localhost:5433(userttc/ pwdttc/ dbttc_gtfsrt)
Shut down with make down. Wipe data with docker compose down -v.
The test suite expects a Postgres instance reachable via TEST_DATABASE_URL
(falls back to DATABASE_URL). The default .env points at the compose-exposed
port 5433, which works once make up is running:
pip install -e '.[dev]'
make testTests marked as requiring the DB are auto-skipped if the server is unreachable.
apps/
api/ FastAPI app, routers, pydantic schemas
collector/ httpx fetcher, protobuf parser, pure normalizer, runner (txn owner), CLI
dashboard/ Streamlit multipage app + thin API client
core/ settings, logging, time helpers, constants
db/ Base, session factory, ORM models, query helpers, alembic migrations
tests/ parser / normalizer / query / API smoke tests + recorded fixture
scripts/ capture_sample.py (record a live feed payload for fixtures)
Five tables. The raw-ingest trio is created by 0001_initial_schema.py; the
vehicle_positions pivot is 0002_pivot_to_vehicle_positions.py; the analytics
output tables land in 0003_trip_trajectories.py:
| table | purpose | key columns |
|---|---|---|
feed_fetch_logs |
every fetch attempt — success or failure | feed_name, fetched_at, success, http_status, duration_ms, entity_count, error_type, error_message |
raw_gtfsrt_snapshots |
fetch metadata per successful fetch (1:1 with the log row) | fetch_log_id UNIQUE, content_sha256, feed_header_timestamp |
vehicle_positions |
one row per FeedEntity.vehicle per snapshot |
snapshot_id, trip_id, route_id, vehicle_id, latitude, longitude, speed_mps, occupancy_status, denormalized fetched_at, PostGIS geom(Point, 4326) GENERATED |
analytics_runs |
one row per apps/analytics invocation |
service_date, route_id, status, rows_written, config_json, started_at, finished_at |
trip_trajectories |
upsampled per-trip trajectory points | run_id, trip_id, start_date, service_date, datetime, travel_distance_m, moving_speed_m_s, observed |
The hottest query is "most recent row per vehicle_id" or "per route_id":
SELECT DISTINCT ON (vehicle_id) *
FROM vehicle_positions
ORDER BY vehicle_id, fetched_at DESC;Joining every lookup back to raw_gtfsrt_snapshots to pull fetched_at would
destroy the (vehicle_id, fetched_at DESC) index's value. Denormalization is
safe because both tables are append-only — fetched_at cannot drift.
Each poll is a distinct observation, even when content_sha256 is identical.
Preserving every poll keeps the monitoring signal "feed is reachable but hasn't
changed" — dedup would silently hide that.
All routes return JSON. Timestamps are ISO-8601 UTC.
| route | purpose |
|---|---|
GET /health |
liveness + db-ok |
GET /feed-status/trip-updates |
last fetch time, success rate (1h), current lag vs feed header |
GET /trip-updates/latest?route_id=&limit= |
DISTINCT ON (trip_id) most recent observation, optionally route-filtered |
GET /trips/{trip_id}/latest |
latest row for a trip + its stop-time updates |
GET /trips/{trip_id}/history?start=&end=&limit= |
append-ordered history for a trip in a time window |
GET /routes/{route_id}/active-trips?window_minutes=&limit= |
trips seen on a route within a recent window |
GET /routes/{route_id}/trip-updates/latest?limit= |
/trip-updates/latest scoped to one route |
GET /replay/trips?start=&end=&route_id=&limit= |
raw append-ordered rows over an arbitrary window |
Example:
curl 'http://localhost:8000/feed-status/trip-updates' | jq
curl 'http://localhost:8000/trip-updates/latest?limit=5' | jq
curl --data-urlencode 'start=2026-04-21T22:00:00+00:00' \
--data-urlencode 'end=2026-04-21T22:30:00+00:00' \
--get 'http://localhost:8000/replay/trips' | jq '.[0]'Streamlit multipage app at http://localhost:8501, backed entirely by the API (no direct DB reads from the dashboard):
- Home — platform summary: fetches today, success rate, current lag.
- Feed Health — per-minute success/failure bar chart and recent fetch log rows.
- Route Explorer — pick a route, see active trips and a delay distribution.
- Trip Detail — latest stop-time updates table + delay-over-time line chart.
- Replay — date-range + optional route filter; rows-per-minute chart + paginated table.
Configured entirely via environment variables (see .env.example). Important knobs:
| var | default | note |
|---|---|---|
DATABASE_URL |
postgresql+psycopg://ttc:ttc@postgres:5432/ttc_gtfsrt |
psycopg v3 dialect |
GTFS_RT_VEHICLE_POSITIONS_URL |
https://gtfsrt.ttc.ca/vehicles/position?format=binary |
?format=binary matters — see below |
COLLECTOR_INTERVAL_SECONDS |
20 |
TTC vehicles update ~every 20 s |
COLLECTOR_HTTP_RETRIES |
2 |
retry connect/timeout/5xx |
COLLECTOR_ROUTE_ALLOWLIST |
(empty) | Comma-separated route_id list. Empty ⇒ ingest every route. When set, the normalizer drops entities whose trip.route_id is missing or not in the list. |
ACTIVE_VEHICLE_WINDOW_MINUTES |
5 |
staleness bound for "active" vehicle queries |
MAX_PAGE_SIZE |
5000 |
hard cap on any limit= parameter |
ANALYTICS_UPSAMPLE_RESOLUTION_S |
10 |
upsampling grid for trip trajectories |
ANALYTICS_MAX_ORTHOGONAL_DISTANCE_M |
200 |
drop trajectory points farther than this from the shape |
ANALYTICS_WORKER_INTERVAL_SECONDS |
120 |
analytics worker tick; smaller = fresher trajectories, higher load |
ANALYTICS_WORKER_SERVICE_DATE_TZ |
America/Toronto |
used by the worker to compute "today" |
LOG_JSON |
true |
structured JSON logging |
Without ?format=binary, the TTC endpoint returns protobuf text format
(~5.6 MB text/plain) rather than binary (~730 KB application/x-protobuf).
The collector defaults to the binary form; the parser also falls back to text
format if the server ever switches back, so the system remains resilient either
way.
- Append-only schema — "latest" is always a query, never an upsert. Implemented
with
DISTINCT ON (key) ORDER BY key, fetched_at DESC, served by a composite index. Upgradable to a materialized view later without an API change. - Raw protobuf stored in Postgres BYTEA — one blob per successful fetch, FK-linked to its fetch log. Object storage (S3/MinIO) is the obvious Phase-2 swap and only touches the collector + a future replay endpoint.
- Failed fetches log a row, no snapshot row — keeps the FK clean (no dangling snapshot without bytes) while still preserving the monitoring signal.
- Normalization is a pure function —
apps/collector/normalizer.pyhas no DB side effects; the transaction is opened inrunner.py. Makes unit tests trivial. - Sync SQLAlchemy over async — polling rate is ~2/min; API is low-QPS internal.
- TIMESTAMPTZ everywhere (UTC). Only the dashboard converts to
America/Torontofor display. - Streamlit calls the FastAPI (never the DB directly) — a future React/Next rewrite doesn't require data-layer changes.
vehicle_positionsis the dominant on-disk cost (~6.5M rows/day, chiefly theraw_entityJSONB column). The raw protobuf payload was dropped in migration0005. Retention / partitioning comes in Phase 2.- No auth on the API. Suitable for internal use or behind a reverse proxy.
- No Prometheus metrics or tracing yet.
- Static GTFS (under
Complete GTFS/) is consumed byapps/analytics/but not yet materialized into dedicated reference tables — joins happen in pandas at runtime.
Each bullet drops in without touching the existing model:
- VehiclePositions — new
feed_name="vehicle-positions"constant, reuseraw_gtfsrt_snapshots, addvehicle_positionstable mirroringtrip_updatesstructure. - Alerts — new
feed_name="alerts", newalertstable. - Static GTFS join — load
routes/stops/tripsfromComplete GTFS/and add name-resolution endpoints; the string keys already match. - Map view — blocked only on VehiclePositions; the dashboard's
api_client.pyis already ready to feed pydeck / folium. - Partitioning + retention —
0004_partition_trip_updates_by_month.pydeclarative partitioning once retention pressure appears. - Materialized latest view —
0005_latest_view.pyif theDISTINCT ONquery becomes a bottleneck.
A batch pipeline that projects every vehicle_positions GPS point onto the
trip's GTFS static shape, derives travel_distance_m/moving_speed_m_s, and
upsamples to a fixed resolution. Output lands in trip_trajectories (one row
per upsampled point, observed=True/False) with a parent analytics_runs row
per invocation.
make analytics-run DATE=2026-04-20 # all routes (full refresh)
make analytics-run DATE=2026-04-20 ROUTE=29 # one route
python -m apps.analytics.main --date 2026-04-20 \
--export-csv ./out/2026-04-20 # also emit legacy-style CSVs
python -m apps.analytics.main --date 2026-04-20 \
--since 2026-04-20T12:00:00+00:00 # incremental: only trips with new VP dataAn optional long-running service in compose that re-runs the analytics for
"today" every ANALYTICS_WORKER_INTERVAL_SECONDS, scoped to trip instances
whose raw vehicle_positions rows have grown since the last tick. Output
lands in the same trip_trajectories table, overwriting each trip
instance's prior rows atomically (delete-then-insert per (trip_id, start_date), enforced by a schema-level unique index). Manual
make analytics-run and the worker can coexist — last writer wins per
trip instance, and no duplicates are possible.
make up # brings up the analytics-worker too
make analytics-worker-logs # tail the worker loopRead them back from a notebook (host port 5433):
import pandas as pd
from sqlalchemy import create_engine
e = create_engine("postgresql+psycopg://ttc:ttc@localhost:5433/ttc_gtfsrt")
pd.read_sql("""
SELECT datetime, travel_distance_m, moving_speed_m_s, observed
FROM trip_trajectories
WHERE route_id = '29' AND service_date = '2026-04-20'
ORDER BY trip_id, start_date, datetime
""", e)make up # build + start the full stack
make down # stop the full stack (keeps data)
make ps # service status
make logs # tail all compose logs
make migrate # alembic upgrade head (via compose)
make collect-once # run a single fetch + persist cycle
make capture-sample # record a fresh protobuf fixture
make test # pytest (host)
make fmt # ruff format + fix
make lint # ruff check
make api # run the API locally (no docker)
make dashboard # run the dashboard locally (no docker)
make analytics-run DATE=2026-04-20 [ROUTE=29] # batch analytics pipeline
make analytics-worker-logs # tail the analytics-worker loop
make db-reset # dry-run — print row counts that would be truncated
make db-reset-confirm # DESTRUCTIVE — TRUNCATE every data tableWhen the data scope changes (e.g. editing COLLECTOR_ROUTE_ALLOWLIST) or
the trajectory pipeline needs a clean slate, follow this sequence. The
data-reset is separate from schema migrations so migrations don't
implicitly destroy data.
# 1. Verify what would be wiped.
make db-reset # dry-run: prints row counts per table
# 2. Actually wipe. Leaves schema + alembic_version intact.
make db-reset-confirm
# 3. Apply any new migrations (e.g. 0004 adds a unique index on
# trip_trajectories; its guard refuses to run if duplicates exist,
# so db-reset first).
make migrate
# 4. Restart services so the collector picks up the new allowlist and
# the analytics worker starts its loop against the fresh data.
docker compose restart collector analytics-worker
# 5. Watch the streams.
make logs # everything
make analytics-worker-logs # just the trajectory refresh loop