Rolling Statistics for Mobility Metrics

Rolling statistics convert noisy, high-frequency movement traces into noise-resilient temporal signals by computing aggregations over a sliding time window that advances with each observation rather than snapping to fixed calendar buckets.

This technique sits at the core of Temporal Aggregation & Window Mapping. Where fixed-interval binning collapses trajectories into discrete slots and discards intra-period variance, a sliding window preserves temporal continuity: every output point reflects the local statistical neighbourhood of that moment in the trace. For mobility data scientists, urban analysts, and fleet engineering teams, this difference is critical — congestion onset, hard-braking events, transit headway drift, and dwell anomalies all exist in the sub-window temporal structure that fixed aggregation erases.

Prerequisites

Before implementing rolling aggregations, confirm that your pipeline meets these baseline requirements. Rolling computations amplify upstream data quality problems — a gap-riddled or unsorted trace will produce silently wrong statistics.

• pandas >= 1.5 — time-based offset strings ('5min', '1h') in .rolling() require a DatetimeIndex or equivalent; the Cython-backed engine is fast enough for tens of millions of GPS pings in a single process.
• numpy >= 1.23, geopandas >= 0.12 — required if computing spatial derivatives (ground speed from coordinates); always project to a metric CRS such as EPSG:3857 or a local UTM zone before distance calculations. Never compute speeds in raw EPSG:4326 — degree-based distances are non-uniform and introduce systematic error at non-equatorial latitudes.
• scipy >= 1.9 — optional, needed only for Savitzky-Golay post-smoothing.
• Time-sorted traces: Each entity’s records must have a monotonically increasing DatetimeIndex. Duplicated or out-of-order timestamps break window alignment and produce non-deterministic outputs. Validate before rolling with assert df.index.is_monotonic_increasing.
• Resolved sampling rate: Extreme gaps (> 5× the median interval) require explicit handling before rolling — see the gap-handling section below. The gap-filling techniques that apply here are distinct from simple resampling.
• UTC timestamps: Convert all timezones to UTC at ingestion. Daylight saving transitions create apparent 1-hour gaps or overlaps that corrupt daily and weekly rolling cycles. Store the original local offset as a separate metadata column if you need it for display.

Upstream, this workflow depends on having clean, GPS-error-corrected traces — see GPS precision & error handling for the denoising steps that should precede any rolling computation.

Failure Mode Taxonomy

Pipeline Overview

The rolling statistics pipeline has five deterministic stages. Stages must run in order — skipping temporal normalisation or gap-flagging before rolling produces incorrect output silently.

Implementation Walkthrough

Stage 1 — Ingest and temporal normalisation

import pandas as pd
import numpy as np

def load_and_normalise(raw_df: pd.DataFrame) -> pd.DataFrame:
    """
    Normalise a raw mobility DataFrame to a UTC DatetimeIndex sorted
    per entity.  Expects columns: entity_id, timestamp, lat, lon,
    speed_kmh (all others are carried through unchanged).

    Parameters
    ----------
    raw_df : pd.DataFrame
        Raw telemetry with at least the columns listed above.

    Returns
    -------
    pd.DataFrame
        UTC-indexed, per-entity sorted DataFrame ready for rolling ops.

    Raises
    ------
    ValueError  if required columns are absent.
    """
    required = {"entity_id", "timestamp", "speed_kmh"}
    missing = required - set(raw_df.columns)
    if missing:
        raise ValueError(f"Missing required columns: {missing}")
    if raw_df.empty:
        return raw_df.copy()

    df = raw_df.copy()
    df["timestamp"] = pd.to_datetime(df["timestamp"], utc=True)
    df = df.sort_values(["entity_id", "timestamp"])
    df = df.set_index("timestamp")

    assert df.index.is_monotonic_increasing, (
        "DatetimeIndex is not globally monotonic — check for "
        "cross-entity sort contamination."
    )
    return df

Note: assert here is a development-time guard. In production, replace with a logged warning or a raise that includes the offending entity IDs.

Stage 2 — Gap detection and masking

