GTFS Validation Rules and Common Schema Errors

Public transit data pipelines fail silently when schema violations slip through ingestion. A misaligned coordinate, an orphaned trip_id, or a time-format edge case will not throw an exception — it will produce plausible-looking output that quietly corrupts routing decisions, fare calculations, and passenger information downstream. For mobility platform teams, transit analysts, and Python GIS engineers working within the broader GTFS Feed Architecture & Fundamentals ecosystem, catching these anomalies before they reach production is non-negotiable.

This guide moves from theoretical compliance to executable validation workflows: layered schema checks, referential integrity enforcement, domain-specific business rules, and Python implementation patterns that scale across enterprise data environments.

Prerequisites

Before building the validation pipeline, confirm your environment meets these baseline requirements:

• GTFS files required: agency.txt, stops.txt, routes.txt, trips.txt, stop_times.txt, and either calendar.txt or calendar_dates.txt (or both)
• Python 3.9+ — install dependencies with pip install pandas>=2.0 pydantic>=2.0
• zipfile is part of the standard library; no additional install needed
• Assumed feed conditions: raw ZIP archive from the publisher, UTF-8 or UTF-8-BOM encoding, .txt files at the archive root (not in subdirectories)
• Background: familiarity with primary/foreign key relationships across GTFS files — covered in Understanding GTFS Static Feed Structure

Validation is not a single script; it is a layered process. Schema validation catches structural violations, referential checks enforce relational integrity, and business rule validation ensures operational plausibility. Skipping any layer introduces silent data degradation that compounds downstream.

Spec Background: What GTFS Validation Actually Checks

The GTFS Schedule Reference defines three categories of constraint that a compliant feed must satisfy.

Presence constraints determine which files and fields are mandatory versus optional. A feed without stops.txt is unconditionally invalid. A feed without shapes.txt is technically valid, though routing quality degrades. Fields like stop_name are required within stops.txt, while stop_desc is optional.

Type and enumeration constraints specify the format of each field value. stop_lat and stop_lon must be valid WGS84 decimal-degree floats. location_type must be an integer from 0 to 4. wheelchair_boarding accepts only 0, 1, or 2. Silent type coercion by pandas — for example, reading stop_lat as a string due to a header encoding issue — bypasses these constraints entirely if not explicitly checked.

Relational constraints define the foreign key graph. The table below maps the key relationships that most commonly produce validation failures:

Validation Pipeline Architecture

A robust GTFS validation pipeline follows a deterministic sequence. Deviating from this order produces false positives, masks root causes, or triggers cascading failures during DataFrame joins.

Step 1 — Archive Extraction and File Discovery

Unpack the ZIP archive and verify the presence of core files. Missing mandatory files should trigger immediate pipeline failure — continuing with an incomplete feed produces misleading partial results.

import zipfile
import io
from pathlib import Path

MANDATORY_FILES = {
    "agency.txt", "stops.txt", "routes.txt",
    "trips.txt", "stop_times.txt",
}
CALENDAR_FILES = {"calendar.txt", "calendar_dates.txt"}

def extract_and_discover(feed_path: str | Path) -> dict[str, io.BytesIO]:
    """
    Extract a GTFS ZIP and return a mapping of filename → BytesIO.
    Raises ValueError for missing mandatory files or nested directory layout.
    """
    feed_path = Path(feed_path)
    buffers: dict[str, io.BytesIO] = {}

    with zipfile.ZipFile(feed_path, "r") as zf:
        names = {info.filename for info in zf.infolist() if not info.is_dir()}
        # Reject nested-directory layouts — many parsers silently skip these
        nested = [n for n in names if "/" in n]
        if nested:
            raise ValueError(f"Archive contains nested paths (not GTFS root layout): {nested[:5]}")

        for name in names:
            if name.endswith(".txt"):
                buffers[name] = io.BytesIO(zf.read(name))

    missing_mandatory = MANDATORY_FILES - buffers.keys()
    if missing_mandatory:
        raise ValueError(f"Missing mandatory GTFS files: {missing_mandatory}")

    if not (CALENDAR_FILES & buffers.keys()):
        raise ValueError("Feed must contain calendar.txt, calendar_dates.txt, or both.")

    return buffers

