Anomaly Scoring for Unexpected Fleet Stops

Confidence scoring for stop detection answers a narrower question than most dispatchers actually care about. It tells you a candidate GPS window really is a stop rather than GPS jitter or a red light — but a confirmed stop can still be completely unremarkable, or it can be the first sign of a stolen trailer, an unauthorized detour, or a driver taking a three-hour break at an unscheduled location. Anomaly scoring for unexpected stops sits downstream of confidence scoring: it takes stops that are already confirmed real and asks whether each one deviates enough from the vehicle’s plan and its own historical pattern to justify a human look.

This page gives a self-contained Python class that scores each confirmed stop using four orthogonal features — a duration z-score against the vehicle’s own history, distance from the nearest planned point of interest, how unusual the time of day is, and whether the stop fell off the planned route corridor — combined into a single 0-100 anomaly score with configurable weights.

Compatibility and Configuration

The Scorer

from __future__ import annotations

import math
from dataclasses import dataclass
from datetime import datetime
from typing import Optional

import numpy as np
import pandas as pd


@dataclass
class AnomalyWeights:
    """Feature weights for the anomaly score. Must sum to 1.0."""

    duration: float = 0.35
    poi_distance: float = 0.30
    time_of_day: float = 0.15
    off_route: float = 0.20

    def validate(self) -> None:
        total = self.duration + self.poi_distance + self.time_of_day + self.off_route
        if not math.isclose(total, 1.0, abs_tol=1e-6):
            raise ValueError(f"Anomaly weights must sum to 1.0, got {total:.4f}")


class StopAnomalyScorer:
    """
    Score confirmed fleet stops for anomaly relative to a vehicle's own
    history and its planned route.

    Parameters
    ----------
    weights : AnomalyWeights
        Relative importance of each feature. Defaults favour duration
        and POI distance, the two strongest predictors of a genuinely
        unplanned stop in last-mile and long-haul fleets alike.
    history_window_days : int
        How many days of past stops to include when computing a
        vehicle's duration baseline. 30 days balances recency against
        having enough samples; shorten to 14 for fast-changing routes,
        lengthen to 60-90 for low-frequency fleets (e.g. weekly runs).
    min_history_stops : int
        Below this many historical stops, fall back to a fleet-wide
        baseline instead of a per-vehicle one to avoid unstable
        z-scores computed on a handful of samples. Default 10.
    distance_threshold_m : float
        Distance from the nearest planned POI beyond which the
        distance feature saturates at maximum anomaly. Default 400 m;
        tighten to 150-200 m for dense urban delivery, loosen to
        800-1000 m for rural or long-haul networks with sparse POIs.
    score_threshold : float
        Score at or above which a stop routes to the review queue or
        an alert. Default 70 on the 0-100 scale. See the Threshold
        section below for how to split review vs. immediate alert.
    """

    def __init__(
        self,
        weights: Optional[AnomalyWeights] = None,
        history_window_days: int = 30,
        min_history_stops: int = 10,
        distance_threshold_m: float = 400.0,
        score_threshold: float = 70.0,
    ):
        self.weights = weights or AnomalyWeights()
        self.weights.validate()
        self.history_window_days = history_window_days
        self.min_history_stops = min_history_stops
        self.distance_threshold_m = distance_threshold_m
        self.score_threshold = score_threshold

    def _duration_zscore_feature(
        self,
        dwell_seconds: float,
        vehicle_history: pd.Series,
        fleet_history: pd.Series,
    ) -> float:
        """
        Compute a normalised [0, 1] anomaly contribution from dwell
        duration. Falls back to fleet_history when vehicle_history has
        fewer than min_history_stops samples (cold start).
        """
        history = (
            vehicle_history
            if len(vehicle_history) >= self.min_history_stops
            else fleet_history
        )
        if len(history) < 2:
            return 0.0  # no usable baseline at all; treat as non-anomalous

        mu = history.mean()
        sigma = history.std(ddof=1)
        if sigma == 0 or np.isnan(sigma):
            return 0.0

        z = abs((dwell_seconds - mu) / sigma)
        # Clip at z=4: beyond four standard deviations, additional
        # deviation no longer increases operational concern linearly.
        return float(np.clip(z / 4.0, 0.0, 1.0))

    def _poi_distance_feature(self, distance_m: float) -> float:
        """Normalise POI distance to [0, 1]; farther = more anomalous."""
        return float(np.clip(distance_m / self.distance_threshold_m, 0.0, 1.0))

    def _time_of_day_feature(
        self, stop_hour: int, vehicle_hour_history: pd.Series
    ) -> float:
        """
        Score how unusual an hour-of-day is for this vehicle using a
        simple circular distance to the historical mode hour. Falls
        back to 0.0 (neutral) when there is no hour history yet.
        """
        if vehicle_hour_history.empty:
            return 0.0
        mode_hour = vehicle_hour_history.mode().iloc[0]
        raw_diff = abs(stop_hour - mode_hour)
        circular_diff = min(raw_diff, 24 - raw_diff)  # wrap around midnight
        return float(np.clip(circular_diff / 12.0, 0.0, 1.0))

    def _off_route_feature(self, off_route: bool) -> float:
        return 1.0 if off_route else 0.0

    def score_stop(
        self,
        dwell_seconds: float,
        poi_distance_m: float,
        stop_time: datetime,
        off_route: bool,
        vehicle_history: pd.Series,
        fleet_history: pd.Series,
        vehicle_hour_history: pd.Series,
    ) -> float:
        """
        Compute a single stop's anomaly score on a 0-100 scale.

        vehicle_history / fleet_history are Series of past dwell_seconds
        values already filtered to history_window_days by the caller.
        vehicle_hour_history is a Series of past stop hours (0-23) for
        the same vehicle, over the same window.
        """
        f_duration = self._duration_zscore_feature(
            dwell_seconds, vehicle_history, fleet_history
        )
        f_distance = self._poi_distance_feature(poi_distance_m)
        f_time = self._time_of_day_feature(stop_time.hour, vehicle_hour_history)
        f_route = self._off_route_feature(off_route)

        score = (
            self.weights.duration * f_duration
            + self.weights.poi_distance * f_distance
            + self.weights.time_of_day * f_time
            + self.weights.off_route * f_route
        )
        return round(float(np.clip(score * 100, 0, 100)), 2)

    def score_batch(
        self,
        stops: pd.DataFrame,
        history: pd.DataFrame,
    ) -> pd.DataFrame:
        """
        Score a batch of confirmed stops.

        stops columns required: vehicle_id, dwell_seconds, poi_distance_m,
        start_time (tz-aware datetime), off_route (bool).
        history columns required: vehicle_id, dwell_seconds, start_time,
        covering at least history_window_days before the earliest stop
        in `stops`.

        Returns stops with an added 'anomaly_score' column and a
        boolean 'flagged' column at score_threshold.
        """
        scores = []
        cutoff_days = pd.Timedelta(days=self.history_window_days)

        for row in stops.itertuples(index=False):
            window_start = row.start_time - cutoff_days
            hist_slice = history[
                (history["start_time"] >= window_start)
                & (history["start_time"] < row.start_time)
            ]
            vehicle_hist = hist_slice.loc[
                hist_slice["vehicle_id"] == row.vehicle_id, "dwell_seconds"
            ]
            fleet_hist = hist_slice["dwell_seconds"]
            vehicle_hour_hist = hist_slice.loc[
                hist_slice["vehicle_id"] == row.vehicle_id, "start_time"
            ].dt.hour

            scores.append(
                self.score_stop(
                    dwell_seconds=row.dwell_seconds,
                    poi_distance_m=row.poi_distance_m,
                    stop_time=row.start_time,
                    off_route=row.off_route,
                    vehicle_history=vehicle_hist,
                    fleet_history=fleet_hist,
                    vehicle_hour_history=vehicle_hour_hist,
                )
            )

        result = stops.copy()
        result["anomaly_score"] = scores
        result["flagged"] = result["anomaly_score"] >= self.score_threshold
        return result

