Mapping Congestion Thresholds to Real-Time Traffic Windows

Congestion detection fails when uniform time bins are imposed on uneven probe streams: sparse observation windows produce false positives while high-density bursts mask brief bottlenecks. The fix is to combine density-aware rolling windows with a calibrated, road-class-specific threshold function, then require a sustained crossing before flagging a link as congested. This page shows a complete, production-ready implementation of that pipeline as part of dynamic time-binning strategies within the broader temporal aggregation and window mapping domain.

Why this happens

Fixed-interval resampling — the default resample("5min") call most engineers reach for first — distributes all observations evenly across uniform buckets regardless of how many probes actually reported in each window. This violates the implicit assumption that every bin is equally reliable. The failure mode is a direct consequence of using static bins on GPS precision data whose density varies by road class, time of day, and fleet instrumentation rate.

On a low-coverage urban arterial at 2 am, a single misreported speed value can drag a bin’s average below the threshold and fire a congestion alert. Conversely, on a heavily probed freeway during peak hour, short-lived bottlenecks (lane drops, merge conflicts) can be averaged out of existence inside a too-wide bin. Both failure modes share the same root cause: the bin width and the observation density are not co-calibrated.

The correct architecture couples the window stride to the probe arrival rate and enforces a minimum observation count per window before assigning any congestion state.

Core mitigation pipeline

1. Compute a free-flow baseline per link: derive 85th-percentile historical speeds stratified by road class, day-of-week, and hour.
2. Derive a dynamic threshold: multiply the baseline by (1 − α), where α is a tolerance factor specific to road class.
3. Apply density-gated rolling windows: resample to the target frequency, discard bins with fewer than min_obs probes, and smooth with exponential weighting to surface sustained slowdowns.
4. Enforce the sustained-crossing rule: flag congestion only when the smoothed speed stays below the threshold for N consecutive window steps, preventing transient signal-cycle dips from triggering state changes.

How the pipeline stages connect

The diagram below shows how raw probe records flow through baseline lookup, density gating, and the sustained-crossing check to produce a per-link congestion state.

Production-ready Python implementation

The function below is fully typed and handles empty DataFrame input, missing columns, sparse bins, and forward-filled threshold alignment. Speed computations are CRS-independent here — the speed_kmh column is assumed pre-computed in a metric projected CRS (e.g., EPSG:32632) upstream; never derive it from raw WGS84 degree differences.

import pandas as pd
import numpy as np
from typing import Optional


def map_congestion_to_windows(
    df: pd.DataFrame,
    link_col: str = "link_id",
    ts_col: str = "timestamp",
    speed_col: str = "speed_kmh",
    free_flow_col: str = "v_ff_kmh",
    tolerance: float = 0.30,
    min_obs_per_window: int = 3,
    sustained_steps: int = 2,
    window_minutes: int = 5,
) -> pd.DataFrame:
    """
    Map real-time probe speed observations to per-link congestion states.

    Parameters
    ----------
    df : DataFrame with columns [link_col, ts_col, speed_col, free_flow_col].
         speed_kmh must be derived from a metric projected CRS — NOT raw WGS84.
    tolerance : α factor; congestion threshold = v_ff × (1 − tolerance).
    min_obs_per_window : bins with fewer observations are marked LOW_CONFIDENCE.
    sustained_steps : consecutive below-threshold windows required to flag CONGESTED.
    window_minutes : resampling frequency in minutes.

    Returns
    -------
    DataFrame with columns [link_id, window_ts, avg_speed, obs_count, congestion_state].
    """
    required = {link_col, ts_col, speed_col, free_flow_col}
    missing = required - set(df.columns)
    if missing:
        raise ValueError(f"Input DataFrame missing columns: {missing}")

    if df.empty:
        return pd.DataFrame(
            columns=["link_id", "window_ts", "avg_speed", "obs_count", "congestion_state"]
        )

    df = df.copy()
    df[ts_col] = pd.to_datetime(df[ts_col])
    df = df.sort_values([link_col, ts_col])

    # Precompute per-observation dynamic threshold
    df["v_thresh"] = df[free_flow_col] * (1.0 - tolerance)

    # Cap implausible speeds at 150 % of free-flow to remove GPS multipath artefacts
    df[speed_col] = df[speed_col].clip(upper=df[free_flow_col] * 1.5)

    results: list[pd.DataFrame] = []
    freq = f"{window_minutes}min"

    for link_id, group in df.groupby(link_col, sort=False):
        group = group.set_index(ts_col)

        # Rolling mean over resampled windows
        rolling_speed = (
            group[speed_col]
            .resample(freq)
            .mean()
            .rolling(window=sustained_steps, min_periods=1)
            .mean()
        )

        # Observation count per bin for density gating
        obs_count = group[speed_col].resample(freq).count()

        # Align threshold to the resampled index via forward-fill
        thresh_aligned = (
            group["v_thresh"]
            .resample(freq)
            .mean()
            .reindex(rolling_speed.index)
            .ffill()
        )

        # Sustained crossing: N consecutive steps below threshold
        below_thresh = (rolling_speed <= thresh_aligned).astype(int)
        sustained = (
            below_thresh
            .rolling(window=sustained_steps, min_periods=sustained_steps)
            .sum()
            >= sustained_steps
        )

        meets_density = obs_count >= min_obs_per_window

        congestion_state = np.where(
            ~meets_density,
            "LOW_CONFIDENCE",
            np.where(sustained & meets_density, "CONGESTED", "FREE_FLOW"),
        )

        results.append(
            pd.DataFrame(
                {
                    "link_id": link_id,
                    "window_ts": rolling_speed.index,
                    "avg_speed": rolling_speed.values,
                    "obs_count": obs_count.values,
                    "congestion_state": congestion_state,
                }
            )
        )

    return pd.concat(results, ignore_index=True) if results else pd.DataFrame(
        columns=["link_id", "window_ts", "avg_speed", "obs_count", "congestion_state"]
    )

