Time-Dependent Speed Profiles in Python

Most routing graphs store a single static speed per edge, which means the same residential street reports free-flow speed at 3 a.m. and gridlock at 8 a.m. equally wrong. This page shows how to attach a time-dependent speed profile to each edge so that a query at a given timestamp resolves to the correct interval-specific speed, and it slots into the broader work of custom cost functions for routing solvers as the piece that makes custom_weight a function of departure time rather than a fixed scalar. It assumes you already have a graph with length and a base travel_time attribute per edge, built and validated as described in the Python Routing Engines & Isochrone Mapping overview.

The core engineering problem is evaluation speed, not data modeling. A naive if 7 <= hour < 9: speed = ... chain scales linearly with the number of buckets and has to be re-run per edge per query — unacceptable for a solver that evaluates millions of edge weights per route search. The fix is the same one search engines use for range queries: sort the bucket boundaries once, then resolve any query timestamp to its bucket with a binary search in O(log n) instead of O(n).

When to Use This Approach

Time-dependent speed profiles are worth the added complexity when:

• Historical GPS or loop-detector data shows a repeatable diurnal speed pattern on a meaningful subset of edges — typically arterials and highway segments near urban cores
• Routes are computed for a known or estimated departure time, not just “now” — batch ETA jobs, delivery-window planning, and isochrone generation at multiple times of day all need this
• A single static travel_time attribute produces measurably wrong ETAs during peak hours (validate this before building the profile pipeline — see the Configuration Reference below for calibration inputs)
• The routing backend supports time-dependent weight evaluation, either natively (OSRM traffic updates, Valhalla’s predictive traffic) or through a custom NetworkX weight callable

Skip this approach for low-variance rural networks where speed rarely deviates from free-flow, and for graphs where departure time is unknown at query time — a live GPS-navigation product answering “fastest route right now” gets more value from real-time traffic feeds than from a historical diurnal curve. Time-dependent profiles are best treated as a baseline layer beneath live traffic, not a replacement for it.

Implementation

The pattern below stores each profile as two parallel sorted numpy arrays — bucket_starts (seconds since midnight) and speeds_kph — and resolves any query timestamp with np.searchsorted, which is numpy’s vectorized binary search. This works for a single lookup or for a batch of thousands of (profile_id, query_time) pairs in one call, which is what a route-search inner loop actually needs.

# requires: numpy (pip install numpy)
# Python 3.9+

from __future__ import annotations
import numpy as np
from dataclasses import dataclass

SECONDS_PER_DAY = 86_400


@dataclass
class SpeedProfile:
    profile_id: int
    bucket_starts: np.ndarray   # int32, sorted, seconds since midnight
    speeds_kph: np.ndarray      # float32, same length as bucket_starts


def build_profile(profile_id: int, bucket_minutes: int, hourly_speeds: dict[float, float]) -> SpeedProfile:
    """
    Build a sorted lookup profile from {hour_of_day: speed_kph} samples.
    hour_of_day may be fractional (e.g. 7.25 for 07:15). Buckets are
    resampled onto a fixed grid of `bucket_minutes` width so every
    profile in a store shares comparable granularity.
    """
    n_buckets = SECONDS_PER_DAY // (bucket_minutes * 60)
    grid_hours = np.arange(n_buckets) * (bucket_minutes / 60.0)

    sample_hours = np.array(sorted(hourly_speeds), dtype=np.float64)
    sample_speeds = np.array([hourly_speeds[h] for h in sorted(hourly_speeds)], dtype=np.float64)

    # Wrap samples so interpolation is continuous across midnight
    wrapped_hours = np.concatenate([sample_hours - 24.0, sample_hours, sample_hours + 24.0])
    wrapped_speeds = np.tile(sample_speeds, 3)

    resampled = np.interp(grid_hours, wrapped_hours, wrapped_speeds)

    bucket_starts = (grid_hours * 3600).astype(np.int32)
    return SpeedProfile(profile_id, bucket_starts, resampled.astype(np.float32))


def lookup_speed(profile: SpeedProfile, query_seconds: np.ndarray) -> np.ndarray:
    """
    Vectorized binary-search lookup. Accepts a scalar or array of
    query timestamps (seconds since midnight, may exceed 86400 for
    multi-day batches) and returns the matching speeds_kph values.
    """
    query_seconds = np.mod(np.asarray(query_seconds), SECONDS_PER_DAY)
    idx = np.searchsorted(profile.bucket_starts, query_seconds, side="right") - 1
    idx = np.clip(idx, 0, len(profile.bucket_starts) - 1)
    return profile.speeds_kph[idx]

