Auto-Tuning Route Cost Thresholds in Python

This page solves one narrow production task: recalibrate a single route-cost penalty weight from a rolling window of live telemetry without letting sensor noise, a one-off road closure, or a runaway feedback loop push the threshold to a value that destabilizes the solver or breaks a compliance ceiling.

The broader calibration loop — modelling every tunable parameter, mapping each bound to a statute, and running the whole feedback state machine — belongs to the Dynamic Threshold Tuning sub-system. Here we zoom all the way in on the arithmetic of moving one number safely: condition the raw signal, gate it on a confidence score, clip it against a deviation bound, and fall back deterministically when the telemetry can no longer be trusted.

Environment & Data Prerequisites

Pin the numerical stack so a threshold that moved during a shift is reproducible months later against the exact library versions that moved it.

python --version   # 3.10 or newer
pip install numpy==2.1.2

The tuner consumes one input: a rolling window of normalized per-route costs for a single cost dimension, expressed in a consistent unit. Everything downstream assumes the shape and ranges below — mixing units here is the single most common way to get a threshold that never moves (see Common Errors).

Every raw value must already be conditioned into the same unit before it reaches the tuner. Reconciling ragged sensor feeds — filling gaps, dropping duplicates, aligning timestamps — is the job of the Telematics & Sensor Data Ingestion pipeline; this page assumes a clean, single-unit array arrives.

Step-by-Step Implementation

The flow is five stages: model the inputs, condition the signal, compute a confidence-gated threshold, wire the fallback, then serialize and run. Each block builds on the previous one; paste them in order into a single module.

1. Model the threshold and its compliance profile

Separate the soft operating parameters the tuner may move from the hard compliance profile it may never cross. The ComplianceProfile is received, never invented — its numbers trace to the ordinances the DOT/FMCSA rule mapping system reconciles against logged driver hours.

import logging
import json
import numpy as np
from dataclasses import dataclass, field
from collections import deque


@dataclass
class ComplianceProfile:
    ordinance_id: str
    max_axle_weight_kg: float
    noise_window_start: int          # hour, 24 h clock
    noise_window_end: int
    mandatory_depot_return_min: int  # max minutes before a required depot return
    emission_penalty_multiplier: float = 1.25


@dataclass
class RouteCostThreshold:
    base_fuel_coeff: float
    labor_rate_per_hr: float
    compliance_penalty_floor: float
    telemetry_confidence: float = 1.0
    max_deviation_pct: float = 0.15   # deviation guard: cap per-cycle move
    ema_alpha: float = 0.3            # EMA learning rate
    _history: deque[float] = field(default_factory=lambda: deque(maxlen=500))

2. Condition the raw signal (EMA then scalar Kalman)

Raw per-route costs carry two kinds of noise: high-frequency jitter from GPS drift and RFID bin-scan latency, and slower measurement error. Smooth the jitter with an exponential moving average, then run a scalar Kalman filter to isolate the deterministic travel cost. The EMA of observation $o_t$ is

$$\theta_t = \alpha\,o_t + (1-\alpha)\,\theta_{t-1}$$

and the scalar (1-D) Kalman filter predicts then corrects, with process noise $Q$ and measurement noise $R$:

$$P_t^- = P_{t-1} + Q,\qquad K_t = \frac{P_t^-}{P_t^- + R},\qquad \hat{x}_t = \hat{x}_{t-1} + K_t\,(z_t - \hat{x}_{t-1})$$

The same scalar Kalman recurrence is used to smooth bin coordinates in the GPS Polling Strategies ingestion path — it is a general-purpose noise-isolation primitive, applied here to cost rather than position.

class AutoTuningEngine:
    def __init__(self, baseline: RouteCostThreshold, compliance: ComplianceProfile):
        self.baseline = baseline
        self.compliance = compliance
        self._lock = False
        self._state_estimate = 0.0
        self._error_covariance = 1.0

    def _apply_kalman_filter(self, raw_measurement: float) -> float:
        """Scalar 1-D Kalman filter: predict then update. Q=0.1, R=0.5."""
        self._error_covariance += 0.1                                   # predict
        kalman_gain = self._error_covariance / (self._error_covariance + 0.5)
        self._state_estimate += kalman_gain * (raw_measurement - self._state_estimate)
        self._error_covariance *= (1 - kalman_gain)                    # update
        return self._state_estimate

    def _compute_ema(self, value: float) -> float:
        alpha = self.baseline.ema_alpha
        if not self.baseline._history:
            return value
        prev = self.baseline._history[-1]
        return alpha * value + (1 - alpha) * prev

3. Compute the confidence-gated adaptive threshold