Handle encoding edge cases before passing buffers downstream. Some agencies publish UTF-8-BOM files; pandas silently handles the BOM when you pass encoding="utf-8-sig" to read_csv.

Step 2 — Schema and Type Validation

Load each .txt file with explicit dtype declarations. This stage aligns with the structural expectations detailed in the stops.txt and stop_times.txt relationships guide, where string-typed stop_id prevents leading-zero truncation on numeric-looking identifiers.

import pandas as pd
from pydantic import BaseModel, field_validator
from typing import Optional

# --- Explicit dtype maps prevent silent coercion ---
STOPS_DTYPES: dict[str, str] = {
    "stop_id": "str",
    "stop_name": "str",
    "stop_lat": "float64",
    "stop_lon": "float64",
    "location_type": "Int8",       # nullable integer
    "wheelchair_boarding": "Int8",
    "parent_station": "str",
}

STOP_TIMES_DTYPES: dict[str, str] = {
    "trip_id": "str",
    "stop_id": "str",
    "stop_sequence": "int32",
    "arrival_time": "str",    # HH:MM:SS — may exceed 23:59:59
    "departure_time": "str",
    "pickup_type": "Int8",
    "drop_off_type": "Int8",
}

TRIPS_DTYPES: dict[str, str] = {
    "trip_id": "str",
    "route_id": "str",
    "service_id": "str",
    "shape_id": "str",
    "direction_id": "Int8",
    "wheelchair_accessible": "Int8",
}

class StopSchema(BaseModel):
    stop_id: str
    stop_name: str
    stop_lat: float
    stop_lon: float
    location_type: Optional[int] = 0

    @field_validator("stop_lat")
    @classmethod
    def validate_lat(cls, v: float) -> float:
        if not (-90.0 <= v <= 90.0):
            raise ValueError(f"stop_lat out of WGS84 range: {v}")
        return v

    @field_validator("stop_lon")
    @classmethod
    def validate_lon(cls, v: float) -> float:
        if not (-180.0 <= v <= 180.0):
            raise ValueError(f"stop_lon out of WGS84 range: {v}")
        return v

    @field_validator("location_type")
    @classmethod
    def validate_location_type(cls, v: Optional[int]) -> Optional[int]:
        if v is not None and v not in range(5):
            raise ValueError(f"location_type must be 0-4, got {v}")
        return v

def load_stops(buf: io.BytesIO) -> tuple[pd.DataFrame, list[str]]:
    df = pd.read_csv(
        buf,
        dtype=STOPS_DTYPES,
        encoding="utf-8-sig",
        keep_default_na=False,
    )
    required = {"stop_id", "stop_name", "stop_lat", "stop_lon"}
    missing_cols = required - set(df.columns)
    if missing_cols:
        raise ValueError(f"stops.txt missing required columns: {missing_cols}")

    errors: list[str] = []
    for idx, row in df.iterrows():
        try:
            StopSchema(**{k: row[k] for k in StopSchema.model_fields if k in row})
        except Exception as exc:
            errors.append(f"stops.txt row {idx}: {exc}")

    return df, errors

For large feeds, replace row-by-row pydantic validation with vectorized range checks. The pydantic approach is ideal for small feeds and CI pipelines where readable error messages matter; vectorized checks are necessary when stops.txt exceeds 100,000 rows.

def vectorized_coordinate_check(df: pd.DataFrame) -> pd.DataFrame:
    """Return rows where stop_lat or stop_lon falls outside WGS84 bounds."""
    lat_ok = df["stop_lat"].between(-90.0, 90.0)
    lon_ok = df["stop_lon"].between(-180.0, 180.0)
    return df[~(lat_ok & lon_ok)].copy()

Step 3 — Referential Integrity Checks

Cross-reference foreign keys across files. This is where most enterprise pipelines encounter cascading failures. Use pandas.merge with indicator=True to surface orphaned records without loading full DataFrames into memory twice.

import logging

logger = logging.getLogger(__name__)