Real-world telematics rarely arrives at perfectly regular intervals. Pandas time-based rolling automatically accounts for irregular spacing by evaluating actual timestamp deltas within the window — but it cannot know that a 45-minute silence means the vehicle was in a tunnel versus that the GPS unit was powered off. Without explicit gap masking, the rolling mean at the first post-gap observation will incorporate stale pre-gap values if they still fall within the window duration.

def mask_gap_boundaries(
    df: pd.DataFrame,
    gap_threshold: pd.Timedelta = pd.Timedelta("2min"),
    columns_to_null: list[str] | None = None,
) -> pd.DataFrame:
    """
    Set speed_kmh (and any other specified columns) to NaN at rows
    immediately following a gap larger than gap_threshold, so that
    rolling windows do not silently bridge data voids.

    Parameters
    ----------
    df : pd.DataFrame
        UTC DatetimeIndex DataFrame, sorted per entity (output of
        load_and_normalise).
    gap_threshold : pd.Timedelta
        Maximum acceptable inter-observation gap.  Rows after a gap
        exceeding this are nulled.  Default 2 min.
    columns_to_null : list[str] | None
        Columns to set NaN at gap boundaries.  Defaults to
        ['speed_kmh'].

    Returns
    -------
    pd.DataFrame  with gap-boundary rows nulled.
    """
    if df.empty:
        return df.copy()

    cols = columns_to_null or ["speed_kmh"]
    out = df.copy()

    for entity_id, grp in out.groupby("entity_id"):
        deltas = grp.index.to_series().diff()
        gap_mask = deltas > gap_threshold
        out.loc[grp.index[gap_mask], cols] = np.nan

    return out

For very sparse trajectories (e.g., once-per-minute polling on long-haul freight), coordinate this step with gap-filling strategies to decide whether to interpolate or simply exclude the affected window.

Stage 3 — Per-entity rolling aggregation

Apply rolling aggregations per entity using groupby(). Each metric requires a separate .rolling() call — pandas time-based rolling does not accept the dict-of-tuples named aggregation form that groupby().agg() supports.

def compute_rolling_metrics(
    df: pd.DataFrame,
    window: str = "5min",
    min_periods: int = 3,
) -> pd.DataFrame:
    """
    Compute rolling mean speed, speed std, point count, and
    acceleration variance per entity over a sliding time window.

    All coordinates must already be in a metric CRS if spatial
    derivatives are used — this function only operates on speed_kmh
    which is assumed to have been computed in projected coordinates
    upstream.

    Parameters
    ----------
    df : pd.DataFrame
        Output of mask_gap_boundaries.  Must have a UTC DatetimeIndex
        and columns: entity_id, speed_kmh.
    window : str
        Pandas offset string, e.g. '5min', '15min', '1h'.
    min_periods : int
        Minimum observations required to produce a non-NaN output.
        Rows with fewer observations in the window return NaN.

    Returns
    -------
    pd.DataFrame  with added columns: speed_mean, speed_std,
                  point_count, acceleration_var.
    """
    if df.empty:
        return df.assign(
            speed_mean=pd.NA,
            speed_std=pd.NA,
            point_count=pd.NA,
            acceleration_var=pd.NA,
        )

    grp = df.groupby("entity_id")["speed_kmh"]

    # Each metric needs a separate call — no dict-agg on rolling
    speed_mean = (
        grp.rolling(window, min_periods=min_periods, closed="right")
           .mean()
           .reset_index(level=0, drop=True)
    )
    speed_std = (
        grp.rolling(window, min_periods=min_periods, closed="right")
           .std()
           .reset_index(level=0, drop=True)
    )
    point_count = (
        grp.rolling(window, min_periods=1, closed="right")
           .count()
           .reset_index(level=0, drop=True)
    )
    acceleration_var = (
        grp.rolling(window, min_periods=min_periods, closed="right")
           .var()
           .reset_index(level=0, drop=True)
    )

    return df.assign(
        speed_mean=speed_mean,
        speed_std=speed_std,
        point_count=point_count,
        acceleration_var=acceleration_var,
    )