Fold the conditioned window into a mean cost, derive a confidence score from its variance, and only recalibrate when confidence clears 0.85. Below that, the signal is too noisy to trust and the engine locks into the fallback. A deviation guard then caps how far the new threshold may move in one cycle, which is what keeps the solver from divergent, oscillating re-partitioning.

    def compute_adaptive_threshold(self, telemetry_stream: np.ndarray) -> float:
        if self._lock:
            return self._apply_fallback()

        smoothed_costs = []
        for raw_cost in telemetry_stream:
            ema_val = self._compute_ema(float(raw_cost))
            kalman_val = self._apply_kalman_filter(ema_val)
            smoothed_costs.append(kalman_val)

        mean_cost = float(np.mean(smoothed_costs))
        variance = float(np.var(smoothed_costs))
        confidence = max(0.0, 1.0 - (variance / (mean_cost + 1e-6)))

        if confidence < 0.85:
            self._lock = True
            logger.warning(
                "Telemetry confidence degraded. Engaging fallback.",
                extra={"payload": {"confidence": round(confidence, 3)}},
            )
            return self._apply_fallback()

        penalty_multiplier = 1.0
        if mean_cost > self.compliance.mandatory_depot_return_min * 0.5:
            penalty_multiplier = self.compliance.emission_penalty_multiplier

        adaptive_threshold = (
            (self.baseline.base_fuel_coeff * mean_cost)
            + (self.baseline.labor_rate_per_hr * 0.8)
            + (self.baseline.compliance_penalty_floor * penalty_multiplier)
        )

        # Deviation guard: never let one cycle move the threshold past max_deviation_pct.
        if abs(adaptive_threshold - mean_cost) / (mean_cost + 1e-6) > self.baseline.max_deviation_pct:
            adaptive_threshold = mean_cost * (1 + self.baseline.max_deviation_pct)

        self.baseline._history.append(adaptive_threshold)
        self.baseline.telemetry_confidence = confidence

        logger.info(
            "Threshold recalibrated",
            extra={"payload": {
                "adaptive_threshold": round(adaptive_threshold, 2),
                "confidence": round(confidence, 3),
                "penalty_multiplier": penalty_multiplier,
            }},
        )
        return adaptive_threshold

4. Wire the deterministic fallback chain

When confidence drops, the engine must degrade toward more conservative behavior, never less. The fallback returns a fixed, pre-validated value built only from the base coefficient and the hard penalty floor — no telemetry, no learning. This is the same conservative-baseline discipline formalized in the Fallback Routing Logic sub-system. The lock stays engaged until an operator explicitly releases it, so a single bad window cannot silently reactivate tuning.

    def _apply_fallback(self) -> float:
        fallback_val = self.baseline.base_fuel_coeff * 1.0 + self.baseline.compliance_penalty_floor
        logger.info(
            "Fallback threshold applied",
            extra={"payload": {"fallback_value": round(fallback_val, 2)}},
        )
        return fallback_val

    def release_lock(self) -> None:
        self._lock = False

5. Serialize the decision and run it

Every tuning decision emits a structured record. Follow the site-wide JSONFormatter pattern — a custom class, not a stdlib import — so calibration events land in the audit store as machine-parseable rows. Note that logger.info and logger.warning forward structured fields only via the extra= keyword; arbitrary keyword arguments are silently dropped by the standard logging module, so every field above is passed inside extra={"payload": ...}.

class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_obj = {
            "timestamp": self.formatTime(record, "%Y-%m-%dT%H:%M:%S.%fZ"),
            "level": record.levelname,
            "module": record.module,
            "message": record.getMessage(),
            "payload": getattr(record, "payload", None),
        }
        return json.dumps(log_obj, default=str)


logger = logging.getLogger("waste_route_autotuner")
logger.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(JSONFormatter())
logger.addHandler(_handler)


if __name__ == "__main__":
    baseline_cfg = RouteCostThreshold(
        base_fuel_coeff=0.82,
        labor_rate_per_hr=34.50,
        compliance_penalty_floor=12.75,
        ema_alpha=0.25,
        max_deviation_pct=0.12,
    )
    compliance_cfg = ComplianceProfile(
        ordinance_id="MUN-2024-WASTE-09",
        max_axle_weight_kg=11500.0,
        noise_window_start=22,
        noise_window_end=6,
        mandatory_depot_return_min=480,
        emission_penalty_multiplier=1.35,
    )
    engine = AutoTuningEngine(baseline_cfg, compliance_cfg)

    # 24 h rolling window of normalized per-route costs (minutes).
    mock_telemetry = np.array([
        42.1, 41.8, 43.5, 44.2, 41.9, 42.0, 45.1, 46.3, 44.8, 43.2,
        42.5, 41.7, 43.9, 44.1, 42.8, 41.5, 43.0, 44.5, 45.2, 43.8,
        42.2, 41.9, 43.1, 44.0,
    ])
    threshold = engine.compute_adaptive_threshold(mock_telemetry)
    print(f"tuned route-cost threshold = {threshold:.2f}")

The tuned value is the number the OR-Tools Implementation layer registers on its cost dimension at model-build time. Because the tuner only produces a scalar and the solver reads it once per cycle, calibration and optimization stay cleanly decoupled.

Compliance Output