def check_referential_integrity(
    stops: pd.DataFrame,
    stop_times: pd.DataFrame,
    trips: pd.DataFrame,
    routes: pd.DataFrame,
    calendar_service_ids: set[str],
) -> list[dict]:
    """
    Run all cross-file FK checks. Returns a list of violation dicts with
    keys: rule_id, file, column, orphan_count, sample_values.
    """
    violations: list[dict] = []

    def fk_check(
        child: pd.DataFrame,
        child_file: str,
        fk_col: str,
        parent: pd.DataFrame,
        pk_col: str,
        rule_id: str,
    ) -> None:
        child_ids = child[[fk_col]].drop_duplicates().dropna()
        merged = child_ids.merge(
            parent[[pk_col]].drop_duplicates(),
            left_on=fk_col,
            right_on=pk_col,
            how="left",
            indicator=True,
        )
        orphans = merged[merged["_merge"] == "left_only"][fk_col]
        if not orphans.empty:
            violations.append({
                "rule_id": rule_id,
                "file": child_file,
                "column": fk_col,
                "orphan_count": len(orphans),
                "sample_values": orphans.head(5).tolist(),
            })
            logger.warning(
                "%s: %d orphaned %s values in %s (sample: %s)",
                rule_id, len(orphans), fk_col, child_file, orphans.head(3).tolist()
            )

    fk_check(stop_times, "stop_times.txt", "stop_id",   stops,  "stop_id",   "E001")
    fk_check(stop_times, "stop_times.txt", "trip_id",   trips,  "trip_id",   "E002")
    fk_check(trips,      "trips.txt",      "route_id",  routes, "route_id",  "E003")

    # service_id must appear in calendar.txt or calendar_dates.txt (or both)
    trip_service_ids = set(trips["service_id"].dropna().unique())
    orphaned_services = trip_service_ids - calendar_service_ids
    if orphaned_services:
        violations.append({
            "rule_id": "E004",
            "file": "trips.txt",
            "column": "service_id",
            "orphan_count": len(orphaned_services),
            "sample_values": list(orphaned_services)[:5],
        })

    return violations

The indicator=True flag avoids a full outer join and keeps memory usage predictable — the merge only loads the two foreign key columns, not entire DataFrames.

Step 4 — Business Rule and Plausibility Validation

Enforce domain-specific constraints that the spec mandates but CSV parsers cannot detect automatically. These checks require understanding how GTFS represents time — covered in detail in Timezone Handling and Schedule Normalization.

import re

_TIME_PATTERN = re.compile(r"^(\d{1,2}):([0-5]\d):([0-5]\d)$")

def parse_gtfs_time(time_str: str) -> int:
    """
    Convert a GTFS HH:MM:SS string (including values > 24:00:00) to
    total seconds since service-day midnight. Returns -1 for invalid strings.
    """
    m = _TIME_PATTERN.match(str(time_str).strip())
    if not m:
        return -1
    h, mn, s = int(m.group(1)), int(m.group(2)), int(m.group(3))
    return h * 3600 + mn * 60 + s

def check_stop_time_sequence(stop_times: pd.DataFrame) -> list[dict]:
    """
    Verify that within each trip:
      1. stop_sequence values are strictly increasing
      2. departure_time >= arrival_time at every stop
      3. arrival_time[n] <= departure_time[n-1] is NOT required by spec
         but a regression of >1 hour is flagged as a plausibility warning
    """
    violations: list[dict] = []

    stop_times = stop_times.copy()
    stop_times["_arr_sec"] = stop_times["arrival_time"].map(parse_gtfs_time)
    stop_times["_dep_sec"] = stop_times["departure_time"].map(parse_gtfs_time)

    invalid_times = stop_times[
        (stop_times["_arr_sec"] == -1) | (stop_times["_dep_sec"] == -1)
    ]
    if not invalid_times.empty:
        violations.append({
            "rule_id": "E010",
            "description": "Unparseable time string in stop_times.txt",
            "row_count": len(invalid_times),
            "sample_trip_ids": invalid_times["trip_id"].head(5).tolist(),
        })

    dwell_violations = stop_times[stop_times["_dep_sec"] < stop_times["_arr_sec"]]
    if not dwell_violations.empty:
        violations.append({
            "rule_id": "E011",
            "description": "departure_time < arrival_time at same stop",
            "row_count": len(dwell_violations),
            "sample_trip_ids": dwell_violations["trip_id"].head(5).tolist(),
        })

    # Check strictly-increasing stop_sequence per trip
    sorted_st = stop_times.sort_values(["trip_id", "stop_sequence"])
    seq_diff = sorted_st.groupby("trip_id")["stop_sequence"].diff()
    non_monotone = sorted_st[seq_diff.notna() & (seq_diff <= 0)]
    if not non_monotone.empty:
        violations.append({
            "rule_id": "E012",
            "description": "stop_sequence not strictly increasing within trip",
            "row_count": len(non_monotone),
            "sample_trip_ids": non_monotone["trip_id"].head(5).tolist(),
        })

    return violations