The closed="right" parameter ensures the current timestamp is included in its own window and that no future observations can enter, eliminating look-ahead bias. This is the correct setting for both real-time streaming and feature engineering for predictive models.

For datasets exceeding available RAM, chunk by entity_id and write each chunk to a partitioned Parquet file:

import pyarrow as pa
import pyarrow.parquet as pq

def process_in_chunks(
    df: pd.DataFrame,
    output_path: str,
    window: str = "5min",
    min_periods: int = 3,
) -> None:
    """Stream rolling metrics to Parquet, one entity at a time."""
    writer = None
    for entity_id, chunk in df.groupby("entity_id"):
        result = compute_rolling_metrics(chunk, window, min_periods)
        table = pa.Table.from_pandas(result)
        if writer is None:
            writer = pq.ParquetWriter(output_path, table.schema)
        writer.write_table(table)
    if writer:
        writer.close()

Mathematical Grounding

The rolling mean over a time window of width W centred (causally) at time t is:

μ̂(t) = (1/n) Σ xᵢ for all i where tᵢ ∈ (t − W, t]

where n is the count of observations falling in that half-open interval. Because n varies across irregularly sampled traces, the rolling mean is an unweighted average of whatever observations happen to fall in the window — not a convolution with a fixed-length kernel. This is why min_periods matters: when n < min_periods, the estimate is too noisy to be reliable and should be suppressed with NaN.

For heading stability (circular statistics), the rolling variance of compass bearings requires the circular variance formula rather than ordinary variance, because 359° and 1° are close in heading space but numerically distant:

def circular_variance_deg(bearings: pd.Series) -> float:
    """Circular variance of a series of compass bearings in degrees.
    Returns a value in [0, 1]: 0 = perfectly uniform direction,
    1 = maximally dispersed."""
    rad = np.deg2rad(bearings.dropna())
    if len(rad) < 2:
        return np.nan
    R = np.abs(np.mean(np.exp(1j * rad)))   # mean resultant length
    return float(1.0 - R)

Apply this inside a rolling apply for heading stability detection — a high circular variance over a 30 s window flags potential GPS multipath errors or genuine route deviation.

Exponentially weighted alternatives (df.ewm(span=...)) assign decaying weight to older observations. They are appropriate for real-time smoothing where stale data should matter less, but they have no finite memory boundary, which makes them unsuitable for features that require strict look-back constraints.

Calibration and Parameter Tuning

Window size determines the time-scale of the signal you can detect. Wider windows suppress noise but delay event detection; narrower windows are sensitive but noisy. The table below covers common transport mode and analytical use-case combinations.

When fixed-width windows misalign with operational rhythms — shift changes, peak-hour surges, transit headway adjustments — pair rolling computations with dynamic time-binning strategies to adapt window boundaries to event density rather than rigid clock ticks. The complementary technique of aligning seasonal travel patterns is needed before you compare rolling features across weeks or months with different demand baselines.

Integration and Compatibility

Rolling statistics are a feature engineering primitive: their output columns feed directly into the following downstream stages.

Stay-point detection: A rolling count of observations with speed_kmh < 2.0 over a 10min window is a strong prior for stay-point detection algorithms. An elevated count flags candidate dwell locations before the spatial clustering step.

Speed and acceleration profiling: speed_mean and acceleration_var from this pipeline are the primary inputs to speed and acceleration profiling for transport mode classification and anomaly scoring.

Directionality and turn analysis: Rolling circular variance on bearings feeds directly into directionality and turn analysis. Persistent high variance signals that a vehicle is navigating a complex interchange or experiencing sensor noise.

Kalman and HMM models: Rolling standard deviation of speed is a practical proxy for process noise covariance when initialising a Kalman filter. The Kalman filter gap-filling approach uses this directly.

Time-series synchronisation: If your pipeline merges rolling features from multiple sensor streams (GPS + accelerometer + CAN bus), synchronise timestamps before rolling — otherwise the merged window will contain observations from different physical instants.

