Dynamic Threshold Tuning for Waste Route Optimization

A waste-routing plan is only as good as the parameters it was solved against, and those parameters rot fast. The average municipal travel speed you tuned last quarter is wrong the moment a lane closes for a water-main break; the disposal-site dwell time you hardcoded is wrong the day a neighboring city’s transfer station goes offline and queues double. When route-cost, capacity, and service-window thresholds are frozen constants, the solver keeps returning confident plans built on stale physics — and the failure surfaces as trucks idling past a noise cutoff, overtime the dispatcher never authorized, or a run that misses the transfer station’s closing gate by four minutes.

This page is the calibration sub-system of the VRP Route Optimization Algorithms pipeline. It shows how to turn frozen constants into bounded, telemetry-driven thresholds that shift with observed conditions but can never cross a regulatory ceiling, how those bounds map to the statutes that authorize them, how to prove the tuner adjusts without oscillating, and what the system must do when the telemetry that feeds it goes dark. Every code block is complete and pasteable into a Python 3.10+ dispatch worker. The narrower question of how to move a single penalty weight is handled in the auto-tuning route cost thresholds deep dive; this page owns the whole calibration loop.

Prerequisites

Pin the numerical and validation stack so a tuning decision is reproducible during an audit — a threshold that moved must be explainable months later against the exact library versions that moved it.

python --version   # 3.10 or newer
pip install \
  ortools==9.11.4210 \
  pydantic==2.9.2 \
  numpy==2.1.2

Every threshold the tuner is allowed to touch is described by a typed record, not a loose float. The record carries the soft operating value the solver currently uses, the hard regulatory ceiling it may never exceed, and the learning rate that governs how fast it moves. Separating these three is what keeps a runaway feedback loop from ever producing an illegal plan.

from __future__ import annotations
from enum import Enum
from pydantic import BaseModel, Field, model_validator


class ThresholdKind(str, Enum):
    ROUTE_COST = "route_cost_seconds"      # travel-time penalty budget
    PAYLOAD_KG = "payload_ceiling_kg"      # working capacity target
    SERVICE_WINDOW = "service_window_sec"  # allowed dwell before a stop


class TunableThreshold(BaseModel):
    kind: ThresholdKind
    value: float = Field(gt=0)          # current soft operating value
    hard_ceiling: float = Field(gt=0)   # statutory / physical bound — never crossed
    hard_floor: float = Field(ge=0)     # minimum safe value
    alpha: float = Field(gt=0.0, le=1.0)  # EMA learning rate

    @model_validator(mode="after")
    def bounds_are_sane(self) -> "TunableThreshold":
        if not (self.hard_floor <= self.value <= self.hard_ceiling):
            raise ValueError(
                f"{self.kind}: value {self.value} escapes "
                f"[{self.hard_floor}, {self.hard_ceiling}]"
            )
        return self

The hard_ceiling and hard_floor are the guardrails the entire page is built around: the tuner is free to explore inside that interval and forbidden from leaving it. A PAYLOAD_KG ceiling traces to the same GVWR-derived bound the Capacity & Weight Limits dimension enforces, and a SERVICE_WINDOW floor traces to the municipal noise ordinance encoded as Time Window Constraints. The tuner never invents these numbers — it receives them.

Core Implementation

Tuning is a bounded exponential moving average. Given the current threshold value $\theta_t$ and a freshly observed value $o_t$ (last shift’s actual median disposal dwell, say), the next soft value is:

$$\theta_{t+1} = \operatorname{clip}\bigl(\,(1-\alpha)\,\theta_t + \alpha\,o_t,\; \theta_{\min},\; \theta_{\max}\bigr)$$

The learning rate $\alpha$ trades responsiveness for stability: a high $\alpha$ chases every spike and risks oscillation, a low $\alpha$ is stable but lags a genuine demand shift. The clip is not optional — it is the mathematical expression of the regulatory guardrail, guaranteeing $\theta_{t+1}\in[\theta_{\min},\theta_{\max}]$ no matter how extreme the observation.