Step 5 — Error Aggregation and Reporting

Compile all violations into a structured report categorised by severity. Standardised reporting enables automated alerting, publisher feedback loops, and CI/CD gating.

import json
from datetime import datetime, timezone

SEVERITY_MAP = {
    "E001": "error", "E002": "error", "E003": "error", "E004": "error",
    "E010": "error", "E011": "error", "E012": "warning",
}

def build_validation_report(
    violations: list[dict],
    feed_path: str,
) -> dict:
    report = {
        "feed": str(feed_path),
        "validated_at": datetime.now(timezone.utc).isoformat(),
        "summary": {"error": 0, "warning": 0, "info": 0},
        "violations": [],
    }
    for v in violations:
        severity = SEVERITY_MAP.get(v.get("rule_id", ""), "info")
        report["summary"][severity] += 1
        report["violations"].append({**v, "severity": severity})

    return report

# Usage:
# report = build_validation_report(all_violations, "nyc_subway_20240601.zip")
# print(json.dumps(report, indent=2))

Validation and Verification

After running the pipeline, confirm correctness with these assertions before the report leaves the validation stage:

def assert_validation_report_shape(report: dict, stops_df: pd.DataFrame) -> None:
    """Raise AssertionError if the report structure is internally inconsistent."""
    assert "violations" in report, "Report missing violations list"
    assert "summary" in report, "Report missing summary"

    counted = sum(report["summary"].values())
    assert counted == len(report["violations"]), (
        f"Summary count ({counted}) != violation list length ({len(report['violations'])})"
    )

    # All stops that appear in stop_times must now exist in stops (post-validation)
    # This assertion only holds after FK fixes have been applied
    # It serves as a regression guard in CI
    assert stops_df["stop_id"].duplicated().sum() == 0, (
        "stops.txt still contains duplicate stop_id values after deduplication step"
    )

Common Schema Errors and Mitigation Strategies

Real-world GTFS feeds rarely conform perfectly to the specification. These are the most frequent violations encountered in production environments.

Duplicate primary keys. Multiple rows in stops.txt or routes.txt share the same stop_id or route_id. The impact is ambiguous joins and unpredictable routing behavior. Fix by deduplicating with a deterministic rule — for example, keep the row with the highest stop_sequence value or most-complete field population — and enforce unique constraints at the database layer post-validation.

Invalid coordinate ranges. stop_lat or stop_lon falls outside [-90, 90] or [-180, 180], or the axes are swapped (a surprisingly common publisher error). Flag coordinates near (0.0, 0.0) separately: these are almost always null-value placeholders that slipped past the publisher’s own checks. See Coordinate Reference Systems for Transit Data for projection-aware remediation when agencies supply non-WGS84 coordinates.

Time format violations and day rollover. stop_times.txt uses values like 24:05:00 or 25:30:00 for overnight service. Standard datetime.strptime will raise a ValueError on these inputs. Parse times as strings first, then convert to pandas.Timedelta or integer seconds. The parse_gtfs_time function above handles the full hour range.

Missing calendar coverage. A service_id appears in trips.txt but has no corresponding record in either calendar.txt or calendar_dates.txt. Routing engines that encounter an uncovered service ID will either skip the trip silently or raise a runtime exception depending on the implementation. Validate that every service_id has at least one active date. For feeds that rely entirely on calendar_dates.txt (a valid pattern for agencies with complex irregular schedules), ensure exception_type=1 records exist — not just exception_type=2 removals.