The companion page computing rolling average speed over sliding time windows covers the specific velocity calculation in detail, including projected-CRS distance normalisation.

Validation

After computing rolling metrics, run these checks before writing to production storage:

def validate_rolling_output(
    df: pd.DataFrame,
    window: str,
    min_periods: int,
) -> None:
    """
    Sanity-check a rolling metrics DataFrame.
    Raises AssertionError with a descriptive message on failure.
    """
    assert "speed_mean" in df.columns, "speed_mean column absent"
    assert "speed_std" in df.columns, "speed_std column absent"

    # No negative speeds
    neg = (df["speed_mean"] < 0).sum()
    assert neg == 0, f"{neg} rows have negative rolling mean speed"

    # NaN rate should be explainable by min_periods, not data loss
    nan_rate = df["speed_mean"].isna().mean()
    assert nan_rate < 0.5, (
        f"Rolling mean NaN rate is {nan_rate:.1%} — "
        "check gap masking and min_periods setting"
    )

    # Rolling std should not exceed raw speed range (sanity bound)
    max_std = df["speed_std"].max(skipna=True)
    max_speed = df["speed_mean"].max(skipna=True)
    assert max_std <= max_speed * 2, (
        f"Rolling std ({max_std:.1f}) exceeds 2× rolling mean max "
        f"({max_speed:.1f}) — likely an unsorted or mixed-entity index"
    )

    print(
        f"Validation passed: {len(df):,} rows, "
        f"NaN rate {nan_rate:.1%}, "
        f"max speed_mean {max_speed:.1f} km/h"
    )

Additional checks to run manually:

• Boundary behaviour: Confirm the first min_periods - 1 rows per entity are NaN in speed_mean when there are fewer observations than min_periods in the first window.
• Memory footprint: df.memory_usage(deep=True).sum() / 1e9 — downcast float64 columns to float32 and encode entity_id as category dtype if the result exceeds your RAM budget.
• Monotonicity: Plot speed_mean for a single entity and verify it follows the underlying signal without step jumps (which indicate a sort or groupby key problem).

FAQ

Why does pandas rolling produce NaN for the first few rows even with min_periods=1? Time-based rolling in pandas counts observations within the time window, not row positions. If the first few timestamps all fall within the window duration, min_periods=1 will still produce values — but if your index contains duplicate timestamps or sub-millisecond precision differences, pandas may evaluate an empty window. Deduplicate the index and ensure monotonicity before rolling.

How do I apply rolling per entity without a Python-level loop? Use df.groupby('entity_id')['column'].rolling(window, ...) — pandas propagates the time-based window logic through the groupby without a loop. Reset the groupby level after the operation with .reset_index(level=0, drop=True).

Can I pass multiple aggregation functions in a single .rolling().agg() call? Not with named-column dict syntax. Pandas rolling does not support the {'new_col': ('source_col', 'func')} form that groupby.agg accepts. Call .rolling().mean(), .rolling().std(), etc. separately and concatenate the results.

What window size should I use for hard-braking detection? For hard-braking events you need micro-maneuver resolution: 10–30 s windows at 1 Hz or higher sampling. Larger windows smooth out the impulse entirely. Use rolling().min() on speed or rolling().max() on deceleration magnitude rather than rolling().mean().

How do I prevent look-ahead bias when building ML features from rolling stats? Set closed='right' (the pandas default) so each window contains only observations up to and including the current timestamp. Never use center=True for predictive features. Validate by checking that the rolling mean at time T matches a manual mean of all rows with timestamp in (T − window, T].

Related:

• Computing rolling average speed over sliding time windows — detailed velocity calculation with projected-CRS distance normalisation
• Dynamic time-binning strategies — adapting window boundaries to event density
• Gap-filling in sparse trajectories — interpolation and Kalman approaches before rolling
• Seasonal & cyclical alignment — decomposing rolling features against diurnal and weekly cycles
• Speed & acceleration profiling — downstream consumer of rolling speed and variance features

Back to Temporal Aggregation & Window Mapping