The tuner also refuses to react to a single outlier. It keeps a rolling window of recent observations and trips a circuit breaker when the newest reading lands more than three standard deviations from the window mean — the signature of a sensor fault or a one-off road closure, not a trend worth chasing.

import statistics
from collections import deque
from dataclasses import dataclass, field


@dataclass
class BoundedTuner:
    """Bounded-EMA threshold tuner with a 3-sigma circuit breaker."""
    threshold: TunableThreshold
    window: int = 12                       # rolling observations (e.g. one shift = 12 routes)
    sigma_limit: float = 3.0
    _history: deque[float] = field(default_factory=lambda: deque(maxlen=12))
    breaker_tripped: bool = False

    def _is_outlier(self, observation: float) -> bool:
        if len(self._history) < 3:
            return False
        mu = statistics.fmean(self._history)
        sd = statistics.pstdev(self._history)
        if sd == 0:
            return False
        return abs(observation - mu) > self.sigma_limit * sd

    def update(self, observation: float) -> float:
        """Fold one observation into the threshold; return the new soft value."""
        if observation <= 0:
            raise ValueError("observation must be positive")

        if self._is_outlier(observation):
            self.breaker_tripped = True
            # Do NOT move the threshold on an anomalous reading.
            return self.threshold.value

        self.breaker_tripped = False
        self._history.append(observation)

        t = self.threshold
        proposed = (1 - t.alpha) * t.value + t.alpha * observation
        clipped = min(max(proposed, t.hard_floor), t.hard_ceiling)
        t.value = clipped
        return clipped

Binding the tuned value into the solver is a single substitution: the freshly clipped threshold becomes the parameter the OR-Tools Implementation layer registers on its dimension. Because the tuner mutates only the value field and the solver reads that field at model-build time, calibration and optimization stay cleanly decoupled — the solver never sees the tuning machinery, only its output.

from ortools.constraint_solver import pywrapcp


def apply_service_window(
    routing: pywrapcp.RoutingModel,
    manager: pywrapcp.RoutingIndexManager,
    time_dim_name: str,
    tuner: BoundedTuner,
    node_windows: dict[int, tuple[int, int]],
) -> None:
    """Widen each node's allowed dwell by the tuned slack, clipped to its hard window."""
    slack = int(tuner.threshold.value)      # tuned service-window seconds
    time_dim = routing.GetDimensionOrDie(time_dim_name)
    for node, (earliest, latest) in node_windows.items():
        idx = manager.NodeToIndex(node)
        # Tuned slack may relax toward `latest` but never past the hard ordinance bound.
        relaxed_earliest = max(earliest - slack, earliest - int(tuner.threshold.hard_ceiling))
        time_dim.CumulVar(idx).SetRange(relaxed_earliest, latest)

Notice the tuned slack only ever loosens the earliest arrival toward the ordinance-fixed latest; the hard upper bound is left untouched so no amount of calibration can push a pickup past a noise cutoff. This is the same CumulVar range mechanism used throughout the pipeline, driven here by a number that moves with the fleet.

Regulatory Mapping

Every bound the tuner clips against must trace to a rule an inspector can cite. The tuner is where dynamic behavior meets static law, so the mapping is not decorative — it is the definition of each hard_ceiling and hard_floor the TunableThreshold carries.

The route-cost ceiling is the one operators most often get wrong. Auto-tuning the drive-time penalty must never let a shift’s cumulative driving exceed the FMCSA hours-of-service limit under 49 CFR 395.3, so the tuner’s hard_ceiling for ROUTE_COST is derived from remaining legal drive time, and the reconciliation between a tuned cost budget and the driver’s logged on-duty hours is owned by the DOT/FMCSA rule mapping system. Every clip against these ceilings must be recorded; the compliance registry versions the bound that was in force at the moment a threshold moved.

Validation & Verification

A tuner that silently never moves, or one that oscillates every cycle, both give false assurance. Prove three properties with unit tests: the value converges toward a stable observation, it never escapes its bounds under an extreme observation, and the circuit breaker trips on an outlier without moving the threshold.