A production graph has many distinct profiles — one per road-class-and-region cluster, not one per edge. Group edges by profile_id, resolve one query time per group, and broadcast the result back with numpy fancy indexing rather than looping in Python:

# requires: numpy (pip install numpy)
# Python 3.9+

def batch_edge_speeds(
    profile_store: dict[int, SpeedProfile],
    edge_profile_ids: np.ndarray,   # int32, shape (n_edges,)
    query_seconds: int,             # single departure time for this route search
) -> np.ndarray:
    """Resolve speeds for every edge in one route search at a fixed departure time."""
    out = np.empty(edge_profile_ids.shape, dtype=np.float32)
    for pid in np.unique(edge_profile_ids):
        profile = profile_store[pid]
        mask = edge_profile_ids == pid
        out[mask] = lookup_speed(profile, np.full(mask.sum(), query_seconds))
    return out

The np.unique grouping keeps the per-profile searchsorted call vectorized across every edge that shares a profile, so a graph with a few hundred distinct profiles and a few million edges still resolves in a handful of array operations instead of a million-iteration Python loop.

Key Parameters and Tuning

Calibrate bucket_minutes against the density of your source data — GPS traces or loop-detector counts per bucket per road class. If a bucket has fewer than roughly 30 samples, widen it or fall back to the coarser regional average for that hour.

Integration Points

NetworkX. Wrap lookup_speed in a weight callable that NetworkX calls per edge during traversal: weight=lambda u, v, d: d["length"] / (lookup_speed(profile_store[d["profile_id"]], np.array([query_seconds]))[0] / 3.6). This is a direct extension of the custom_weight pattern described in custom cost functions for routing solvers — the only change is that the weight now closes over a fixed query_seconds chosen per route search rather than a static coefficient.

OSRM. OSRM does not accept a Python callable at query time; time-dependent speeds have to be baked in during osrm-contract as a traffic-update CSV (node_id_1,node_id_2,speed_kph per time slice) or encoded directly in the Lua profile’s speed_handler. Precompute one CSV per bucket from the same SpeedProfile arrays used above, and re-run osrm-contract --segment-speed-file per bucket if you need genuinely time-sliced routing rather than a single averaged profile. The integrating custom traffic weights into OSRM guide covers the CSV format and the Lua callback interface in detail.

Profile assignment. Assigning a profile_id to each edge is a separate preprocessing step — cluster edges by highway tag, region, and observed diurnal shape (k-means on normalized 24-bucket vectors works well), then write the resulting profile_id as an edge attribute alongside length and base travel_time.

Validation Checklist

1. Monotonic bucket_starts. Assert np.all(np.diff(profile.bucket_starts) > 0) for every profile — a non-monotonic array silently breaks searchsorted’s correctness guarantee.
2. Full 24h coverage. Confirm profile.bucket_starts[0] == 0 and that the array covers the full day with no gaps once wrapped; a missing final bucket leaves the last minutes of the day unresolved.
3. Midnight wraparound. Query t = 0 and t = 86399 and confirm both resolve to sensible late-night/early-morning buckets, not an index error or a duplicate boundary bucket.
4. Speed floor enforcement. Assert (profile.speeds_kph >= min_speed_floor_kph).all() after resampling; a near-zero bucket will produce an unbounded travel_time when divided into length.
5. Batch vs scalar equivalence. Compare lookup_speed(profile, np.array([t]))[0] against a manual per-record lookup for a random sample of timestamps to confirm the vectorized path and the reference path agree exactly.
6. ETA sanity check. Route a known corridor at 04:00, 08:00, and 17:00 departure times and confirm the resulting travel_time ordering matches known peak-hour congestion for that corridor — an inverted ordering usually means buckets were built against the wrong day type.

Related

• Custom cost functions for routing solvers — the broader weighting framework this profile lookup plugs into as a time-varying custom_weight component
• Integrating custom traffic weights into OSRM — encoding these profiles as OSRM traffic-update CSVs and Lua speed callbacks
• Python Routing Engines & Isochrone Mapping — the overview covering engine selection and graph construction this technique builds on
• NetworkX Dijkstra vs A* for route calculation — algorithm choice implications when edge weights vary by departure time rather than staying static