Validation block

After running map_congestion_to_windows, verify the output before feeding it to downstream routing or pricing systems:

result = map_congestion_to_windows(df)

# Shape sanity: one row per (link × window step)
assert not result.empty, "Output is empty — check input df and groupby keys"
assert set(result["congestion_state"].unique()).issubset(
    {"CONGESTED", "FREE_FLOW", "LOW_CONFIDENCE"}
), "Unexpected congestion states in output"

# Density gate is firing: LOW_CONFIDENCE should not dominate
low_conf_rate = (result["congestion_state"] == "LOW_CONFIDENCE").mean()
if low_conf_rate > 0.30:
    import warnings
    warnings.warn(
        f"LOW_CONFIDENCE rate {low_conf_rate:.1%} is high — "
        "consider reducing min_obs_per_window or increasing probe coverage.",
        UserWarning,
    )

# Speed range sanity
assert result["avg_speed"].dropna().ge(0).all(), "Negative speeds present"
assert result["obs_count"].ge(0).all(), "Negative observation counts"

Also cross-reference a random sample of CONGESTED windows against loop detector volumes or incident management system logs. Target precision > 0.85 and recall > 0.80 in operational back-tests before promoting to production routing.

Common mistakes and gotchas

• Computing speed from raw WGS84 degrees. Haversine-derived speeds for short inter-probe distances accumulate floating-point error that can shift bins by ±2–4 km/h. Always compute speed_kmh upstream from a metric projected CRS — see coordinate reference system mapping for correct projection workflows.
• Reusing a single global tolerance α across all road classes. Freeways tolerate a 35–45 % speed drop before congestion is genuine; collectors become operationally disrupted at 15–25 %. A single value produces systematic false negatives on arterials and false positives on freeways. Stratify by highway tag or GIS road-class field.
• Setting sustained_steps=1. This is equivalent to a raw threshold with no smoothing, which fires on every signal-cycle slowdown. Urban arterials with 90-second cycles at 5-minute windows need at least sustained_steps=2 (10 minutes of sustained deviation) to distinguish genuine congestion from stop-and-go at controlled intersections.
• Ignoring LOW_CONFIDENCE bins downstream. Passing LOW_CONFIDENCE rows into an ETA engine without filtering treats sparse bins the same as fully-observed ones. Filter or weight by obs_count in the consuming model.
• Applying rolling().mean() before the density gate. If the density gate discards a bin mid-window, the rolling mean carries NaN forward incorrectly. Compute the density gate first, set insufficient bins to NaN, then apply the rolling mean with min_periods=sustained_steps.
• Not forward-filling the threshold after resample. resample can introduce NaN rows for intervals with zero observations. Without ffill(), the <= comparison silently produces NaN booleans, which the rolling().sum() treats as zero — suppressing congestion detections in sparse corridors.

Threshold calibration by road class

Back-test each configuration against at least 30 days of historical data and at least three distinct incident types (lane closure, weather event, scheduled roadworks) before deploying.

Related

• Dynamic Time-Binning Strategies — choosing window strides and bin widths for mobility data
• Choosing Optimal Bin Sizes for Urban Mobility Heatmaps — bin-size selection for spatial aggregation
• Computing Rolling Average Speed Over Sliding Time Windows — rolling statistics for mobility metrics
• Calculating Instantaneous Speed from Discrete GPS Points — upstream speed computation from raw GPS
• Sampling Rate Optimization — managing probe density trade-offs at the ingestion stage

Back to Dynamic Time-Binning Strategies