Duplicate stop_sequence values within a trip. The spec requires strictly increasing stop_sequence per trip_id, but it does not require values to be consecutive. Agencies frequently reset sequences (e.g., 1, 2, 2, 3) when editing trips by hand. Detect with a groupby().diff() approach as shown in Step 4 above.

agency_id omitted from single-agency feeds. The spec makes agency_id optional in routes.txt when only one agency is present. However, downstream systems that validate against a strict schema will reject feeds where agency_id is absent. Flag this as a warning rather than an error, and document the publisher behavior for the feed registry.

Failure Modes and Edge Cases

• UTF-8-BOM headers: Some Windows-based publisher tools write a byte-order mark at the start of the first column header. pandas.read_csv with encoding="utf-8" will include the BOM in the column name (e.g., agency_id), causing all FK joins on that column to fail silently. Use encoding="utf-8-sig" consistently.
• Carriage-return line endings: GTFS files from Windows publishing pipelines often use \r\n. Pass lineterminator="\n" to read_csv only if you observe spurious trailing \r characters in string columns.
• Mixed stop_id types: An agency may use purely numeric stop IDs in one file ("1001") and zero-padded strings in another ("01001"). Enforce dtype=str for all ID columns and do not rely on pandas auto-inference.
• Empty optional files: Some publishers include shapes.txt as an empty file (header only, no data rows). Validate row counts before attempting joins — a zero-row shapes.txt is valid but should produce an info-level notice that geometry will be absent.
• calendar_dates.txt-only feeds: Agencies with highly irregular service (demand-responsive routes, seasonal operations) may omit calendar.txt entirely and express all active dates in calendar_dates.txt. Your pipeline must not require calendar.txt to be non-empty.
• Feeds with feed_info.txt version strings: The agency metadata and feed versioning practices guide covers how feed_version in feed_info.txt should be used to gate re-processing — validation pipelines should read this field before running expensive integrity checks on unchanged feeds.

Performance and Scale Notes

For feeds larger than 500 MB — common for multi-agency aggregators or dense metro networks — the row-by-row pydantic validation approach becomes impractical.

Chunked reading. Use pd.read_csv(..., chunksize=50_000) for stop_times.txt, which is typically the largest file. Accumulate FK sets across chunks and run the merge check only after all chunks are processed.

Vectorised checks over iteration. Replace pydantic row validators with vectorised pandas operations for coordinate and time checks. The difference between iterrows-based and vectorised validation on a 2 million-row stop_times.txt is typically 45 seconds versus under 2 seconds.

Parquet intermediate storage. After initial ingestion and type enforcement, write DataFrames to Parquet with df.to_parquet("stops.parquet", compression="zstd"). Subsequent validation runs can skip CSV parsing and load directly from Parquet, reducing cold-start time by 60-80% for repeated checks. For comprehensive memory management strategies, see Memory-Efficient Processing for Large Feeds.

Multi-agency batching. When validating dozens of feeds in parallel, use concurrent.futures.ProcessPoolExecutor with each feed in its own process. Validation is CPU-bound; thread-based parallelism offers no benefit due to the GIL.

Incremental validation. For feeds that update daily, hash the raw ZIP (hashlib.sha256) and compare against a stored digest. Skip full validation if the hash is unchanged. For changed feeds, diff primary key sets between the previous and current version to isolate regressions rather than re-validating the entire archive.

Related

• How to Validate a GTFS Feed with Python — complete implementation with the MobilityData Java validator and Python fallback
• Mastering stops.txt and stop_times.txt Relationships — deep dive into the join logic and sequence integrity that validation must protect
• Timezone Handling and Schedule Normalization — context for why time parsing in stop_times.txt requires special treatment
• Memory-Efficient Processing for Large Feeds — chunked reading and Parquet strategies for feeds that exceed available RAM
• Agency Metadata and Feed Versioning Practices — how feed_info.txt version strings prevent redundant re-validation

Up: GTFS Feed Architecture & Fundamentals | Home