Execution and Tuning Guidelines

Call score_batch once per scoring run (typically hourly or nightly) against a stops DataFrame produced by your stop detection pipeline and a history DataFrame pulled from the same table, filtered to the relevant lookback window before the call — the method itself re-slices per row to respect each stop’s own point-in-time history, which prevents future data from leaking into a score.

• weights — the four weights must sum to 1.0 (validate() raises otherwise). Duration and POI distance are the strongest signals in most fleets and default to a combined 0.65; if your fleet has unreliable POI data, shift weight from poi_distance toward duration and off_route rather than leaving a noisy feature at full weight.
• history_window_days — shorter windows react faster to genuine behavior changes (a new route assignment) but produce noisier baselines; longer windows are more stable but slower to adapt after a legitimate schedule change. 30 days is a reasonable starting point for daily-delivery fleets.
• distance_threshold_m — the distance at which the POI feature saturates. Set this from the actual spread of your POI-to-stop distances for known-good stops, not a guess — plot the distribution and pick roughly the 90th percentile of legitimate visits as the saturation point.
• score_threshold — the line between “no action” and “flagged.” Many teams use two thresholds instead of one: 50-69 routes to a passive review queue, 70+ triggers an active dispatcher alert. Layer a second AnomalyWeights-scored threshold on top of score_batch’s output if you need that split.

Common Pitfalls

Up: Confidence Scoring for Stop Detection | Stop Detection & Dwell Time Analytics

Related

• Confidence Scoring for Stop Detection in Fleet Telematics — the upstream gate that confirms a candidate window is a real stop before anomaly scoring runs
• Location Typing and POI Matching for Stops — the source of the planned-POI distances used as an anomaly feature here
• DBSCAN for Fleet Stop Clustering — one common upstream source of the confirmed stop records this scorer consumes
• Time-Window Based Dwell Calculation — produces the dwell_seconds values that feed the duration z-score feature