Running the module against the clean example window emits one INFO record. Every field exists to reconstruct, during an audit, exactly why the threshold moved:

{
  "timestamp": "2026-07-02T14:03:11.482Z",
  "level": "INFO",
  "module": "autotuner",
  "message": "Threshold recalibrated",
  "payload": {
    "adaptive_threshold": 48.32,
    "confidence": 0.987,
    "penalty_multiplier": 1.0
  }
}

When confidence collapses, the record instead reads "Telemetry confidence degraded. Engaging fallback." with the failing confidence value — the row that proves the system chose the conservative path rather than tuning on untrustworthy data.

Verification

Prove the two properties that matter: on a clean, low-variance window the engine tunes (does not lock), and on a high-variance window it engages the fallback instead of emitting a wild threshold.

def test_engine_tunes_on_clean_window() -> None:
    engine = _fresh_engine()
    clean = np.array([42.0, 42.3, 41.8, 42.1, 42.4, 41.9, 42.2, 42.0])
    result = engine.compute_adaptive_threshold(clean)
    assert engine._lock is False, "clean window should not trip the fallback"
    assert result > 0.0


def test_engine_falls_back_on_noisy_window() -> None:
    engine = _fresh_engine()
    noisy = np.array([12.0, 88.0, 5.0, 95.0, 3.0, 99.0, 8.0, 91.0])
    result = engine.compute_adaptive_threshold(noisy)
    assert engine._lock is True, "high variance must engage the deterministic fallback"
    # Fallback value is base_fuel_coeff * 1.0 + compliance_penalty_floor.
    assert result == 0.82 * 1.0 + 12.75


def _fresh_engine() -> "AutoTuningEngine":
    baseline = RouteCostThreshold(
        base_fuel_coeff=0.82, labor_rate_per_hr=34.50,
        compliance_penalty_floor=12.75, ema_alpha=0.25, max_deviation_pct=0.12,
    )
    compliance = ComplianceProfile(
        ordinance_id="MUN-2024-WASTE-09", max_axle_weight_kg=11500.0,
        noise_window_start=22, noise_window_end=6,
        mandatory_depot_return_min=480, emission_penalty_multiplier=1.35,
    )
    return AutoTuningEngine(baseline, compliance)

Run with pytest -q. Both tests pass against the module above and pin the behavior that separates a safe tuner from one that chases noise.

Common Errors

ModuleNotFoundError: No module named 'numpy' The tuner depends on NumPy for the mean/variance reduction. Root cause: the pinned dependency was never installed, or the module runs under a different interpreter than the one pip targeted. Fix with pip install numpy==2.1.2 inside the same virtual environment that runs the worker, and confirm with python -c "import numpy; print(numpy.__version__)".

RuntimeWarning: Mean of empty slice → threshold returns nan Passing an empty window (np.array([])) makes np.mean return nan, which then propagates through the whole cost expression. Root cause: a telemetry blackout produced a zero-length window. Guard the entry point — if telemetry_stream.size == 0: return self._apply_fallback() — so a missing feed degrades to the conservative baseline instead of writing nan into the solver’s cost matrix.

Engine permanently locked in fallback (confidence always below 0.85) If every cycle logs "Engaging fallback" even on healthy routes, the confidence score 1 - variance/mean is being crushed by artificially large variance. Root cause is almost always mixed units — seconds and minutes in the same window, or unnormalized raw costs. Normalize upstream so the array matches the 20–90 minute range in the prerequisites table, and only then let the tuner see it; calling release_lock() without fixing the unit mismatch will just re-lock on the next window.

Frequently Asked Questions

Why smooth with both an EMA and a Kalman filter instead of one?

They target different noise. The EMA damps high-frequency jitter cheaply, and the scalar Kalman filter then models the residual measurement error with an adaptive gain, tightening its estimate as the covariance shrinks. Chaining them isolates the deterministic cost signal more cleanly than either alone, at negligible cost for a one-dimensional stream.

What does the 0.85 confidence gate actually protect against?

It stops the tuner from recalibrating on a window whose variance is high relative to its mean — the statistical signature of a sensor fault or a chaotic day. Below the gate the engine locks and returns the fixed fallback, so a noisy feed can never translate into an erratic threshold that forces the solver into repeated re-partitioning.

Why does the deviation guard cap the move instead of rejecting it?

A hard rejection would leave the threshold stale during a genuine, gradual demand shift. Capping the per-cycle move to max_deviation_pct lets the value track real change while bounding how far a single window can drag it, which is what keeps solver convergence stable between planning cycles.

Up: Dynamic Threshold Tuning

Related

• Dynamic Threshold Tuning — the full calibration loop this single-weight tuner plugs into
• OR-Tools Implementation — the dimension the tuned threshold binds into at model-build time
• GPS Polling Strategies — the same scalar Kalman recurrence applied to bin coordinates
• Fallback Routing Logic — conservative static baselines when confidence collapses
• DOT/FMCSA Rule Mapping — how the tuned cost ceiling reconciles against logged hours-of-service