def test_tuner_converges_but_stays_bounded() -> None:
    th = TunableThreshold(
        kind=ThresholdKind.SERVICE_WINDOW,
        value=300.0, hard_floor=120.0, hard_ceiling=600.0, alpha=0.3,
    )
    tuner = BoundedTuner(threshold=th)

    # Convergence: repeated 480 s observations pull the value up toward 480, not past it.
    for _ in range(20):
        tuner.update(480.0)
    assert 470.0 <= th.value <= 480.0, "EMA failed to converge on the observation"

    # Bounding: an absurd 10,000 s reading is rejected as an outlier and, even if
    # it were folded in, the clip guarantees the value stays under the hard ceiling.
    tuner.update(10_000.0)
    assert th.value <= th.hard_ceiling, "tuned value escaped the regulatory ceiling"


def test_circuit_breaker_ignores_outliers() -> None:
    th = TunableThreshold(
        kind=ThresholdKind.ROUTE_COST,
        value=100.0, hard_floor=10.0, hard_ceiling=1000.0, alpha=0.5,
    )
    tuner = BoundedTuner(threshold=th)
    for _ in range(6):
        tuner.update(100.0)         # build a tight, zero-variance-ish window
    before = th.value
    tuner.update(9_000.0)           # 3-sigma outlier
    assert tuner.breaker_tripped is True
    assert th.value == before, "threshold moved on an anomalous reading"

In production, every accepted or rejected tuning decision emits a structured record. Follow the site-wide JSONFormatter pattern so calibration events land in the audit database as machine-parseable rows, not free text an auditor has to grep.

import hashlib
import json
import logging


class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            "level": record.levelname,
            "event": record.getMessage(),
            "kind": getattr(record, "kind", None),
            "old_value": getattr(record, "old_value", None),
            "new_value": getattr(record, "new_value", None),
            "rule_id": getattr(record, "rule_id", None),
            "breaker_tripped": getattr(record, "breaker_tripped", None),
        }
        # Append-only integrity: hash the decision so tampering is detectable.
        payload["sha256"] = hashlib.sha256(
            json.dumps(payload, sort_keys=True).encode()
        ).hexdigest()
        return json.dumps(payload, separators=(",", ":"))


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


def tune_and_log(tuner: BoundedTuner, observation: float, rule_id: str) -> float:
    old = tuner.threshold.value
    new = tuner.update(observation)
    logger.info(
        "threshold_adjustment",
        extra={"kind": tuner.threshold.kind.value, "old_value": old,
               "new_value": new, "rule_id": rule_id,
               "breaker_tripped": tuner.breaker_tripped},
    )
    return new

The three fields an auditor checks are old_value, new_value, and rule_id: every threshold movement must name the regulation whose ceiling bounded it, and new_value must never exceed that rule’s limit. The sha256 field makes the record self-verifying so a tampered audit row fails its own hash on replay.

Failure Modes & Edge Cases

Dynamic tuning adds a new failure surface: the feedback signal itself can be wrong. The system must degrade toward more conservative behavior, never less, when the inputs are untrustworthy.

1. Telemetry blackout. When post-route feeds — weigh-in-motion scales, disposal-site dwell timers, GPS traces — stop arriving, the tuner has nothing to average. Never extrapolate on a stale buffer; freeze the threshold at its last audited value and flag routes for review. Reconciling and gap-filling those feeds before they ever reach the tuner is the job of the Telematics & Sensor Data Ingestion pipeline.

2. Circuit breaker latched into a degraded state. A run of outliers (a multi-day storm, a closed transfer station) can trip the breaker repeatedly. Rather than thrash, drop to a pre-validated static baseline and stay there until variance normalizes — the same conservative-fallback discipline formalized in the fallback routing logic sub-system.

def resolve_threshold(tuner: BoundedTuner, static_baseline: float,
                      consecutive_trips: int) -> float:
    """Prefer the tuned value; fall back to a static baseline while degraded."""
    DEGRADED_AFTER = 3
    if consecutive_trips >= DEGRADED_AFTER:
        logger.warning(
            "threshold_degraded_fallback",
            extra={"kind": tuner.threshold.kind.value,
                   "old_value": tuner.threshold.value,
                   "new_value": static_baseline, "rule_id": "FALLBACK-STATIC",
                   "breaker_tripped": True},
        )
        # Clamp even the baseline to the hard bounds — belt and suspenders.
        return min(max(static_baseline, tuner.threshold.hard_floor),
                   tuner.threshold.hard_ceiling)
    return tuner.threshold.value

3. Solver infeasibility after a tightening. If the tuner lowers a working ceiling and the next solve returns no feasible plan, the fault is the calibration, not the demand. Roll the threshold back to its previous audited value, re-solve, and log the rollback — a tuned parameter must never be the reason a plan cannot be produced.

4. Oscillation from an over-aggressive learning rate. A threshold that swings wide every cycle signals alpha set too high for the signal’s noise. Halve alpha and widen the rolling window; the convergence unit test above is the guardrail that catches this in CI before it reaches dispatch.

Integration Checklist

Complete every item before a tuning loop reaches production dispatch:

• Every TunableThreshold carries a hard_ceiling and hard_floor derived from a cited regulation, not a guessed number.
• The EMA update clips to those bounds unconditionally — no code path writes a value outside the interval.
• The 3-sigma circuit breaker is enabled and its unit test passes in CI.
• alpha and the window size are tuned so the convergence test shows no oscillation.
• Every threshold movement emits a threshold_adjustment record naming the rule_id and carrying a sha256 integrity hash.
• Telemetry blackout freezes the threshold at its last audited value and flags affected routes.
• Repeated breaker trips fall back to a pre-validated static baseline, itself clamped to the hard bounds.
• A tightening that makes the next solve infeasible triggers an automatic rollback and a logged event.
• The route-cost ceiling is derived from remaining FMCSA legal drive time under 49 CFR 395.3.

FAQ

Why bound the tuner instead of letting it learn the optimal value freely?

Because “optimal” for an unconstrained learner means whatever minimizes cost, and cost is minimized by ignoring regulations. The hard ceiling and floor encode statutory and physical limits the fleet cannot legally cross, so the tuner is only ever allowed to explore inside a legal envelope. An unbounded learner will eventually propose an illegal plan the first time the cost math favors it.

How do I choose the learning rate alpha?

Start low — 0.1 to 0.3 — and let the convergence unit test tell you if it is too slow or too jumpy. A high alpha chases every transient spike and oscillates; a low alpha is stable but lags a genuine demand shift by several cycles. Tune it against replayed historical telemetry, not in production, and widen the rolling window before you raise alpha.

What exactly trips the circuit breaker?

An incoming observation that lands more than three standard deviations from the mean of the rolling window. That is the statistical signature of a sensor fault or a one-off event (a burst water main, a closed transfer station) rather than a trend. When it trips, the threshold does not move and the anomalous reading is excluded from the window so it cannot poison future means.

Should the threshold ever move mid-solve?

No. The tuner mutates the threshold’s value only between planning cycles, and the solver reads that value once at model-build time. Changing a bound mid-solve would make the search non-deterministic and un-auditable. Calibrate on post-route telemetry, then solve the next cycle against the frozen, newly tuned parameter.

How is a threshold change proven legitimate during an audit?

Every adjustment emits a structured record with the old value, new value, the rule_id of the bound that constrained it, and a SHA-256 hash of the decision. The hash makes each row self-verifying — a tampered record fails its own checksum on replay — and the rule_id ties the movement to the specific regulation in force at that moment, so an auditor can reconstruct exactly why a parameter changed.

Up: VRP Route Optimization Algorithms

Related

• Auto-Tuning Route Cost Thresholds — moving a single penalty weight with EMA and Kalman conditioning
• OR-Tools Implementation — dimension registration the tuned value binds into
• Capacity & Weight Limits — GVWR-derived payload ceilings the tuner clips against
• Time Window Constraints — ordinance-fixed service windows that bound tuned slack
• Fallback Routing Logic — conservative static baselines when the breaker latches degraded