Initial commit

app/services/routing/batch_efficiency.py

@@ -0,0 +1,860 @@
+"""
+Batch Efficiency Analyser
+=========================
+Pure-function service that analyses a delivery batch and returns:
+
+  - fleet_summary        : aggregate metrics + load-balance scores
+  - rider_timelines      : per-rider start/finish/pace/utilisation
+  - substitution_opportunities : ranked, scored transfer plans
+  - top_recommendation   : best action with confidence, root-cause & risk factors
+
+No DB access here.  The route handler owns fetching; this layer is
+fully testable with any list of dicts.
+
+Expected delivery dict keys (all optional except userid):
+  userid          : int   rider id
+  pickupcustomer  : str   e.g. "Daily Grubs Bhuvaneshwari"
+  assigntime      : str   "YYYY-MM-DD HH:MM:SS"
+  pickuptime      : str   "YYYY-MM-DD HH:MM:SS" or None
+  deliverytime    : str   "YYYY-MM-DD HH:MM:SS" or None
+  dlat / droplat  : float delivery lat (either key accepted)
+  dlon / droplon  : float delivery lon (either key accepted)
+  deliveryid      : int   order id (for transfer manifests)
+"""
+
+from __future__ import annotations
+
+import math
+import statistics
+from collections import defaultdict
+from datetime import datetime, timedelta
+from typing import Any
+
+# ---------------------------------------------------------------------------
+# Defaults
+# ---------------------------------------------------------------------------
+
+DEFAULT_KITCHEN_COORDS: dict[str, tuple[float, float]] = {
+    "vidhya":        (11.01633, 77.01478),
+    "jayanthi":      (11.03887, 76.93008),
+    "nandhini":      (11.04324, 77.00068),
+    "bhuvaneshwari": (11.00352, 76.95455),
+    "selvarani":     (10.99274, 77.00535),
+}
+
+DEFAULT_KITCHEN_FRAGMENTS: list[str] = [
+    "bhuvaneshwari", "jayanthi", "nandhini", "vidhya", "selvarani"
+]
+
+DEFAULT_ROAD_KMH: float = 13.0
+DEFAULT_IDLE_THRESHOLD_MIN: float = 30.0
+DEFAULT_MAX_TRANSFER: int = 4
+
+
+# ---------------------------------------------------------------------------
+# Internal helpers
+# ---------------------------------------------------------------------------
+
+def _hav(la1: float, lo1: float, la2: float, lo2: float) -> float:
+    R = 6371.0
+    la1, lo1, la2, lo2 = map(math.radians, [la1, lo1, la2, lo2])
+    a = (math.sin((la2 - la1) / 2) ** 2
+         + math.cos(la1) * math.cos(la2) * math.sin((lo2 - lo1) / 2) ** 2)
+    return R * 2 * math.asin(math.sqrt(max(0.0, min(1.0, a))))
+
+
+def _travel_min(km: float, kmh: float = DEFAULT_ROAD_KMH) -> float:
+    return km / kmh * 60.0
+
+
+def _parse_ts(s: Any) -> datetime | None:
+    if not s:
+        return None
+    s = str(s).strip()
+    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M"):
+        try:
+            return datetime.strptime(s, fmt)
+        except ValueError:
+            continue
+    return None
+
+
+def _fmt(dt: datetime | None) -> str | None:
+    return dt.strftime("%H:%M:%S") if dt else None
+
+
+def _get_coord(order: dict) -> tuple[float, float] | None:
+    lat = order.get("dlat") or order.get("droplat") or order.get("deliverylat")
+    lon = order.get("dlon") or order.get("droplon") or order.get("deliverylong")
+    try:
+        return float(lat), float(lon)
+    except (TypeError, ValueError):
+        return None
+
+
+def _detect_kitchen(
+    pickupcustomer: str | None,
+    fragments: list[str],
+) -> str | None:
+    kl = (pickupcustomer or "").lower()
+    for frag in fragments:
+        if frag in kl:
+            return frag
+    return None
+
+
+def _stdev(values: list[float]) -> float:
+    """Population stdev; returns 0 for fewer than 2 values."""
+    if len(values) < 2:
+        return 0.0
+    try:
+        return statistics.stdev(values)
+    except Exception:
+        return 0.0
+
+
+def _score_candidate(
+    o: dict,
+    arrive_at_kitchen: datetime,
+    k_coord: tuple[float, float],
+    road_kmh: float,
+) -> float:
+    """
+    Score a candidate order for transfer to the idle rider.
+    Higher = better candidate.
+
+    Two components:
+      time_gain  : minutes saved vs the original delivery time.
+                   Positive means idle rider genuinely arrives earlier.
+      geo_penalty: haversine distance from kitchen to the order drop point.
+                   Penalises far orders that inflate the idle rider's extra km.
+
+    Orders with negative time_gain (idle rider would be slower) still get a
+    score, allowing the caller to filter them out with a feasibility check.
+    """
+    d_ts = _parse_ts(o.get("deliverytime"))
+    coord = _get_coord(o)
+    if not d_ts or not coord:
+        return -9999.0
+    dist_km = _hav(k_coord[0], k_coord[1], coord[0], coord[1])
+    est_deliver = arrive_at_kitchen + timedelta(minutes=_travel_min(dist_km, road_kmh))
+    time_gain_min = (d_ts - est_deliver).total_seconds() / 60.0
+    # Weight: time gain matters more than geography (70/30 split).
+    # Penalise 2 min per km of extra distance so nearby clusters float up.
+    return time_gain_min * 0.7 - dist_km * 2.0 * 0.3
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+def analyse_batch(
+    deliveries: list[dict],
+    rider_names: dict[int, str] | None = None,
+    kitchen_coords: dict[str, tuple[float, float]] | None = None,
+    kitchen_fragments: list[str] | None = None,
+    road_kmh: float = DEFAULT_ROAD_KMH,
+    idle_threshold_min: float = DEFAULT_IDLE_THRESHOLD_MIN,
+    max_transfer: int = DEFAULT_MAX_TRANSFER,
+) -> dict:
+    """
+    Analyse a delivery batch and return substitution opportunities.
+
+    Parameters
+    ----------
+    deliveries        : list of order dicts (see module docstring)
+    rider_names       : optional {userid: name} override
+    kitchen_coords    : optional kitchen pickup coordinates override
+    kitchen_fragments : optional kitchen detection strings override
+    road_kmh          : estimated loaded-rider road speed
+    idle_threshold_min: minimum idle window to flag as opportunity
+    max_transfer      : maximum orders to suggest transferring to one rider
+    """
+    if kitchen_coords is None:
+        kitchen_coords = DEFAULT_KITCHEN_COORDS
+    if kitchen_fragments is None:
+        kitchen_fragments = DEFAULT_KITCHEN_FRAGMENTS
+    if rider_names is None:
+        rider_names = {}
+
+    # ------------------------------------------------------------------
+    # 1. Group deliveries by rider
+    # ------------------------------------------------------------------
+    by_rider: dict[int, list[dict]] = defaultdict(list)
+    for o in deliveries:
+        uid = o.get("userid")
+        if uid is not None:
+            by_rider[int(uid)].append(o)
+
+    if not by_rider:
+        return {
+            "fleet_summary": {},
+            "rider_timelines": [],
+            "substitution_opportunities": [],
+            "top_recommendation": None,
+            "error": "No deliveries with valid userid found.",
+        }
+
+    # ------------------------------------------------------------------
+    # 2. Per-rider timeline
+    # ------------------------------------------------------------------
+    timelines: list[dict] = []
+
+    for uid, orders in by_rider.items():
+        # Detect primary kitchen by majority vote
+        kitchen_votes: dict[str, int] = defaultdict(int)
+        for o in orders:
+            k = _detect_kitchen(o.get("pickupcustomer"), kitchen_fragments)
+            if k:
+                kitchen_votes[k] += 1
+        primary_kitchen = (
+            max(kitchen_votes, key=kitchen_votes.get) if kitchen_votes else None
+        )
+        kitchen_confidence = (
+            round(kitchen_votes[primary_kitchen] / len(orders), 2)
+            if primary_kitchen else 0.0
+        )
+
+        # Timestamps
+        finish_ts = None
+        start_ts = None
+        last_coord: tuple[float, float] | None = None
+        completed_orders = 0
+
+        for o in orders:
+            a = _parse_ts(o.get("assigntime"))
+            d = _parse_ts(o.get("deliverytime"))
+            if a and (start_ts is None or a < start_ts):
+                start_ts = a
+            if d:
+                completed_orders += 1
+                if finish_ts is None or d > finish_ts:
+                    finish_ts = d
+                    coord = _get_coord(o)
+                    if coord:
+                        last_coord = coord
+
+        # Active duration and pace
+        active_minutes: float | None = None
+        pace_orders_per_hour: float | None = None
+        if start_ts and finish_ts and finish_ts > start_ts:
+            active_minutes = round((finish_ts - start_ts).total_seconds() / 60, 1)
+            pace_orders_per_hour = round(completed_orders / (active_minutes / 60), 1) if active_minutes else None
+
+        timelines.append({
+            "userid": uid,
+            "name": rider_names.get(uid, f"Rider {uid}"),
+            "kitchen": primary_kitchen,
+            "kitchen_confidence": kitchen_confidence,
+            "order_count": len(orders),
+            "completed_orders": completed_orders,
+            "pending_orders": len(orders) - completed_orders,
+            "started_at": _fmt(start_ts),
+            "finished_at": _fmt(finish_ts),
+            "active_minutes": active_minutes,
+            "pace_orders_per_hour": pace_orders_per_hour,
+            "_finish_dt": finish_ts,
+            "_start_dt": start_ts,
+            "last_position": (
+                {"lat": round(last_coord[0], 6), "lon": round(last_coord[1], 6)}
+                if last_coord else None
+            ),
+            "_last_coord": last_coord,
+        })
+
+    # Sort by finish time
+    timelines.sort(key=lambda t: t["_finish_dt"] or datetime.min)
+
+    # ------------------------------------------------------------------
+    # 3. Fleet summary
+    # ------------------------------------------------------------------
+    valid_start = [t["_start_dt"] for t in timelines if t["_start_dt"]]
+    valid_finish = [t["_finish_dt"] for t in timelines if t["_finish_dt"]]
+
+    fleet_start = min(valid_start) if valid_start else None
+    fleet_done = max(valid_finish) if valid_finish else None
+
+    # Load balance: stdev of order counts and finish-time spread
+    order_counts = [t["order_count"] for t in timelines]
+    finish_offsets_min = (
+        [(f - fleet_start).total_seconds() / 60 for f in valid_finish]
+        if fleet_start and valid_finish else []
+    )
+    finish_spread_min = (
+        round((max(valid_finish) - min(valid_finish)).total_seconds() / 60)
+        if len(valid_finish) >= 2 else 0
+    )
+    load_balance_stdev = round(_stdev([float(c) for c in order_counts]), 2)
+    finish_time_stdev = round(_stdev(finish_offsets_min), 1)
+
+    # Utilisation: how much of the batch window each rider was actively delivering
+    batch_duration_min = (
+        round((fleet_done - fleet_start).total_seconds() / 60)
+        if fleet_start and fleet_done else None
+    )
+    avg_active_minutes = (
+        round(
+            sum(t["active_minutes"] for t in timelines if t["active_minutes"])
+            / max(1, sum(1 for t in timelines if t["active_minutes"])),
+            1,
+        )
+        if any(t["active_minutes"] for t in timelines) else None
+    )
+    avg_utilisation_pct = (
+        round(avg_active_minutes / batch_duration_min * 100, 1)
+        if avg_active_minutes and batch_duration_min else None
+    )
+
+    fleet_summary = {
+        "total_orders": len(deliveries),
+        "total_riders": len(by_rider),
+        "fleet_start": _fmt(fleet_start),
+        "fleet_done": _fmt(fleet_done),
+        "total_duration_minutes": batch_duration_min,
+        "orders_per_rider_avg": round(len(deliveries) / len(by_rider), 1),
+        "load_balance_stdev": load_balance_stdev,
+        "finish_time_spread_minutes": finish_spread_min,
+        "finish_time_stdev_minutes": finish_time_stdev,
+        "avg_utilisation_pct": avg_utilisation_pct,
+        "avg_active_minutes": avg_active_minutes,
+    }
+
+    if not fleet_done:
+        return {
+            "fleet_summary": fleet_summary,
+            "rider_timelines": _clean_timelines(timelines, fleet_done),
+            "substitution_opportunities": [],
+            "top_recommendation": None,
+            "error": "No completed deliveries found.",
+        }
+
+    # Mark each rider's idle status and free window
+    for t in timelines:
+        fd = t["_finish_dt"]
+        if fd:
+            idle_min = (fleet_done - fd).total_seconds() / 60.0
+            t["idle_minutes"] = round(idle_min)
+            t["free_window_minutes"] = round(idle_min)  # time available for substitution
+            t["status"] = "idle" if idle_min >= idle_threshold_min else "active"
+        else:
+            t["idle_minutes"] = 0
+            t["free_window_minutes"] = 0
+            t["status"] = "unknown"
+
+    # ------------------------------------------------------------------
+    # 4. Substitution opportunities
+    # ------------------------------------------------------------------
+    opportunities: list[dict] = []
+
+    idle_riders = [t for t in timelines if t["status"] == "idle" and t["_last_coord"]]
+
+    for idle in idle_riders:
+        idle_uid = idle["userid"]
+        idle_finish = idle["_finish_dt"]
+        idle_coord = idle["_last_coord"]
+        free_window_min = idle["free_window_minutes"]
+
+        if idle["kitchen"] is None:
+            continue  # can't determine origin kitchen
+
+        for target_kitchen, k_coord in kitchen_coords.items():
+            if target_kitchen == idle["kitchen"]:
+                continue  # skip own kitchen
+
+            # Travel from idle rider's last drop to the target kitchen
+            travel_km = _hav(idle_coord[0], idle_coord[1], k_coord[0], k_coord[1])
+            travel_min = _travel_min(travel_km, road_kmh)
+
+            # Skip if idle rider can't even reach the kitchen before fleet is done
+            if travel_min >= free_window_min:
+                continue
+
+            arrive_at_kitchen = idle_finish + timedelta(minutes=travel_min)
+
+            # ----------------------------------------------------------
+            # Candidate selection: score every order from this kitchen
+            # that was delivered after the idle rider could arrive.
+            # Score = time_gain (70%) + geo proximity (30%).
+            # This prefers orders where idle rider is genuinely faster
+            # AND that are close to the kitchen (less detour).
+            # ----------------------------------------------------------
+            candidate_orders: list[tuple[int, dict, float]] = []  # (rid, order, score)
+            for rid, r_orders in by_rider.items():
+                if rid == idle_uid:
+                    continue
+                r_kitchen = next(
+                    (t["kitchen"] for t in timelines if t["userid"] == rid), None
+                )
+                if r_kitchen != target_kitchen:
+                    continue
+                for o in r_orders:
+                    d_ts = _parse_ts(o.get("deliverytime"))
+                    if not d_ts or d_ts <= arrive_at_kitchen:
+                        continue
+                    score = _score_candidate(o, arrive_at_kitchen, k_coord, road_kmh)
+                    candidate_orders.append((rid, o, score))
+
+            if not candidate_orders:
+                continue
+
+            # Sort best candidates first (highest score = most time gained, closest)
+            candidate_orders.sort(key=lambda x: x[2], reverse=True)
+            take_pool = candidate_orders[:max_transfer]
+
+            # Greedy nearest-neighbour route from the kitchen through selected orders
+            unvisited = list(range(len(take_pool)))
+            curr_nn = k_coord
+            greedy_take: list[tuple[int, dict]] = []
+            while unvisited:
+                ni = min(
+                    unvisited,
+                    key=lambda i: (
+                        _hav(curr_nn[0], curr_nn[1], *c)
+                        if (c := _get_coord(take_pool[i][1])) else 999.0
+                    ),
+                )
+                unvisited.remove(ni)
+                greedy_take.append((take_pool[ni][0], take_pool[ni][1]))
+                coord = _get_coord(take_pool[ni][1])
+                if coord:
+                    curr_nn = coord
+
+            # Simulate idle rider executing the greedy route
+            curr_pos = k_coord
+            est_time = arrive_at_kitchen
+            transfer_manifests: list[dict] = []
+            total_delivery_leg_min = 0.0
+
+            for orig_rid, o in greedy_take:
+                coord = _get_coord(o)
+                if not coord:
+                    continue
+                d_km = _hav(curr_pos[0], curr_pos[1], coord[0], coord[1])
+                d_min = _travel_min(d_km, road_kmh)
+                total_delivery_leg_min += d_min
+                est_deliver = est_time + timedelta(minutes=d_min)
+                orig_deliver = _parse_ts(o.get("deliverytime"))
+
+                improvement = (
+                    round((orig_deliver - est_deliver).total_seconds() / 60)
+                    if orig_deliver and est_deliver else None
+                )
+                is_feasible = improvement is not None and improvement > 0
+
+                transfer_manifests.append({
+                    "deliveryid": o.get("deliveryid"),
+                    "from_rider_id": orig_rid,
+                    "from_rider_name": rider_names.get(orig_rid, f"Rider {orig_rid}"),
+                    "original_delivery_time": _fmt(orig_deliver),
+                    "estimated_delivery_time": _fmt(est_deliver),
+                    "improvement_minutes": improvement,
+                    "is_feasible": is_feasible,
+                    "location": {"lat": round(coord[0], 6), "lon": round(coord[1], 6)},
+                })
+                curr_pos = coord
+                est_time = est_deliver
+
+            # Only keep orders where idle rider is actually faster
+            feasible_manifests = [m for m in transfer_manifests if m["is_feasible"]]
+            if not feasible_manifests:
+                continue
+
+            idle_new_finish = est_time
+
+            # Check idle rider can complete within their free window
+            total_obligation_min = travel_min + total_delivery_leg_min
+            if total_obligation_min > free_window_min:
+                # Idle rider would finish after fleet, extending rather than helping
+                continue
+
+            # New fleet done after the transfer
+            taken_order_objs = {id(o) for (_, o) in greedy_take}
+
+            new_finish_by_rider: dict[int, datetime | None] = {}
+            for rid, r_orders in by_rider.items():
+                if rid == idle_uid:
+                    new_finish_by_rider[rid] = idle_new_finish
+                    continue
+                remaining = [o for o in r_orders if id(o) not in taken_order_objs]
+                finishes = [_parse_ts(o.get("deliverytime")) for o in remaining]
+                valid = [f for f in finishes if f]
+                new_finish_by_rider[rid] = max(valid) if valid else None
+
+            new_fleet_done = max(
+                (v for v in new_finish_by_rider.values() if v),
+                default=fleet_done,
+            )
+            fleet_improvement = round(
+                (fleet_done - new_fleet_done).total_seconds() / 60
+            )
+
+            # Most relieved rider
+            orig_last_by_rider = {
+                rid: max(
+                    (f for f in [_parse_ts(o.get("deliverytime")) for o in r_o] if f),
+                    default=None,
+                )
+                for rid, r_o in by_rider.items()
+            }
+            most_impacted_rid = max(
+                (rid for rid in {r for r, _ in greedy_take}),
+                key=lambda r: (orig_last_by_rider.get(r) or datetime.min),
+                default=None,
+            )
+            orig_overloaded_finish = orig_last_by_rider.get(most_impacted_rid)
+            new_overloaded_finish = new_finish_by_rider.get(most_impacted_rid)
+            time_saved = (
+                round((orig_overloaded_finish - new_overloaded_finish).total_seconds() / 60)
+                if orig_overloaded_finish and new_overloaded_finish else 0
+            )
+
+            # Total extra km for idle rider (idle→kitchen + all delivery legs)
+            total_extra_km = travel_km + sum(
+                _hav(
+                    (k_coord if i == 0 else (_get_coord(greedy_take[i-1][1]) or k_coord))[0],
+                    (k_coord if i == 0 else (_get_coord(greedy_take[i-1][1]) or k_coord))[1],
+                    *(_get_coord(o) or k_coord),
+                )
+                for i, (_, o) in enumerate(greedy_take)
+                if _get_coord(o)
+            )
+
+            # ----------------------------------------------------------
+            # Confidence score (0-100)
+            # Measures how comfortable this transfer is given real constraints.
+            #
+            # Component 1 – Slack ratio: how much free time the idle rider
+            #   has beyond the time they'll spend doing the transfer.
+            #   (free_window - total_obligation) / free_window → 0..1
+            #
+            # Component 2 – Feasibility ratio: what fraction of the
+            #   transferred orders actually deliver earlier than original.
+            #   feasible_count / total_transferred → 0..1
+            #
+            # Component 3 – Fleet gain ratio: minutes saved as a fraction
+            #   of total batch duration.  Capped at 20 min improvement for
+            #   full score so small batches don't produce inflated scores.
+            # ----------------------------------------------------------
+            slack_ratio = max(0.0, (free_window_min - total_obligation_min) / free_window_min)
+            feasibility_ratio = len(feasible_manifests) / max(1, len(transfer_manifests))
+            fleet_gain_ratio = min(1.0, fleet_improvement / 20.0) if fleet_improvement > 0 else 0.0
+
+            confidence_score = round(
+                (slack_ratio * 0.4 + feasibility_ratio * 0.35 + fleet_gain_ratio * 0.25) * 100
+            )
+
+            efficiency_ratio = (
+                round(fleet_improvement / max(0.1, total_extra_km), 2)
+                if total_extra_km > 0 else 0.0
+            )
+            if efficiency_ratio >= 5:
+                efficiency_rating = "high"
+            elif efficiency_ratio >= 2:
+                efficiency_rating = "medium"
+            else:
+                efficiency_rating = "low"
+
+            opportunities.append({
+                "idle_rider": {
+                    "userid": idle_uid,
+                    "name": idle["name"],
+                    "primary_kitchen": idle["kitchen"],
+                    "order_count": idle["order_count"],
+                    "finished_at": _fmt(idle_finish),
+                    "idle_minutes": idle["idle_minutes"],
+                    "free_window_minutes": free_window_min,
+                    "last_position": idle["last_position"],
+                },
+                "target_kitchen": target_kitchen,
+                "travel_to_kitchen_km": round(travel_km, 1),
+                "travel_to_kitchen_minutes": round(travel_min),
+                "arrive_at_kitchen": _fmt(arrive_at_kitchen),
+                "orders_to_transfer": transfer_manifests,
+                "total_orders_transferred": len(feasible_manifests),
+                "feasible_orders_count": len(feasible_manifests),
+                "most_relieved_rider": {
+                    "userid": most_impacted_rid,
+                    "name": rider_names.get(most_impacted_rid, f"Rider {most_impacted_rid}"),
+                    "original_finish": _fmt(orig_overloaded_finish),
+                    "new_finish": _fmt(new_overloaded_finish),
+                    "time_saved_minutes": time_saved,
+                },
+                "extra_km_for_idle_rider": round(total_extra_km, 1),
+                "total_obligation_minutes": round(total_obligation_min),
+                "idle_rider_new_finish": _fmt(idle_new_finish),
+                "original_fleet_done": _fmt(fleet_done),
+                "new_fleet_done": _fmt(new_fleet_done),
+                "fleet_improvement_minutes": fleet_improvement,
+                "confidence_score": confidence_score,
+                "efficiency_ratio": efficiency_ratio,
+                "efficiency_rating": efficiency_rating,
+            })
+
+    # Keep only net-positive opportunities
+    opportunities = [o for o in opportunities if o["fleet_improvement_minutes"] > 0]
+
+    # Sort: confidence first (overall quality), then fleet improvement, then rider time saved
+    opportunities.sort(
+        key=lambda x: (
+            -x["confidence_score"],
+            -x["fleet_improvement_minutes"],
+            -x["most_relieved_rider"]["time_saved_minutes"],
+        )
+    )
+
+    # ------------------------------------------------------------------
+    # 5. Top recommendation
+    # ------------------------------------------------------------------
+    top_recommendation = _build_recommendation(
+        opportunities, idle_threshold_min, fleet_done, timelines, fleet_summary
+    )
+
+    return {
+        "fleet_summary": fleet_summary,
+        "rider_timelines": _clean_timelines(timelines, fleet_done),
+        "substitution_opportunities": opportunities,
+        "top_recommendation": top_recommendation,
+    }
+
+
+# ---------------------------------------------------------------------------
+# Helpers for clean output
+# ---------------------------------------------------------------------------
+
+def _clean_timelines(
+    timelines: list[dict],
+    fleet_done: datetime | None,
+) -> list[dict]:
+    out = []
+    for t in timelines:
+        out.append({
+            "userid": t["userid"],
+            "name": t["name"],
+            "kitchen": t["kitchen"],
+            "kitchen_confidence": t.get("kitchen_confidence", 0.0),
+            "order_count": t["order_count"],
+            "completed_orders": t.get("completed_orders", t["order_count"]),
+            "pending_orders": t.get("pending_orders", 0),
+            "started_at": t["started_at"],
+            "finished_at": t["finished_at"],
+            "active_minutes": t.get("active_minutes"),
+            "pace_orders_per_hour": t.get("pace_orders_per_hour"),
+            "idle_minutes": t.get("idle_minutes", 0),
+            "free_window_minutes": t.get("free_window_minutes", 0),
+            "status": t.get("status", "unknown"),
+            "last_position": t.get("last_position"),
+        })
+    return out
+
+
+def _build_recommendation(
+    opportunities: list[dict],
+    idle_threshold: float,
+    fleet_done: datetime | None,
+    timelines: list[dict],
+    fleet_summary: dict,
+) -> dict | None:
+    if not opportunities:
+        # Diagnose WHY there are no opportunities even if some riders were idle
+        idle_count = sum(1 for t in timelines if t.get("status") == "idle")
+        if idle_count == 0:
+            reason = "All riders finished within the idle threshold window — batch was well balanced."
+        else:
+            reason = (
+                f"{idle_count} rider(s) finished early but no feasible substitution found: "
+                "either travel time exceeds the idle window, or all candidate orders "
+                "would be delivered later by the idle rider than the original."
+            )
+        return {
+            "action": "none",
+            "reason": reason,
+            "fleet_balance_assessment": _balance_assessment(fleet_summary),
+        }
+
+    best = opportunities[0]
+    idle = best["idle_rider"]
+    target = best["target_kitchen"]
+    relieved = best["most_relieved_rider"]
+    primary_kitchen = idle["primary_kitchen"] or "unknown"
+    confidence = best["confidence_score"]
+
+    # Root cause: why was this rider idle?
+    root_cause = _diagnose_root_cause(idle, timelines, fleet_summary)
+
+    # Risk factors
+    risk_factors = _identify_risks(best, idle_threshold)
+
+    description = (
+        f"{idle['name']} ({primary_kitchen}) finished all {idle['order_count']} orders "
+        f"at {idle['finished_at']} — {idle['idle_minutes']} min before the fleet finished. "
+        f"Assigning {best['feasible_orders_count']} {target} orders: "
+        f"travel {best['travel_to_kitchen_km']} km ({best['travel_to_kitchen_minutes']} min), "
+        f"arrive at {target} kitchen at {best['arrive_at_kitchen']}. "
+        f"Relieves {relieved['name']} by {relieved['time_saved_minutes']} min "
+        f"({relieved['original_finish']} → {relieved['new_finish']}). "
+        f"Fleet finishes {best['fleet_improvement_minutes']} min earlier "
+        f"({best['original_fleet_done']} → {best['new_fleet_done']}). "
+        f"Confidence: {confidence}/100."
+    )
+
+    # Dynamic thresholds derived from the actual batch data
+    idle_rider_loads = [t["order_count"] for t in timelines if t.get("kitchen") == primary_kitchen]
+    target_rider_loads = [t["order_count"] for t in timelines if t.get("kitchen") == target]
+    activate_idle_threshold = max(6, idle.get("order_count", 6) + 2)
+    activate_target_threshold = max(8, round(sum(target_rider_loads) / max(1, len(target_rider_loads)) * 1.1))
+
+    activate_rule = {
+        "condition": "AND",
+        "rules": [
+            {
+                "field": f"{primary_kitchen}_order_count",
+                "operator": "<=",
+                "value": activate_idle_threshold,
+                "reason": (
+                    f"{idle['name']} had {idle['order_count']} orders today and was idle "
+                    f"{idle['idle_minutes']} min. Dual-kitchen kicks in when their load "
+                    f"stays at or below {activate_idle_threshold}."
+                ),
+            },
+            {
+                "field": f"{target}_order_count",
+                "operator": ">=",
+                "value": activate_target_threshold,
+                "reason": (
+                    f"{target.capitalize()} had enough orders today to justify the detour "
+                    f"({sum(target_rider_loads)} total across {len(target_rider_loads)} rider(s)). "
+                    f"Activate when {target} load is ≥ {activate_target_threshold}."
+                ),
+            },
+        ],
+    }
+
+    return {
+        "action": "dual_kitchen_assignment",
+        "idle_rider_id": idle["userid"],
+        "idle_rider_name": idle["name"],
+        "primary_kitchen": primary_kitchen,
+        "second_kitchen": target,
+        "second_kitchen_dispatch_after": best["arrive_at_kitchen"],
+        "description": description,
+        "fleet_improvement_minutes": best["fleet_improvement_minutes"],
+        "confidence_score": confidence,
+        "efficiency_rating": best["efficiency_rating"],
+        "root_cause": root_cause,
+        "risk_factors": risk_factors,
+        "activate_when": activate_rule,
+        "fleet_balance_assessment": _balance_assessment(fleet_summary),
+        "api_hint": {
+            "endpoint": "/api/v1/optimize",
+            "note": (
+                f"In the next batch, pre-assign the last "
+                f"{best['feasible_orders_count']} {target} orders to "
+                f"rider {idle['userid']} ({idle['name']}) with a "
+                f"dispatch-after time of {best['arrive_at_kitchen']}."
+            ),
+        },
+    }
+
+
+def _diagnose_root_cause(
+    idle: dict,
+    timelines: list[dict],
+    fleet_summary: dict,
+) -> str:
+    """
+    Explain WHY this rider finished early by comparing their load
+    against the fleet average and their kitchen's order volume.
+    """
+    avg_orders = fleet_summary.get("orders_per_rider_avg", 0)
+    rider_orders = idle["order_count"]
+    kitchen = idle["primary_kitchen"] or "their kitchen"
+    name = idle["name"]
+
+    if avg_orders > 0 and rider_orders < avg_orders * 0.7:
+        shortfall = round(avg_orders - rider_orders, 1)
+        return (
+            f"{name} received {rider_orders} orders vs fleet average of {avg_orders:.1f} "
+            f"(−{shortfall:.1f}). {kitchen.capitalize()} kitchen generated fewer orders "
+            f"than the fleet needed to keep this rider fully utilised. "
+            f"This is a systematic under-loading of the {kitchen} kitchen in this batch."
+        )
+    elif rider_orders <= 4:
+        return (
+            f"{name} had only {rider_orders} orders — a very light load regardless of fleet average. "
+            f"Likely a short-demand window at {kitchen} kitchen. "
+            f"Dual-kitchen assignment is especially effective when primary kitchen load is ≤ 4 orders."
+        )
+    else:
+        spread = fleet_summary.get("finish_time_spread_minutes", 0)
+        return (
+            f"{name} is simply faster than peers — finished {idle['idle_minutes']} min ahead "
+            f"despite a normal load of {rider_orders} orders. "
+            f"Fleet finish-time spread is {spread} min, indicating uneven workload distribution."
+        )
+
+
+def _identify_risks(best: dict, idle_threshold: float) -> list[str]:
+    """
+    Enumerate operational risks for the recommended substitution.
+    """
+    risks: list[str] = []
+    travel_min = best["travel_to_kitchen_minutes"]
+    obligation = best["total_obligation_minutes"]
+    free_window = best["idle_rider"]["free_window_minutes"]
+    confidence = best["confidence_score"]
+    extra_km = best["extra_km_for_idle_rider"]
+
+    slack_min = free_window - obligation
+    if slack_min < 10:
+        risks.append(
+            f"Tight schedule: only {slack_min} min of slack between idle rider's "
+            f"estimated finish and fleet completion. Any delay (traffic, kitchen wait) "
+            f"would eliminate the benefit."
+        )
+    if travel_min > 15:
+        risks.append(
+            f"Long commute to target kitchen ({travel_min} min). "
+            f"Kitchen departure time must be precise — a late start erodes time savings."
+        )
+    if extra_km > 8:
+        risks.append(
+            f"Extra {extra_km} km for the idle rider adds fuel cost and rider fatigue. "
+            f"Verify this is worthwhile if fleet improvement is marginal."
+        )
+    if confidence < 50:
+        risks.append(
+            f"Low confidence ({confidence}/100): limited slack or few feasible transfers. "
+            f"Consider this as a contingency plan rather than a guaranteed improvement."
+        )
+    if best.get("feasible_orders_count", 0) < best.get("total_orders_transferred", 1):
+        risks.append(
+            "Not all proposed transfers save time — some orders are included to fill "
+            "the idle rider's route but don't improve individual delivery times."
+        )
+    if not risks:
+        risks.append("No significant risks identified. Transfer looks operationally sound.")
+
+    return risks
+
+
+def _balance_assessment(fleet_summary: dict) -> str:
+    """Short human-readable verdict on batch balance quality."""
+    spread = fleet_summary.get("finish_time_spread_minutes", 0)
+    stdev = fleet_summary.get("load_balance_stdev", 0)
+    util = fleet_summary.get("avg_utilisation_pct")
+
+    if spread <= 10 and stdev <= 1:
+        verdict = "Excellent — riders finished close together with balanced loads."
+    elif spread <= 20 and stdev <= 2:
+        verdict = "Good — minor imbalance, acceptable for this fleet size."
+    elif spread <= 35:
+        verdict = f"Moderate imbalance — {spread} min spread between earliest and latest finish."
+    else:
+        verdict = (
+            f"High imbalance — {spread} min spread. Some riders sat idle while others overran. "
+            f"Pre-planning dual-kitchen assignments is strongly recommended."
+        )
+
+    if util is not None:
+        verdict += f" Average rider utilisation: {util}% of batch window."
+    return verdict

app/services/routing/clustering_service.py

@@ -0,0 +1,300 @@
+"""
+Geographic Clustering Service for Order Assignment
+Uses K-means clustering to group orders by kitchen location.
+Enhanced with Geohash encoding for spatial learning (idea.txt data encoding).
+"""
+
+import logging
+import numpy as np
+from typing import List, Dict, Any, Tuple
+from collections import defaultdict
+from math import radians, cos, sin, asin, sqrt
+
+logger = logging.getLogger(__name__)
+
+
+class GeohashEncoder:
+    """
+    Geohash Encoding for spatial data.
+    Converts lat/lon coordinates to grid cell strings for locality capture.
+
+    Precision levels:
+    - 4 chars: ~156km x 156km (regional)
+    - 5 chars: ~39km x 19km (city-level)
+    - 6 chars: ~4.9km x 4.9km (neighborhood)
+    - 7 chars: ~1.2km x 609m (local zone)
+    """
+
+    BASE32 = "0123456789bcdefghjkmnpqrstuvwxyz"
+
+    @classmethod
+    def encode(cls, lat: float, lon: float, precision: int = 6) -> str:
+        """
+        Encode lat/lon to geohash string.
+
+        Args:
+            lat: Latitude (-90 to 90)
+            lon: Longitude (-180 to 180)
+            precision: Number of characters (4-12)
+
+        Returns:
+            Geohash string
+        """
+        if lat == 0 and lon == 0:
+            return "unknown"
+
+        lat_min, lat_max = -90.0, 90.0
+        lon_min, lon_max = -180.0, 180.0
+
+        hash_chars = []
+        is_lon = True
+
+        for _ in range(precision):
+            char_bits = 0
+            for _ in range(5):
+                if is_lon:
+                    mid = (lon_min + lon_max) / 2
+                    if lon >= mid:
+                        char_bits = (char_bits << 1) | 1
+                        lon_min = mid
+                    else:
+                        char_bits = char_bits << 1
+                        lon_max = mid
+                else:
+                    mid = (lat_min + lat_max) / 2
+                    if lat >= mid:
+                        char_bits = (char_bits << 1) | 1
+                        lat_min = mid
+                    else:
+                        char_bits = char_bits << 1
+                        lat_max = mid
+                is_lon = not is_lon
+            hash_chars.append(cls.BASE32[char_bits])
+
+        return "".join(hash_chars)
+
+    @classmethod
+    def get_zone_from_geohash(cls, geohash: str) -> Dict[str, Any]:
+        """
+        Extract zone metadata from geohash for ML features.
+
+        Returns:
+            Dict with zone info: zone_id, precision, cell_size, etc.
+        """
+        if not geohash or geohash == "unknown":
+            return {"zone_id": "unknown", "precision": 0}
+
+        precision = len(geohash)
+
+        # Approximate cell sizes (in km)
+        lat_error = 180.0 / (2 ** (precision * 5 // 2)) / 2
+        lon_error = 360.0 / (2 ** (precision * 5 // 2 + 1)) / 2
+
+        # Cell size approximation
+        cell_width_km = lat_error * 111  # 1 degree lat ≈ 111km
+        cell_height_km = (
+            lon_error * 111 * cos(11 * 3.14159 / 180)
+        )  # Adjust for Coimbatore lat
+
+        return {
+            "zone_id": geohash,
+            "precision": precision,
+            "cell_width_km": round(cell_width_km, 3),
+            "cell_height_km": round(cell_height_km, 3),
+            "prefix_4": geohash[:4] if len(geohash) >= 4 else geohash,
+            "prefix_5": geohash[:5] if len(geohash) >= 5 else geohash,
+            "prefix_6": geohash[:6] if len(geohash) >= 6 else geohash,
+        }
+
+
+class ClusteringService:
+    """Clusters orders geographically to enable balanced rider assignment."""
+
+    def __init__(self):
+        self.earth_radius_km = 6371
+        self.geohash_encoder = GeohashEncoder()
+
+    def haversine(self, lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+        """Calculate distance between two points in km."""
+        lon1, lat1, lon2, lat2 = map(
+            radians, [float(lon1), float(lat1), float(lon2), float(lat2)]
+        )
+        dlon = lon2 - lon1
+        dlat = lat2 - lat1
+        a = sin(dlat / 2) ** 2 + cos(lat1) * cos(lat2) * sin(dlon / 2) ** 2
+        c = 2 * asin(min(1.0, sqrt(a)))
+        return c * self.earth_radius_km
+
+    def get_kitchen_location(self, order: Dict[str, Any]) -> Tuple[float, float]:
+        """Extract kitchen coordinates from order."""
+        try:
+            lat = float(order.get("pickuplat", 0))
+            lon = float(order.get("pickuplon") or order.get("pickuplong", 0))
+            if lat != 0 and lon != 0:
+                return lat, lon
+        except (ValueError, TypeError):
+            pass
+        return 0.0, 0.0
+
+    def _encode_location_features(self, lat: float, lon: float) -> Dict[str, Any]:
+        """
+        Encode location with multiple spatial features (Data Encoding: Geohash + Distance).
+
+        Returns:
+            Dict with geohash, zone info, distance features
+        """
+        features = {}
+
+        # Primary geohash encoding (6 chars = ~5km zone for Coimbatore)
+        geohash_6 = self.geohash_encoder.encode(lat, lon, 6)
+        features["geohash_6"] = geohash_6
+
+        # Fine-grained geohash (7 chars = ~1.2km zone)
+        geohash_7 = self.geohash_encoder.encode(lat, lon, 7)
+        features["geohash_7"] = geohash_7
+
+        # Coarse geohash for regional grouping (4 chars = ~156km)
+        geohash_4 = self.geohash_encoder.encode(lat, lon, 4)
+        features["geohash_4"] = geohash_4
+
+        # Zone metadata
+        features["zone_info"] = self.geohash_encoder.get_zone_from_geohash(geohash_6)
+
+        return features
+
+    def cluster_orders_by_kitchen(
+        self, orders: List[Dict[str, Any]], max_cluster_radius_km: float = 3.0
+    ) -> List[Dict[str, Any]]:
+        """
+        Cluster orders by kitchen proximity.
+
+        Returns list of clusters, each containing:
+        - centroid: (lat, lon) of cluster center
+        - orders: list of orders in this cluster
+        - kitchen_names: set of kitchen names in cluster
+        - total_orders: count
+        - geohash_6: geohash encoding of centroid (NEW)
+        - zone_info: zone metadata (NEW)
+        - spatial_features: all location encodings (NEW)
+        """
+        if not orders:
+            return []
+
+        # Group by kitchen location
+        kitchen_groups = defaultdict(list)
+        kitchen_coords = {}
+
+        for order in orders:
+            k_name = self._get_kitchen_name(order)
+            k_lat, k_lon = self.get_kitchen_location(order)
+
+            if k_lat == 0:
+                # Fallback: use delivery location if pickup missing
+                k_lat = float(order.get("deliverylat", 0))
+                k_lon = float(order.get("deliverylong", 0))
+
+            if k_lat != 0:
+                kitchen_groups[k_name].append(order)
+                kitchen_coords[k_name] = (k_lat, k_lon)
+
+        # Now cluster kitchens that are close together
+        clusters = []
+        processed_kitchens = set()
+
+        for k_name, k_orders in kitchen_groups.items():
+            if k_name in processed_kitchens:
+                continue
+
+            # Start a new cluster with this kitchen
+            cluster_kitchens = [k_name]
+            cluster_orders = k_orders[:]
+            processed_kitchens.add(k_name)
+
+            k_lat, k_lon = kitchen_coords[k_name]
+
+            # ── CHAIN-MERGE FIX ──────────────────────────────────────────
+            # Original bug: only checked against the SEED kitchen.
+            # Example: A→B = 2km, B→C = 2km, A→C = 4km (max_radius = 3km).
+            # Old code: C is NOT merged (A→C > 3km).
+            # New code: iteratively re-check remaining kitchens against the
+            # CURRENT centroid after each merge, so chains like A→B→C
+            # are correctly collapsed into one cluster.
+            # ─────────────────────────────────────────────────────────────
+            changed = True
+            while changed:
+                changed = False
+                # Recompute centroid of current cluster
+                c_lats = [kitchen_coords[n][0] for n in cluster_kitchens if n in kitchen_coords]
+                c_lons = [kitchen_coords[n][1] for n in cluster_kitchens if n in kitchen_coords]
+                c_lat = sum(c_lats) / len(c_lats) if c_lats else k_lat
+                c_lon = sum(c_lons) / len(c_lons) if c_lons else k_lon
+
+                for other_name, other_coords in kitchen_coords.items():
+                    if other_name in processed_kitchens:
+                        continue
+                    other_lat, other_lon = other_coords
+                    dist = self.haversine(c_lat, c_lon, other_lat, other_lon)
+                    if dist <= max_cluster_radius_km:
+                        cluster_kitchens.append(other_name)
+                        cluster_orders.extend(kitchen_groups[other_name])
+                        processed_kitchens.add(other_name)
+                        changed = True  # centroid shifted — re-scan remaining
+
+            # Calculate cluster centroid
+            lats = []
+            lons = []
+            for order in cluster_orders:
+                lat, lon = self.get_kitchen_location(order)
+                if lat != 0:
+                    lats.append(lat)
+                    lons.append(lon)
+
+            if lats:
+                centroid_lat = sum(lats) / len(lats)
+                centroid_lon = sum(lons) / len(lons)
+            else:
+                centroid_lat, centroid_lon = k_lat, k_lon
+
+            # ENHANCED: Add geohash encoding features
+            spatial_features = self._encode_location_features(
+                centroid_lat, centroid_lon
+            )
+
+            clusters.append(
+                {
+                    "centroid": (centroid_lat, centroid_lon),
+                    "orders": cluster_orders,
+                    "kitchen_names": set(cluster_kitchens),
+                    "total_orders": len(cluster_orders),
+                    # NEW: Geohash encoding features
+                    "geohash_6": spatial_features["geohash_6"],
+                    "geohash_7": spatial_features["geohash_7"],
+                    "geohash_4": spatial_features["geohash_4"],
+                    "zone_info": spatial_features["zone_info"],
+                    "spatial_features": spatial_features,
+                }
+            )
+
+        # Sort clusters by order count (largest first)
+        clusters.sort(key=lambda x: x["total_orders"], reverse=True)
+
+        logger.info(
+            f"Created {len(clusters)} clusters from {len(kitchen_groups)} kitchens with geohash encoding"
+        )
+        return clusters
+
+    def _get_kitchen_name(self, order: Dict[str, Any]) -> str:
+        """Extract kitchen name from order."""
+        possible_keys = [
+            "pickupcustomer",   # confirmed primary field in production orders
+            "locationname",     # confirmed backup field in production orders
+            "storename", "store_name",
+            "restaurantname", "restaurant_name",
+            "kitchenname", "kitchen_name",
+            "partnername", "partner_name",
+            "tenantname",
+        ]
+        for key in possible_keys:
+            if key in order and order[key]:
+                return str(order[key]).strip()
+        return "Unknown"

app/services/routing/delivery_history_service.py

@@ -0,0 +1,864 @@
+"""
+Delivery History Service — Empirical ETA from ground truth
+==========================================================
+
+WHY THIS EXISTS
+---------------
+Until now every ETA in the system was a *formula guess*
+(`RealisticETACalculator`: distance / configured_speed + fixed buffers) that was
+never checked against what actually happened in the field. Senior feedback was
+that manual riders deliver faster than our estimates — i.e. the guess is wrong.
+
+This service closes the loop. The external `nearledb` Postgres already records,
+for every completed delivery, when the rider picked up (`pickuptime`) and when
+the order was delivered (`deliverytime`). From those we reconstruct **actual
+per-leg travel times** and learn empirical medians, so ETAs reflect reality
+instead of a hand-tuned formula.
+
+GROUND-TRUTH RECONSTRUCTION (per-leg, matches the optimizer)
+-----------------------------------------------------------
+The optimizer feeds *aerial* (straight-line) leg distance to `calculate_eta`
+(see route_optimizer.py — `step_dist` comes from `aerial_matrix`, no road
+factor). To learn a model that plugs in behind the same call, we reconstruct
+observations the same way:
+
+  * Group a rider's completed deliveries by day, ordered by `deliverytime`.
+  * For each pair of consecutive deliveries:
+        leg_min = deliverytime[i] - deliverytime[i-1]      (real door-to-door time)
+        leg_km  = aerial haversine(drop[i-1], drop[i])     (same metric as optimizer)
+  * The first delivery of a group is the kitchen→first-drop leg:
+        leg_min = deliverytime[0] - pickuptime[0]          (distance unknown w/o
+        kitchen coords, so it only feeds the non-distance keys).
+
+Only `droplat/droplon`, `pickuptime`, `deliverytime`, `pickupcustomer`, `userid`
+are needed — all confirmed-present columns (same set batch_analytics reads).
+
+AGGREGATION + LOOKUP (hierarchical, cold-start safe)
+----------------------------------------------------
+Each observation feeds several keys at decreasing specificity. At prediction
+time we walk the same hierarchy and use the first key with enough samples,
+falling back to the formula when history is too thin:
+
+    kzd  kitchen | drop_zone | traffic | dist_bucket   (most specific)
+    kz   kitchen | drop_zone | traffic
+    zd   drop_zone | traffic | dist_bucket
+    z    drop_zone | traffic
+    dt   dist_bucket | traffic
+    t    traffic                                        (least specific)
+    →    RealisticETACalculator formula                 (no data)
+
+`rkz` (rider | kitchen | zone | traffic) is also stored for Phase-2 learned
+rider affinity; it is not used by the default lookup yet.
+"""
+
+import logging
+import math
+import os
+import sqlite3
+import threading
+import statistics
+import time
+from datetime import datetime, timedelta
+from collections import defaultdict
+from typing import Any, Dict, List, Optional, Tuple
+
+from app.services.routing.zone_service import ZoneService
+
+logger = logging.getLogger(__name__)
+
+_DB_PATH = os.getenv("ML_DB_PATH", "ml_data/ml_store.db")
+_WRITE_LOCK = threading.Lock()
+
+# Aerial leg-distance buckets (km) — must match the metric the optimizer feeds
+# to calculate_eta (pure haversine, no road factor).
+_DIST_BUCKETS: List[Tuple[float, float]] = [
+    (0.0, 1.0), (1.0, 2.0), (2.0, 3.0), (3.0, 5.0), (5.0, 8.0), (8.0, 12.0), (12.0, 1e9)
+]
+
+# Sanity filters for reconstructed legs.
+_MAX_LEG_MIN = 60.0    # gaps longer than this are batch boundaries / idle, not a leg
+_MAX_LEG_KM = 40.0     # implausible single hop
+
+
+# ---------------------------------------------------------------------------
+# Pure helpers
+# ---------------------------------------------------------------------------
+
+def dist_bucket(km: Optional[float]) -> Optional[str]:
+    """Aerial leg distance -> discrete bucket label, or None if unknown."""
+    if km is None:
+        return None
+    for lo, hi in _DIST_BUCKETS:
+        if km <= hi:
+            return f"{lo:g}-{hi:g}" if hi < 1e9 else f"{lo:g}+"
+    return f"{_DIST_BUCKETS[-1][0]:g}+"
+
+
+def hour_to_traffic(hour: int) -> str:
+    """Time-of-day -> traffic category, mirroring get_time_of_day_category()."""
+    if (8 <= hour < 10) or (12 <= hour < 14) or (17 <= hour < 20):
+        return "peak"
+    if hour < 7 or hour >= 22:
+        return "light"
+    return "normal"
+
+
+def normalize_kitchen(name: Any) -> str:
+    """Normalize a kitchen / pickup-customer name to a stable key token."""
+    if not name:
+        return "?"
+    return " ".join(str(name).strip().lower().split())
+
+
+def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
+    """Great-circle (aerial) distance in km."""
+    try:
+        rlat1, rlon1, rlat2, rlon2 = map(math.radians, (lat1, lon1, lat2, lon2))
+        dlat = rlat2 - rlat1
+        dlon = rlon2 - rlon1
+        a = math.sin(dlat / 2) ** 2 + math.cos(rlat1) * math.cos(rlat2) * math.sin(dlon / 2) ** 2
+        return 6371.0 * 2 * math.asin(min(1.0, math.sqrt(a)))
+    except Exception:
+        return 0.0
+
+
+def _parse_dt(val: Any) -> Optional[datetime]:
+    if val in (None, "", 0):
+        return None
+    if isinstance(val, datetime):
+        return val
+    s = str(val).strip()
+    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S.%f",
+                "%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%d %H:%M"):
+        try:
+            return datetime.strptime(s.split("+")[0].strip(), fmt)
+        except Exception:
+            continue
+    try:
+        from dateutil.parser import parse as _du
+        return _du(s)
+    except Exception:
+        return None
+
+
+# ---------------------------------------------------------------------------
+# Shared Postgres connector (single source of truth for nearledb creds)
+# ---------------------------------------------------------------------------
+
+def connect_nearledb(connect_timeout: int = 10):
+    """
+    Open a connection to the external nearledb Postgres.
+
+    Credentials come from DB_* env vars (same defaults batch_analytics used).
+    Raises on failure — callers convert to their own error type.
+    """
+    import psycopg2  # imported lazily so the app boots without it
+
+    conn = psycopg2.connect(
+        host=os.getenv("DB_HOST", "66.116.207.225"),
+        port=int(os.getenv("DB_PORT", "6432")),
+        dbname=os.getenv("DB_NAME", "nearledb"),
+        user=os.getenv("DB_USER", "admin"),
+        password=os.getenv("DB_PASSWORD", "Package@123#"),
+        connect_timeout=connect_timeout,
+    )
+    # Best-effort read-only session. (Port 6432 is pgbouncer, which rejects the
+    # `options` startup param, so we set it post-connect; our code only issues
+    # SELECTs regardless — we never write to or alter the source DB.)
+    try:
+        conn.set_session(readonly=True)
+    except Exception:
+        pass
+    return conn
+
+
+# ---------------------------------------------------------------------------
+# Key construction (build-side and lookup-side share this so they stay in sync)
+# ---------------------------------------------------------------------------
+
+def _keys_for_observation(kitchen: str, zone: str, traffic: str,
+                          bucket: Optional[str], rider_id: Any) -> List[str]:
+    """All aggregation keys an observation contributes to."""
+    keys = [
+        f"kz|{kitchen}|{zone}|{traffic}",
+        f"z|{zone}|{traffic}",
+        f"t|{traffic}",
+        f"rkz|{rider_id}|{kitchen}|{zone}|{traffic}",
+    ]
+    if bucket is not None:
+        keys += [
+            f"kzd|{kitchen}|{zone}|{traffic}|{bucket}",
+            f"zd|{zone}|{traffic}|{bucket}",
+            f"dt|{bucket}|{traffic}",
+        ]
+    return keys
+
+
+def _lookup_keys(kitchen: Optional[str], zone: str, traffic: str,
+                 bucket: Optional[str]) -> List[str]:
+    """Ordered candidate keys, most specific first, for prediction-time lookup."""
+    ordered: List[str] = []
+    if kitchen and bucket is not None:
+        ordered.append(f"kzd|{kitchen}|{zone}|{traffic}|{bucket}")
+    if kitchen:
+        ordered.append(f"kz|{kitchen}|{zone}|{traffic}")
+    if bucket is not None:
+        ordered.append(f"zd|{zone}|{traffic}|{bucket}")
+    ordered.append(f"z|{zone}|{traffic}")
+    if bucket is not None:
+        ordered.append(f"dt|{bucket}|{traffic}")
+    ordered.append(f"t|{traffic}")
+    return ordered
+
+
+# ---------------------------------------------------------------------------
+# Service
+# ---------------------------------------------------------------------------
+
+class DeliveryHistoryService:
+    """Ingests completed deliveries, learns empirical leg-time medians, serves lookups."""
+
+    def __init__(self):
+        self._db_path = _DB_PATH
+        self._zone = ZoneService()
+        # cache: full_key -> (count, median_min, p75_min)
+        self._cache: Dict[str, Tuple[int, float, float]] = {}
+        self._last_refreshed: Optional[datetime] = None
+        self._last_summary: Dict[str, Any] = {}
+        self._refresh_lock = threading.Lock()
+        self._refresh_attempt_at: Optional[datetime] = None
+        self._scheduler_started = False
+        self._ensure_db()
+        self._load_cache()
+
+    # -- schema -------------------------------------------------------------
+
+    def _ensure_db(self) -> None:
+        try:
+            os.makedirs(os.path.dirname(self._db_path) or ".", exist_ok=True)
+            conn = sqlite3.connect(self._db_path)
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS delivery_history (
+                    id            INTEGER PRIMARY KEY AUTOINCREMENT,
+                    delivery_ts   TEXT,
+                    rider_id      TEXT,
+                    kitchen       TEXT,
+                    drop_zone     TEXT,
+                    traffic       TEXT,
+                    dist_bucket   TEXT,
+                    leg_km        REAL,
+                    leg_min       REAL,
+                    is_first      INTEGER DEFAULT 0
+                )
+            """)
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS eta_stats (
+                    key          TEXT PRIMARY KEY,
+                    key_type     TEXT,
+                    sample_count INTEGER,
+                    median_min   REAL,
+                    p75_min      REAL,
+                    updated_at   TEXT
+                )
+            """)
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_eta_stats_type ON eta_stats(key_type)")
+            # Local mirror of the nearledb `deliveries` rows we care about.
+            # The request path NEVER touches Postgres — only this table.
+            # Deduped by deliveryid so incremental syncs are idempotent.
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS delivery_raw (
+                    deliveryid    TEXT PRIMARY KEY,
+                    userid        TEXT,
+                    pickupcustomer TEXT,
+                    pickuptime    TEXT,
+                    deliverytime  TEXT,
+                    dlat          REAL,
+                    dlon          REAL,
+                    plat          REAL,
+                    plon          REAL
+                )
+            """)
+            # Migration for stores created before pickup coords were added.
+            for _ddl in (
+                "ALTER TABLE delivery_raw ADD COLUMN plat REAL",
+                "ALTER TABLE delivery_raw ADD COLUMN plon REAL",
+            ):
+                try:
+                    conn.execute(_ddl)
+                except Exception:
+                    pass
+            conn.execute("CREATE INDEX IF NOT EXISTS idx_raw_dtime ON delivery_raw(deliverytime)")
+            # Single-row sync watermark / bookkeeping.
+            conn.execute("""
+                CREATE TABLE IF NOT EXISTS eta_sync_state (
+                    id             INTEGER PRIMARY KEY CHECK (id = 1),
+                    last_delivery_ts TEXT,
+                    last_synced_at TEXT,
+                    total_rows     INTEGER DEFAULT 0
+                )
+            """)
+            conn.execute("INSERT OR IGNORE INTO eta_sync_state (id, total_rows) VALUES (1, 0)")
+            conn.commit()
+            conn.close()
+        except Exception as e:
+            logger.error(f"[DeliveryHistory] DB init failed: {e}")
+
+    # -- ingestion ----------------------------------------------------------
+
+    def fetch_completed_deliveries(
+        self, days: int, tenant_id: int = 916, since: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """
+        READ-ONLY pull from nearledb `deliveries`.
+
+        * `since` set      -> incremental: only rows newer than the watermark.
+        * `since` None     -> initial backfill of the last `days`.
+
+        Only the `deliveries` table is read; the session is read-only (no writes,
+        no schema changes — enforced at the connection level).
+        """
+        conn = connect_nearledb()
+        try:
+            cur = conn.cursor()
+            if since:
+                where_time = "d.deliverytime::timestamp > %s"
+                time_param = since
+            else:
+                where_time = "d.deliverytime::timestamp >= %s"
+                time_param = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S")
+            cur.execute(
+                f"""
+                SELECT
+                    d.deliveryid,
+                    d.userid,
+                    d.pickupcustomer,
+                    d.pickuptime,
+                    d.deliverytime,
+                    COALESCE(d.droplat,  d.deliverylat)  AS dlat,
+                    COALESCE(d.droplon, d.deliverylong)  AS dlon,
+                    d.pickuplat                          AS plat,
+                    d.pickuplon                          AS plon
+                FROM deliveries d
+                WHERE d.tenantid = %s
+                  AND d.deliverytime IS NOT NULL
+                  AND d.pickuptime  IS NOT NULL
+                  AND {where_time}
+                  AND COALESCE(d.droplat, d.deliverylat) IS NOT NULL
+                  AND d.userid IS NOT NULL
+                ORDER BY d.deliverytime
+                """,
+                (tenant_id, time_param),
+            )
+            cols = [c.name for c in cur.description]
+            rows = [dict(zip(cols, r)) for r in cur.fetchall()]
+            cur.close()
+            return rows
+        finally:
+            conn.close()
+
+    # -- watermark / local raw store ---------------------------------------
+
+    def _get_watermark(self) -> Optional[str]:
+        try:
+            conn = sqlite3.connect(self._db_path)
+            row = conn.execute(
+                "SELECT last_delivery_ts FROM eta_sync_state WHERE id = 1"
+            ).fetchone()
+            conn.close()
+            return row[0] if row and row[0] else None
+        except Exception:
+            return None
+
+    def sync_from_db(self, days: int = 14, tenant_id: int = 916,
+                     full: bool = False) -> Dict[str, Any]:
+        """
+        Pull NEW completed deliveries from nearledb into the local mirror.
+
+        Idempotent (INSERT OR IGNORE on deliveryid). On the first run (empty
+        watermark) or `full=True`, backfills the last `days`; afterwards only
+        rows newer than the watermark are fetched — minimal DB load.
+        """
+        watermark = None if full else self._get_watermark()
+        rows = self.fetch_completed_deliveries(days, tenant_id, since=watermark)
+
+        inserted = 0
+        max_ts = watermark
+        conn = sqlite3.connect(self._db_path)
+        try:
+            if full:
+                # A full sync truly rebuilds the local mirror (also backfills any
+                # newly-added columns that INSERT OR IGNORE would otherwise skip).
+                conn.execute("DELETE FROM delivery_raw")
+                conn.execute("UPDATE eta_sync_state SET last_delivery_ts = NULL WHERE id = 1")
+                max_ts = None
+            for r in rows:
+                dt = str(r.get("deliverytime") or "")
+                try:
+                    cur = conn.execute(
+                        "INSERT OR IGNORE INTO delivery_raw "
+                        "(deliveryid, userid, pickupcustomer, pickuptime, deliverytime, dlat, dlon, plat, plon) "
+                        "VALUES (?,?,?,?,?,?,?,?,?)",
+                        (str(r.get("deliveryid")), str(r.get("userid")),
+                         r.get("pickupcustomer"), str(r.get("pickuptime") or ""),
+                         dt, r.get("dlat"), r.get("dlon"), r.get("plat"), r.get("plon")),
+                    )
+                    inserted += cur.rowcount
+                except Exception:
+                    continue
+                if dt and (max_ts is None or dt > max_ts):
+                    max_ts = dt
+            total = conn.execute("SELECT COUNT(*) FROM delivery_raw").fetchone()[0]
+            conn.execute(
+                "UPDATE eta_sync_state SET last_delivery_ts = ?, last_synced_at = ?, total_rows = ? WHERE id = 1",
+                (max_ts, datetime.utcnow().isoformat(), total),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+        return {"fetched": len(rows), "inserted": inserted, "local_total": total,
+                "watermark": max_ts, "mode": "full" if full else "incremental"}
+
+    def _prune_raw(self, days: int) -> int:
+        """Drop local rows older than the rolling window to bound the store."""
+        cutoff = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S")
+        try:
+            conn = sqlite3.connect(self._db_path)
+            cur = conn.execute("DELETE FROM delivery_raw WHERE deliverytime < ?", (cutoff,))
+            deleted = cur.rowcount
+            conn.commit()
+            conn.close()
+            return deleted
+        except Exception:
+            return 0
+
+    def _load_raw_rows(self, days: int) -> List[Dict[str, Any]]:
+        """Read the local mirror within the rolling window (no DB hit)."""
+        cutoff = (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S")
+        conn = sqlite3.connect(self._db_path)
+        rows = conn.execute(
+            "SELECT deliveryid, userid, pickupcustomer, pickuptime, deliverytime, dlat, dlon, plat, plon "
+            "FROM delivery_raw WHERE deliverytime >= ? ORDER BY deliverytime",
+            (cutoff,),
+        ).fetchall()
+        conn.close()
+        cols = ["deliveryid", "userid", "pickupcustomer", "pickuptime", "deliverytime", "dlat", "dlon", "plat", "plon"]
+        return [dict(zip(cols, r)) for r in rows]
+
+    def sample_batches(self, days: int = 14, min_drops: int = 4, max_drops: int = 15,
+                       limit: int = 10) -> List[Dict[str, Any]]:
+        """
+        Real recent delivery batches from the local mirror for the road-sequencing
+        decision agent. Grouped by (rider, day, kitchen); origin = that kitchen's
+        pickup coords (centroid fallback). Most-recent batches first.
+        Returns [{origin, drops, rider, day, kitchen}].
+        """
+        rows = self._load_raw_rows(days)
+        groups: Dict[Tuple[str, str, str], List[Dict[str, Any]]] = defaultdict(list)
+        for r in rows:
+            dt = _parse_dt(r.get("deliverytime"))
+            if dt is None:
+                continue
+            try:
+                if not (float(r["dlat"]) and float(r["dlon"])):
+                    continue
+            except (TypeError, ValueError):
+                continue
+            key = (str(r.get("userid")), dt.strftime("%Y-%m-%d"),
+                   normalize_kitchen(r.get("pickupcustomer")))
+            groups[key].append(r)
+
+        batches: List[Dict[str, Any]] = []
+        for key in sorted(groups, key=lambda k: k[1], reverse=True):
+            items = groups[key]
+            if not (min_drops <= len(items) <= max_drops):
+                continue
+            drops = [(float(i["dlat"]), float(i["dlon"])) for i in items]
+            origin = None
+            for i in items:
+                try:
+                    pla, plo = float(i["plat"]), float(i["plon"])
+                    if pla and plo:
+                        origin = (pla, plo)
+                        break
+                except (TypeError, ValueError):
+                    continue
+            if origin is None:
+                origin = (sum(d[0] for d in drops) / len(drops),
+                          sum(d[1] for d in drops) / len(drops))
+            batches.append({"origin": origin, "drops": drops,
+                            "rider": key[0], "day": key[1], "kitchen": key[2]})
+            if len(batches) >= limit:
+                break
+        return batches
+
+    def _build_observations(self, rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
+        """Reconstruct per-leg observations from ordered delivery rows."""
+        # Group by (rider, calendar day of delivery)
+        groups: Dict[Tuple[str, str], List[Dict[str, Any]]] = defaultdict(list)
+        for r in rows:
+            dt = _parse_dt(r.get("deliverytime"))
+            pt = _parse_dt(r.get("pickuptime"))
+            if dt is None or pt is None:
+                continue
+            try:
+                dlat = float(r.get("dlat"))
+                dlon = float(r.get("dlon"))
+            except (TypeError, ValueError):
+                continue
+            if dlat == 0 or dlon == 0:
+                continue
+            groups[(str(r.get("userid")), dt.strftime("%Y-%m-%d"))].append({
+                "dt": dt, "pt": pt, "lat": dlat, "lon": dlon,
+                "kitchen": normalize_kitchen(r.get("pickupcustomer")),
+                "rider": str(r.get("userid")),
+            })
+
+        obs: List[Dict[str, Any]] = []
+        for _key, items in groups.items():
+            items.sort(key=lambda x: x["dt"])
+            prev = None
+            for i, it in enumerate(items):
+                zone = self._zone.determine_zone(it["lat"], it["lon"])
+                traffic = hour_to_traffic(it["dt"].hour)
+                if i == 0:
+                    # kitchen -> first drop; distance unknown without kitchen coords
+                    leg_min = (it["dt"] - it["pt"]).total_seconds() / 60.0
+                    leg_km = None
+                    is_first = 1
+                else:
+                    leg_min = (it["dt"] - prev["dt"]).total_seconds() / 60.0
+                    leg_km = _haversine_km(prev["lat"], prev["lon"], it["lat"], it["lon"])
+                    is_first = 0
+                prev = it
+                # filter implausible legs
+                if leg_min <= 0 or leg_min > _MAX_LEG_MIN:
+                    continue
+                if leg_km is not None and (leg_km <= 0 or leg_km > _MAX_LEG_KM):
+                    continue
+                obs.append({
+                    "delivery_ts": it["dt"].strftime("%Y-%m-%d %H:%M:%S"),
+                    "rider_id": it["rider"],
+                    "kitchen": it["kitchen"],
+                    "drop_zone": zone,
+                    "traffic": traffic,
+                    "dist_bucket": dist_bucket(leg_km),
+                    "leg_km": leg_km,
+                    "leg_min": round(leg_min, 2),
+                    "is_first": is_first,
+                })
+        return obs
+
+    @staticmethod
+    def _aggregate(observations: List[Dict[str, Any]]) -> Dict[str, Tuple[int, float, float]]:
+        """Group observations by every key and compute (count, median, p75)."""
+        buckets: Dict[str, List[float]] = defaultdict(list)
+        for o in observations:
+            for k in _keys_for_observation(
+                o["kitchen"], o["drop_zone"], o["traffic"], o["dist_bucket"], o["rider_id"]
+            ):
+                buckets[k].append(o["leg_min"])
+        stats: Dict[str, Tuple[int, float, float]] = {}
+        for k, vals in buckets.items():
+            vals.sort()
+            n = len(vals)
+            median = statistics.median(vals)
+            p75 = vals[min(n - 1, int(math.ceil(0.75 * n)) - 1)] if n else median
+            stats[k] = (n, round(median, 2), round(p75, 2))
+        return stats
+
+    def rebuild_aggregates(self, days: int = 14) -> Dict[str, Any]:
+        """
+        Rebuild empirical medians from the LOCAL mirror only (no DB hit).
+        Reconstructs per-leg observations, aggregates, persists, reloads cache.
+        """
+        rows = self._load_raw_rows(days)
+        observations = self._build_observations(rows)
+        stats = self._aggregate(observations)
+
+        now = datetime.utcnow().isoformat()
+        try:
+            conn = sqlite3.connect(self._db_path)
+            conn.execute("DELETE FROM delivery_history")
+            conn.execute("DELETE FROM eta_stats")
+            conn.executemany(
+                "INSERT INTO delivery_history "
+                "(delivery_ts, rider_id, kitchen, drop_zone, traffic, dist_bucket, leg_km, leg_min, is_first) "
+                "VALUES (?,?,?,?,?,?,?,?,?)",
+                [(o["delivery_ts"], o["rider_id"], o["kitchen"], o["drop_zone"],
+                  o["traffic"], o["dist_bucket"], o["leg_km"], o["leg_min"], o["is_first"])
+                 for o in observations],
+            )
+            conn.executemany(
+                "INSERT INTO eta_stats (key, key_type, sample_count, median_min, p75_min, updated_at) "
+                "VALUES (?,?,?,?,?,?)",
+                [(k, k.split("|", 1)[0], n, med, p75, now) for k, (n, med, p75) in stats.items()],
+            )
+            conn.commit()
+            conn.close()
+        except Exception as e:
+            logger.error(f"[DeliveryHistory] persist failed: {e}", exc_info=True)
+            return {"status": "persist_failed", "error": str(e)}
+
+        self._load_cache()
+        self._last_refreshed = datetime.utcnow()
+        n_by_type: Dict[str, int] = defaultdict(int)
+        for k in stats:
+            n_by_type[k.split("|", 1)[0]] += 1
+        summary = {
+            "status": "ok",
+            "local_rows": len(rows),
+            "observations": len(observations),
+            "stat_keys": len(stats),
+            "keys_by_type": dict(n_by_type),
+            "history_days": days,
+            "rebuilt_at": self._last_refreshed.isoformat(),
+        }
+        logger.info(
+            f"[DeliveryHistory] aggregates rebuilt: local_rows={len(rows)} "
+            f"obs={len(observations)} keys={len(stats)} ({dict(n_by_type)})"
+        )
+        return summary
+
+    def refresh_eta_stats(self, days: int = 14, tenant_id: int = 916,
+                          full: bool = False) -> Dict[str, Any]:
+        """
+        Full pipeline: incremental DB sync -> prune local store -> rebuild
+        aggregates locally. This is what the scheduler and the admin endpoint
+        call. The DB is touched only by the sync step (new rows only).
+        """
+        with _WRITE_LOCK:
+            try:
+                sync = self.sync_from_db(days=days, tenant_id=tenant_id, full=full)
+            except Exception as e:
+                logger.error(f"[DeliveryHistory] sync failed: {e}", exc_info=True)
+                # Still try to serve whatever is already local.
+                rebuilt = self.rebuild_aggregates(days)
+                self._last_summary = {"status": "sync_failed", "error": str(e), "rebuild": rebuilt}
+                return self._last_summary
+
+            self._prune_raw(days)
+            rebuilt = self.rebuild_aggregates(days)
+            self._last_summary = {"status": "ok", "sync": sync, "rebuild": rebuilt}
+            return self._last_summary
+
+    # -- cache + lookup -----------------------------------------------------
+
+    def _load_cache(self) -> None:
+        try:
+            conn = sqlite3.connect(self._db_path)
+            rows = conn.execute(
+                "SELECT key, sample_count, median_min, p75_min FROM eta_stats"
+            ).fetchall()
+            conn.close()
+            self._cache = {k: (int(n), float(med), float(p75)) for k, n, med, p75 in rows}
+            if self._cache:
+                logger.info(f"[DeliveryHistory] loaded {len(self._cache)} eta_stats keys into cache")
+        except Exception as e:
+            logger.warning(f"[DeliveryHistory] cache load failed: {e}")
+            self._cache = {}
+
+    def has_data(self) -> bool:
+        return bool(self._cache)
+
+    def maybe_background_refresh(self, days: int = 14, tenant_id: int = 916,
+                                 cooldown_s: int = 600) -> bool:
+        """
+        If the cache is empty, kick off a one-shot background refresh (at most
+        once per cooldown). Keeps the prediction hot path non-blocking — the
+        current call falls back to the formula; later calls use empirical data.
+        Returns True if a refresh thread was started.
+        """
+        if self._cache:
+            return False
+        with self._refresh_lock:
+            if self._cache:
+                return False
+            now = datetime.utcnow()
+            if (self._refresh_attempt_at is not None
+                    and (now - self._refresh_attempt_at).total_seconds() < cooldown_s):
+                return False
+            self._refresh_attempt_at = now
+
+        def _run():
+            try:
+                self.refresh_eta_stats(days, tenant_id)
+            except Exception as e:
+                logger.warning(f"[DeliveryHistory] background refresh failed: {e}")
+
+        threading.Thread(target=_run, daemon=True, name="eta-refresh").start()
+        return True
+
+    def local_count(self) -> int:
+        try:
+            conn = sqlite3.connect(self._db_path)
+            n = conn.execute("SELECT COUNT(*) FROM delivery_raw").fetchone()[0]
+            conn.close()
+            return int(n)
+        except Exception:
+            return 0
+
+    def ensure_background_sync(self, interval_hours: int = 6, days: int = 14,
+                               tenant_id: int = 916) -> bool:
+        """
+        Start the autonomous sync agent (once per process). It refreshes
+        immediately on startup (full backfill if the local store is empty,
+        otherwise incremental), then re-syncs every `interval_hours`.
+
+        Runs in a daemon thread so it never blocks the app; all request-path
+        ETA lookups read the local SQLite mirror, never Postgres.
+        """
+        with self._refresh_lock:
+            if self._scheduler_started:
+                return False
+            self._scheduler_started = True
+
+        def _loop():
+            logger.info(
+                f"[ETA-Agent] autonomous sync started — interval={interval_hours}h, window={days}d"
+            )
+            first = True
+            while True:
+                try:
+                    full = first and self.local_count() == 0
+                    result = self.refresh_eta_stats(days=days, tenant_id=tenant_id, full=full)
+                    logger.info(f"[ETA-Agent] sync cycle done: {result.get('status')}")
+                except Exception as e:
+                    logger.warning(f"[ETA-Agent] sync cycle failed (will retry): {e}")
+                # Recompute learned rider affinity from the freshly-synced mirror.
+                try:
+                    from app.services.routing.rider_affinity_service import get_rider_affinity
+                    get_rider_affinity().refresh(days=days)
+                except Exception as e:
+                    logger.debug(f"[ETA-Agent] affinity refresh skipped: {e}")
+                first = False
+                time.sleep(max(1, int(interval_hours)) * 3600)
+
+        threading.Thread(target=_loop, daemon=True, name="eta-sync-agent").start()
+        return True
+
+    def lookup(
+        self,
+        distance_km: float,
+        traffic_cat: str,
+        kitchen: Optional[str] = None,
+        drop_coords: Optional[Tuple[float, float]] = None,
+        min_samples: int = 20,
+        stat: str = "median",
+    ) -> Optional[Dict[str, Any]]:
+        """
+        Return the empirical leg time (minutes) for a context, or None if no key
+        has >= min_samples. Walks the specificity hierarchy.
+        """
+        if not self._cache:
+            return None
+        zone = "Unknown"
+        if drop_coords and drop_coords[0] and drop_coords[1]:
+            zone = self._zone.determine_zone(float(drop_coords[0]), float(drop_coords[1]))
+        bucket = dist_bucket(distance_km) if distance_km and distance_km > 0 else None
+        k_norm = normalize_kitchen(kitchen) if kitchen else None
+
+        for full_key in _lookup_keys(k_norm, zone, traffic_cat, bucket):
+            hit = self._cache.get(full_key)
+            if hit and hit[0] >= min_samples:
+                count, median, p75 = hit
+                value = p75 if stat == "p75" else median
+                return {
+                    "value_min": value,
+                    "sample_count": count,
+                    "source_key": full_key,
+                    "key_type": full_key.split("|", 1)[0],
+                }
+        return None
+
+    # -- diagnostics / backtest --------------------------------------------
+
+    def get_summary(self) -> Dict[str, Any]:
+        return {
+            "has_data": self.has_data(),
+            "cache_keys": len(self._cache),
+            "last_summary": self._last_summary,
+        }
+
+    def backtest(self, days: int = 14, tenant_id: int = 916,
+                 min_samples: int = 20, stat: str = "median") -> Dict[str, Any]:
+        """
+        Honest time-split backtest: build empirical stats on the older 80% of
+        observations, then compare formula vs empirical MAE on the most recent
+        20% (held out). Proves whether empirical beats the formula before trust.
+        """
+        from app.services.routing.realistic_eta_calculator import RealisticETACalculator
+        formula = RealisticETACalculator()
+
+        # Backtest runs entirely on the local mirror — no DB hit.
+        rows = self._load_raw_rows(days)
+        obs = self._build_observations(rows)
+        obs = [o for o in obs if o["is_first"] == 0]  # need leg_km for both predictors
+        if len(obs) < 50:
+            return {"status": "insufficient_data", "observations": len(obs)}
+
+        obs.sort(key=lambda o: o["delivery_ts"])
+        split = int(len(obs) * 0.8)
+        train, test = obs[:split], obs[split:]
+        train_stats = self._aggregate(train)
+
+        def _empirical(o) -> Optional[float]:
+            for full_key in _lookup_keys(o["kitchen"], o["drop_zone"], o["traffic"], o["dist_bucket"]):
+                hit = train_stats.get(full_key)
+                if hit and hit[0] >= min_samples:
+                    return hit[2] if stat == "p75" else hit[1]
+            return None
+
+        f_err, e_err, e_err_fallback, covered = [], [], [], 0
+        for o in test:
+            actual = o["leg_min"]
+            f_pred = formula.calculate_eta(
+                distance_km=o["leg_km"], is_first_order=False,
+                order_type="Economy", time_of_day=o["traffic"],
+            )
+            f_err.append(abs(f_pred - actual))
+            emp = _empirical(o)
+            if emp is not None:
+                covered += 1
+                e_err.append(abs(emp - actual))
+                e_err_fallback.append(abs(emp - actual))
+            else:
+                e_err_fallback.append(abs(f_pred - actual))  # fallback to formula
+
+        def _mae(xs):
+            return round(sum(xs) / len(xs), 2) if xs else None
+
+        return {
+            "status": "ok",
+            "history_days": days,
+            "observations": len(obs),
+            "test_size": len(test),
+            "empirical_coverage_pct": round(100.0 * covered / len(test), 1),
+            "formula_mae_min": _mae(f_err),
+            "empirical_mae_min_covered": _mae(e_err),
+            "empirical_mae_min_with_fallback": _mae(e_err_fallback),
+            "min_samples": min_samples,
+            "stat": stat,
+            "interpretation": (
+                "empirical_better"
+                if (_mae(e_err_fallback) is not None and _mae(f_err) is not None
+                    and _mae(e_err_fallback) < _mae(f_err))
+                else "no_improvement"
+            ),
+        }
+
+
+# ---------------------------------------------------------------------------
+# Singleton
+# ---------------------------------------------------------------------------
+
+_service: Optional[DeliveryHistoryService] = None
+_service_lock = threading.Lock()
+
+
+def get_delivery_history_service() -> DeliveryHistoryService:
+    """Get (and lazily build) the DeliveryHistoryService singleton."""
+    global _service
+    with _service_lock:
+        if _service is None:
+            _service = DeliveryHistoryService()
+        return _service

app/services/routing/empirical_eta_calculator.py

@@ -0,0 +1,143 @@
+"""
+Empirical ETA Calculator
+========================
+
+Drop-in replacement for `RealisticETACalculator` that prefers ETAs *learned
+from actual delivery times* (see delivery_history_service.py) and falls back to
+the original formula whenever history is too thin.
+
+Design goals
+------------
+* **Interface-compatible**: `calculate_eta(...)` keeps the same positional args
+  and `int`-minutes return as the formula calculator, so existing call sites
+  work unchanged. Extra args (`kitchen`, `drop_coords`, `rider_id`) are optional
+  and let callers that *have* that context (the optimizer does) get sharper,
+  zone-aware estimates.
+* **Safe by default**: if empirical data is missing for a context, or the
+  feature is disabled via `eta_empirical_enabled`, it returns exactly what the
+  formula would — zero behavior change until real data exists.
+* **Non-blocking**: never calls Postgres on the hot path. If the stats cache is
+  empty it kicks off a one-shot background refresh and serves the formula
+  meanwhile.
+
+Empirical values are real door-to-door leg times (gap between consecutive
+deliveries), so they already include travel + drop service time. We only add the
+kitchen pickup buffer for the first leg, mirroring the formula's semantics.
+"""
+
+import logging
+from typing import Any, List, Optional, Tuple
+
+from app.services.routing.realistic_eta_calculator import (
+    RealisticETACalculator,
+    get_time_of_day_category,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class EmpiricalETACalculator:
+    """ETA calculator backed by empirical history, with formula fallback."""
+
+    def __init__(self):
+        # Composed formula calculator — the fallback and the source of buffers.
+        self.formula = RealisticETACalculator()
+
+    # ------------------------------------------------------------------
+    # Main entry point (signature is a superset of RealisticETACalculator)
+    # ------------------------------------------------------------------
+
+    def calculate_eta(
+        self,
+        distance_km: float,
+        is_first_order: bool = False,
+        order_type: str = "Economy",
+        time_of_day: str = "peak",
+        kitchen: Optional[str] = None,
+        drop_coords: Optional[Tuple[float, float]] = None,
+        rider_id: Optional[Any] = None,
+    ) -> int:
+        """Return ETA in minutes — empirical if available, else the formula."""
+        if distance_km is not None and distance_km <= 0 and not is_first_order:
+            return 0
+
+        from app.config.dynamic_config import get_config
+        cfg = get_config()
+
+        if not bool(cfg.get("eta_empirical_enabled", True)):
+            return self._formula_eta(distance_km, is_first_order, order_type, time_of_day)
+
+        try:
+            from app.services.routing.delivery_history_service import (
+                get_delivery_history_service,
+            )
+            svc = get_delivery_history_service()
+
+            if not svc.has_data():
+                # Populate in the background; serve the formula for now.
+                svc.maybe_background_refresh(
+                    days=int(cfg.get("eta_history_days", 14)),
+                )
+                return self._formula_eta(distance_km, is_first_order, order_type, time_of_day)
+
+            hit = svc.lookup(
+                distance_km=float(distance_km or 0.0),
+                traffic_cat=time_of_day,
+                kitchen=kitchen,
+                drop_coords=drop_coords,
+                min_samples=int(cfg.get("eta_min_samples", 20)),
+                stat=str(cfg.get("eta_stat", "median")),
+            )
+        except Exception as e:
+            logger.debug(f"[EmpiricalETA] lookup failed, using formula: {e}")
+            hit = None
+
+        if not hit:
+            return self._formula_eta(distance_km, is_first_order, order_type, time_of_day)
+
+        value = float(hit["value_min"])
+        # First leg includes time spent picking up at the kitchen; the empirical
+        # leg gap starts at pickup completion, so add the same buffer the formula uses.
+        if is_first_order:
+            value += float(cfg.get("eta_pickup_time_min", 3.0))
+        return int(value) + 1  # round up for safety, matching the formula
+
+    def _formula_eta(self, distance_km, is_first_order, order_type, time_of_day) -> int:
+        return self.formula.calculate_eta(
+            distance_km=distance_km,
+            is_first_order=is_first_order,
+            order_type=order_type,
+            time_of_day=time_of_day,
+        )
+
+    # ------------------------------------------------------------------
+    # Batch helper (kept compatible with RealisticETACalculator)
+    # ------------------------------------------------------------------
+
+    def calculate_batch_eta(self, orders: List[dict]) -> List[dict]:
+        """Calculate ETAs for a batch in sequence (formula-parity batch path)."""
+        traffic = get_time_of_day_category()
+        for order in orders:
+            distance_km = float(order.get("previouskms", 0) or 0)
+            step = order.get("step", 1)
+            order_type = order.get("ordertype", "Economy")
+            drop = None
+            try:
+                dlat = float(order.get("deliverylat") or order.get("droplat") or 0)
+                dlon = float(order.get("deliverylong") or order.get("droplon") or 0)
+                if dlat and dlon:
+                    drop = (dlat, dlon)
+            except (TypeError, ValueError):
+                drop = None
+            eta = self.calculate_eta(
+                distance_km=distance_km,
+                is_first_order=(step == 1),
+                order_type=order_type,
+                time_of_day=traffic,
+                kitchen=order.get("pickupcustomer") or order.get("locationname"),
+                drop_coords=drop,
+                rider_id=order.get("userid"),
+            )
+            order["eta"] = str(eta)
+            order["eta_empirical"] = True
+        return orders

app/services/routing/kalman_filter.py

@@ -0,0 +1,327 @@
+"""
+GPS Kalman Filter \u2014 rider-api
+
+A 1D Kalman filter applied independently to latitude and longitude
+to smooth noisy GPS coordinates from riders and delivery points.
+
+Why Kalman for GPS?
+- GPS readings contain measurement noise (\u00b15\u201315m typical, \u00b150m poor signal)
+- Rider location pings can "jump" due to bad signal or device error
+- Kalman filter gives an optimal estimate by balancing:
+    (1) Previous predicted position (process model)
+    (2) New GPS measurement (observation model)
+
+Design:
+- Separate filter instance per rider (stateful \u2014 preserves history)
+- `CoordinateKalmanFilter` \u2014 single lat/lon smoother
+- `GPSKalmanFilter`          \u2014 wraps two CoordinateKalmanFilters (lat + lon)
+- `RiderKalmanRegistry`      \u2014 manages per-rider filter instances
+- `smooth_coordinates()`     \u2014 stateless single-shot smoother for delivery coords
+
+Usage:
+    # Stateless (one-shot, no history \u2014 for delivery coords):
+    smooth_lat, smooth_lon = smooth_coordinates(raw_lat, raw_lon)
+
+    # Stateful (per-rider, preserves motion history):
+    registry = RiderKalmanRegistry()
+    lat, lon = registry.update(rider_id=1116, lat=11.0067, lon=76.9558)
+"""
+
+import logging
+import time
+from typing import Dict, Optional, Tuple
+
+logger = logging.getLogger(__name__)
+
+
+# \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500
+# CORE 1D KALMAN FILTER
+# \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500
+
+class CoordinateKalmanFilter:
+    """
+    1-dimensional Kalman filter for a single GPS coordinate (lat or lon).
+
+    State model: position only (constant position with random walk).
+
+    Equations:
+        Prediction:   x\u0302\u2096\u207b  = x\u0302\u2096\u208b\u2081          (no movement assumed between pings)
+                      P\u0302\u2096\u207b   = P\u2096\u208b\u2081 + Q       (uncertainty grows over time)
+
+        Update:       K\u2096    = P\u0302\u2096\u207b / (P\u0302\u2096\u207b + R)   (Kalman gain)
+                      x\u0302\u2096    = x\u0302\u2096\u207b + K\u2096\u00b7(z\u2096 - x\u0302\u2096\u207b)  (weighted fusion)
+                      P\u2096    = (1 - K\u2096)\u00b7P\u0302\u2096\u207b          (update uncertainty)
+
+    Parameters:
+        process_noise (Q): How much position can change between measurements.
+                           Higher = filter trusts new measurements more (less smoothing).
+        measurement_noise (R): GPS measurement uncertainty.
+                               Higher = filter trusts history more (more smoothing).
+    """
+
+    def __init__(
+        self,
+        process_noise: float = 1e-4,
+        measurement_noise: float = 0.01,
+        initial_uncertainty: float = 1.0,
+    ):
+        self.Q = process_noise
+        self.R = measurement_noise
+        self._x: Optional[float] = None
+        self._P: float = initial_uncertainty
+
+    @property
+    def initialized(self) -> bool:
+        return self._x is not None
+
+    def update(self, measurement: float) -> float:
+        """Process one new measurement and return the filtered estimate."""
+        if not self.initialized:
+            self._x = measurement
+            return self._x
+
+        # Predict
+        x_prior = self._x
+        P_prior = self._P + self.Q
+
+        # Update
+        K = P_prior / (P_prior + self.R)
+        self._x = x_prior + K * (measurement - x_prior)
+        self._P = (1.0 - K) * P_prior
+
+        return self._x
+
+    def reset(self):
+        self._x = None
+        self._P = 1.0
+
+
+# \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500
+# 2D GPS KALMAN FILTER (lat + lon)
+# \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500
+
+class GPSKalmanFilter:
+    """
+    Two-dimensional GPS smoother using independent 1D Kalman filters
+    for latitude and longitude.
+    """
+
+    def __init__(
+        self,
+        process_noise: float = 1e-4,
+        measurement_noise: float = 0.01,
+    ):
+        self.lat_filter = CoordinateKalmanFilter(process_noise, measurement_noise)
+        self.lon_filter = CoordinateKalmanFilter(process_noise, measurement_noise)
+        self.last_updated: float = time.time()
+        self.update_count: int = 0
+
+    def update(self, lat: float, lon: float) -> Tuple[float, float]:
+        """Feed a new GPS reading and get the smoothed (lat, lon)."""
+        if not self._is_valid_coord(lat, lon):
+            if self.lat_filter.initialized:
+                return self.lat_filter._x, self.lon_filter._x
+            return lat, lon
+
+        smooth_lat = self.lat_filter.update(lat)
+        smooth_lon = self.lon_filter.update(lon)
+        self.last_updated = time.time()
+        self.update_count += 1
+
+        return smooth_lat, smooth_lon
+
+    def get_estimate(self) -> Optional[Tuple[float, float]]:
+        if self.lat_filter.initialized:
+            return self.lat_filter._x, self.lon_filter._x
+        return None
+
+    def reset(self):
+        self.lat_filter.reset()
+        self.lon_filter.reset()
+        self.update_count = 0
+
+    @staticmethod
+    def _is_valid_coord(lat: float, lon: float) -> bool:
+        try:
+            lat, lon = float(lat), float(lon)
+            return (
+                -90.0 <= lat <= 90.0
+                and -180.0 <= lon <= 180.0
+                and not (lat == 0.0 and lon == 0.0)
+            )
+        except (TypeError, ValueError):
+            return False
+
+
+# \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500
+# PER-RIDER FILTER REGISTRY
+# \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500
+
+class RiderKalmanRegistry:
+    """
+    Maintains per-rider Kalman filter instances across calls.
+    Stale filters (> 30 min silence) are automatically reset.
+    """
+
+    def __init__(
+        self,
+        process_noise: float = 1e-4,
+        measurement_noise: float = 0.01,
+        stale_seconds: float = 1800.0,
+    ):
+        self._filters: Dict[str, GPSKalmanFilter] = {}
+        self._process_noise = process_noise
+        self._measurement_noise = measurement_noise
+        self._stale_seconds = stale_seconds
+
+    def _get_or_create(self, rider_id) -> GPSKalmanFilter:
+        key = str(rider_id)
+        now = time.time()
+        if key in self._filters:
+            f = self._filters[key]
+            if now - f.last_updated > self._stale_seconds:
+                f.reset()
+            return f
+        self._filters[key] = GPSKalmanFilter(
+            process_noise=self._process_noise,
+            measurement_noise=self._measurement_noise,
+        )
+        return self._filters[key]
+
+    def update(self, rider_id, lat: float, lon: float) -> Tuple[float, float]:
+        return self._get_or_create(rider_id).update(lat, lon)
+
+    def get_estimate(self, rider_id) -> Optional[Tuple[float, float]]:
+        key = str(rider_id)
+        if key in self._filters:
+            return self._filters[key].get_estimate()
+        return None
+
+    def reset_rider(self, rider_id):
+        key = str(rider_id)
+        if key in self._filters:
+            self._filters[key].reset()
+
+    def clear_all(self):
+        self._filters.clear()
+
+
+# \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500
+# GLOBAL REGISTRY (process-level singleton)
+# \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500
+
+_global_registry = RiderKalmanRegistry()
+
+
+def get_registry() -> RiderKalmanRegistry:
+    """Get the process-level rider Kalman filter registry."""
+    return _global_registry
+
+
+# \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500
+# STATELESS COORDINATE SMOOTHER
+# \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500
+
+def smooth_coordinates(
+    lat: float,
+    lon: float,
+    *,
+    prior_lat: Optional[float] = None,
+    prior_lon: Optional[float] = None,
+    process_noise: float = 1e-4,
+    measurement_noise: float = 0.01,
+) -> Tuple[float, float]:
+    """
+    Stateless single-shot GPS smoother.
+    If a prior is provided, blends the new reading towards it.
+    """
+    f = GPSKalmanFilter(process_noise=process_noise, measurement_noise=measurement_noise)
+    if prior_lat is not None and prior_lon is not None:
+        try:
+            _flat = float(prior_lat)
+            _flon = float(prior_lon)
+            if GPSKalmanFilter._is_valid_coord(_flat, _flon):
+                f.update(_flat, _flon)
+        except (TypeError, ValueError):
+            pass
+    return f.update(lat, lon)
+
+
+# \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500
+# BATCH SMOOTHERS
+# \u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500\u2500
+
+def smooth_rider_locations(riders: list) -> list:
+    """
+    Apply Kalman smoothing to a list of rider dicts in-place using
+    the global per-rider registry (history preserved across calls).
+
+    Reads/writes: latitude, longitude (and currentlat/currentlong if present).
+    Adds: _kalman_smoothed = True on each processed rider.
+    """
+    registry = get_registry()
+    for rider in riders:
+        try:
+            rider_id = (
+                rider.get("userid") or rider.get("riderid") or
+                rider.get("id") or "unknown"
+            )
+            raw_lat = float(rider.get("latitude") or rider.get("currentlat") or 0)
+            raw_lon = float(rider.get("longitude") or rider.get("currentlong") or 0)
+            if raw_lat == 0.0 and raw_lon == 0.0:
+                continue
+            smooth_lat, smooth_lon = registry.update(rider_id, raw_lat, raw_lon)
+            # Cast back to string for Go compatibility
+            s_lat, s_lon = str(round(smooth_lat, 8)), str(round(smooth_lon, 8))
+            rider["latitude"] = s_lat
+            rider["longitude"] = s_lon
+            if "currentlat" in rider:
+                rider["currentlat"] = s_lat
+            if "currentlong" in rider:
+                rider["currentlong"] = s_lon
+            rider["_kalman_smoothed"] = True
+        except Exception as e:
+            logger.debug(f"Kalman rider smoothing skipped: {e}")
+    return riders
+
+
+def smooth_order_coordinates(orders: list) -> list:
+    """
+    Validate and lightly normalise delivery coordinates in a list of order dicts.
+
+    DESIGN NOTE — why we do NOT use Kalman filtering here:
+    ─────────────────────────────────────────────────────
+    Kalman filtering is a *temporal* smoother: it blends successive measurements
+    from the same sensor over time. Delivery coordinates are a single static
+    point (one measurement). Feeding the kitchen location as a "prior" would
+    pull the customer's address toward the kitchen — exactly wrong.
+
+    Per-customer GPS accuracy is handled upstream by the FAISS coordinate store
+    (verified historical rider-confirmed delivery points). This function only
+    normalises the coordinate fields to floats so downstream code never sees
+    raw strings or None values.
+
+    Modifies orders in-place. Returns the same list.
+    """
+    for order in orders:
+        try:
+            dlat_raw = order.get("deliverylat") or order.get("droplat")
+            dlon_raw = order.get("deliverylong") or order.get("droplon")
+            if dlat_raw is None or dlon_raw is None:
+                continue
+            dlat = float(dlat_raw)
+            dlon = float(dlon_raw)
+            if not GPSKalmanFilter._is_valid_coord(dlat, dlon):
+                continue
+            # Normalise to string (Go service expects string coordinates)
+            s_lat = str(round(dlat, 8))
+            s_lon = str(round(dlon, 8))
+            order["deliverylat"] = s_lat
+            order["deliverylong"] = s_lon
+            if "droplat" in order:
+                order["droplat"] = s_lat
+            if "droplon" in order:
+                order["droplon"] = s_lon
+        except Exception as e:
+            logger.debug(f"Coordinate normalisation skipped: {e}")
+    return orders

app/services/routing/realistic_eta_calculator.py

@@ -0,0 +1,127 @@
+"""
+Realistic ETA Calculator for Delivery Operations
+
+Accounts for:
+- City traffic conditions
+- Stop time at pickup/delivery
+- Navigation time
+- Parking/finding address time
+- Different speeds for different order types
+"""
+
+import logging
+from typing import Dict, Any
+
+logger = logging.getLogger(__name__)
+
+
+class RealisticETACalculator:
+    """
+    Calculates realistic ETAs accounting for real-world delivery conditions.
+    """
+    
+    def __init__(self):
+        from app.config.dynamic_config import get_config
+        cfg = get_config()
+
+        # BASE SPEED (km/h) - Driven by the DB configuration
+        base_speed = cfg.get("avg_speed_kmh", 18.0)
+        
+        # REALISTIC SPEEDS based on time of day
+        self.CITY_SPEED_HEAVY_TRAFFIC = base_speed * 0.7  # Usually ~12 km/h
+        self.CITY_SPEED_MODERATE      = base_speed        # Usually ~18 km/h
+        self.CITY_SPEED_LIGHT         = base_speed * 1.2  # Usually ~21.6 km/h
+        
+        # TIME BUFFERS (minutes)
+        self.PICKUP_TIME = cfg.get("eta_pickup_time_min", 3.0)
+        self.DELIVERY_TIME = cfg.get("eta_delivery_time_min", 4.0)
+        self.NAVIGATION_BUFFER = cfg.get("eta_navigation_buffer_min", 1.5)
+        
+        # DISTANCE-BASED SPEED SELECTION
+        # Short distances (<2km) are slower due to more stops/starts
+        # Long distances (>8km) might have highway portions
+        self.SHORT_TRIP_FACTOR = cfg.get("eta_short_trip_factor", 0.8)
+        self.LONG_TRIP_FACTOR  = cfg.get("eta_long_trip_factor", 1.1)
+        
+    def calculate_eta(self, 
+                     distance_km: float, 
+                     is_first_order: bool = False,
+                     order_type: str = "Economy",
+                     time_of_day: str = "peak") -> int:
+        """
+        Calculate realistic ETA in minutes.
+        
+        Args:
+            distance_km: Distance to travel in kilometers
+            is_first_order: If True, includes pickup time
+            order_type: "Economy", "Premium", or "Risky"
+            time_of_day: "peak", "normal", or "light" traffic
+            
+        Returns:
+            ETA in minutes (rounded up for safety)
+        """
+        
+        if distance_km <= 0:
+            return 0
+        
+        # 1. SELECT SPEED BASED ON CONDITIONS
+        if time_of_day == "peak":
+            base_speed = self.CITY_SPEED_HEAVY_TRAFFIC
+        elif time_of_day == "light":
+            base_speed = self.CITY_SPEED_LIGHT
+        else:
+            base_speed = self.CITY_SPEED_MODERATE
+        
+        # 2. ADJUST SPEED BASED ON DISTANCE
+        # Short trips are slower (more intersections, traffic lights)
+        if distance_km < 2.0:
+            effective_speed = base_speed * self.SHORT_TRIP_FACTOR
+        elif distance_km > 8.0:
+            effective_speed = base_speed * self.LONG_TRIP_FACTOR
+        else:
+            effective_speed = base_speed
+        
+        # 3. CALCULATE TRAVEL TIME
+        travel_time = (distance_km / effective_speed) * 60  # Convert to minutes
+        
+        # 4. ADD BUFFERS
+        total_time = travel_time
+        
+        # Pickup time (only for first order in sequence)
+        if is_first_order:
+            total_time += self.PICKUP_TIME
+        
+        # Delivery time (always)
+        total_time += self.DELIVERY_TIME
+        
+        # Navigation buffer (proportional to distance)
+        if distance_km > 3.0:
+            total_time += self.NAVIGATION_BUFFER
+        
+        # 5. SAFETY MARGIN (Round up to next minute)
+        # Riders prefer to arrive early than late
+        eta_minutes = int(total_time) + 1
+        
+        return eta_minutes
+    
+
+def get_time_of_day_category() -> str:
+    """
+    Determine current traffic conditions based on time.
+    
+    Returns:
+        "peak", "normal", or "light"
+    """
+    from datetime import datetime
+    
+    
+    current_hour = datetime.now().hour
+    
+    # Peak hours: 8-10 AM, 12-2 PM, 5-8 PM
+    if (8 <= current_hour < 10) or (12 <= current_hour < 14) or (17 <= current_hour < 20):
+        return "peak"
+    # Light traffic: Late night/early morning
+    elif current_hour < 7 or current_hour >= 22:
+        return "light"
+    else:
+        return "normal"

app/services/routing/rider_affinity_service.py

@@ -0,0 +1,171 @@
+"""
+Rider Affinity Service (learned, agentic)
+=========================================
+
+Learns each rider's real kitchen affinity and operating area from the local
+delivery mirror (`delivery_raw`) and exposes them MERGED with the curated config
+in `app/config/rider_preferences.py`:
+
+  * get_preferred_kitchens() = curated RIDER_PREFERRED_KITCHENS  ∪  learned
+    (rider served a kitchen >= `rider_affinity_min_deliveries` times). Union only —
+    never drops a curated owner.
+  * get_home_locations()     = curated RIDER_HOME_LOCATIONS, with a learned drop
+    centroid filled in ONLY for riders missing from the config.
+
+These feed SOFT steering (preference discount / home bonus / distance bypass) in
+the optimizer + assignment service. HARD kitchen locks and BLOCKED_RIDERS remain
+sourced from the curated config — learned data can only broaden preference, never
+change who is *eligible* for a kitchen. Recomputed by the autonomous agent loop.
+"""
+
+import logging
+import threading
+from collections import defaultdict
+from datetime import datetime
+from typing import Any, Dict, List, Optional, Tuple
+
+from app.config.rider_preferences import RIDER_PREFERRED_KITCHENS, RIDER_HOME_LOCATIONS
+
+logger = logging.getLogger(__name__)
+
+
+def _to_int_rid(v: Any) -> Optional[int]:
+    try:
+        return int(v)
+    except (TypeError, ValueError):
+        return None
+
+
+class RiderAffinityService:
+    def __init__(self):
+        self._learned_pref: Dict[int, List[str]] = {}
+        self._learned_home: Dict[int, Tuple[float, float]] = {}
+        self._last_refreshed: Optional[datetime] = None
+        self._lock = threading.Lock()
+
+    # ------------------------------------------------------------------
+
+    def refresh(self, days: Optional[int] = None) -> Dict[str, Any]:
+        """Recompute learned affinity from the local delivery mirror (no DB hit)."""
+        from app.config.dynamic_config import get_config
+        from app.services.routing.delivery_history_service import (
+            get_delivery_history_service, normalize_kitchen,
+        )
+        cfg = get_config()
+        window = int(days if days is not None else cfg.get("eta_history_days", 14))
+        min_n = int(cfg.get("rider_affinity_min_deliveries", 10))
+
+        rows = get_delivery_history_service()._load_raw_rows(window)
+        counts: Dict[int, Dict[str, int]] = defaultdict(lambda: defaultdict(int))
+        coords: Dict[int, List[Tuple[float, float]]] = defaultdict(list)
+        for r in rows:
+            rid = _to_int_rid(r.get("userid"))
+            if rid is None:
+                continue
+            kitchen = normalize_kitchen(r.get("pickupcustomer"))
+            if kitchen and kitchen != "?":
+                counts[rid][kitchen] += 1
+            try:
+                la, lo = float(r.get("dlat")), float(r.get("dlon"))
+                if la and lo:
+                    coords[rid].append((la, lo))
+            except (TypeError, ValueError):
+                continue
+
+        learned_pref = {
+            rid: [k for k, n in kc.items() if n >= min_n]
+            for rid, kc in counts.items()
+        }
+        learned_pref = {rid: ks for rid, ks in learned_pref.items() if ks}
+        learned_home = {
+            rid: (round(sum(p[0] for p in pts) / len(pts), 6),
+                  round(sum(p[1] for p in pts) / len(pts), 6))
+            for rid, pts in coords.items() if pts
+        }
+
+        with self._lock:
+            self._learned_pref = learned_pref
+            self._learned_home = learned_home
+            self._last_refreshed = datetime.utcnow()
+
+        summary = {
+            "status": "ok",
+            "riders_with_learned_kitchens": len(learned_pref),
+            "riders_with_learned_home": len(learned_home),
+            "min_deliveries": min_n,
+            "window_days": window,
+            "refreshed_at": self._last_refreshed.isoformat(),
+        }
+        logger.info(
+            f"[Affinity] learned kitchens for {len(learned_pref)} riders, "
+            f"home for {len(learned_home)} (min_deliveries={min_n})"
+        )
+        return summary
+
+    def _ensure_loaded(self) -> None:
+        if self._last_refreshed is None:
+            try:
+                self.refresh()
+            except Exception as e:
+                logger.debug(f"[Affinity] lazy refresh failed: {e}")
+
+    # ------------------------------------------------------------------
+
+    def get_preferred_kitchens(self) -> Dict[int, List[str]]:
+        """Curated config ∪ learned (union — never drops a curated owner)."""
+        from app.config.dynamic_config import get_config
+        if not bool(get_config().get("rider_affinity_enabled", True)):
+            return {rid: list(v) for rid, v in RIDER_PREFERRED_KITCHENS.items()}
+        self._ensure_loaded()
+        merged: Dict[int, List[str]] = {rid: list(v) for rid, v in RIDER_PREFERRED_KITCHENS.items()}
+        with self._lock:
+            learned = {rid: list(v) for rid, v in self._learned_pref.items()}
+        for rid, kitchens in learned.items():
+            base = merged.setdefault(rid, [])
+            base_lower = {b.lower() for b in base}
+            for k in kitchens:
+                if k.lower() not in base_lower:
+                    base.append(k)
+        return merged
+
+    def get_home_locations(self) -> Dict[int, Tuple[float, float]]:
+        """Curated home primary; learned drop centroid only for riders absent from config."""
+        from app.config.dynamic_config import get_config
+        if not bool(get_config().get("rider_affinity_enabled", True)):
+            return dict(RIDER_HOME_LOCATIONS)
+        self._ensure_loaded()
+        merged: Dict[int, Tuple[float, float]] = dict(RIDER_HOME_LOCATIONS)
+        with self._lock:
+            learned = dict(self._learned_home)
+        for rid, home in learned.items():
+            if rid not in merged or merged.get(rid) in (None, (0.0, 0.0)):
+                merged[rid] = home
+        return merged
+
+    def get_summary(self) -> Dict[str, Any]:
+        """Learned-vs-config diff for the admin endpoint."""
+        self._ensure_loaded()
+        with self._lock:
+            learned_pref = {rid: list(v) for rid, v in self._learned_pref.items()}
+            learned_home = dict(self._learned_home)
+        config_riders = set(RIDER_PREFERRED_KITCHENS)
+        new_pref_riders = sorted(set(learned_pref) - config_riders)
+        return {
+            "last_refreshed": self._last_refreshed.isoformat() if self._last_refreshed else None,
+            "config_preferred_riders": sorted(config_riders),
+            "learned_preferred": {str(k): v for k, v in sorted(learned_pref.items())},
+            "riders_newly_learned_not_in_config": new_pref_riders,
+            "learned_home_count": len(learned_home),
+        }
+
+
+_affinity: Optional[RiderAffinityService] = None
+_affinity_lock = threading.Lock()
+
+
+def get_rider_affinity() -> RiderAffinityService:
+    global _affinity
+    with _affinity_lock:
+        if _affinity is None:
+            _affinity = RiderAffinityService()
+        return _affinity

app/services/routing/road_sequencing_agent.py

@@ -0,0 +1,233 @@
+"""
+Road-Sequencing Decision Agent
+==============================
+
+Autonomous controller for the `routing_use_road_distance` feature. Instead of a
+human flipping a config flag, this agent periodically *measures* whether road-aware
+sequencing (road travel-time matrix + OR-Tools open-TSP) actually beats the default
+straight-line ordering on REAL recent batches, and turns the feature on or off by
+itself — re-checking every cycle so it self-corrects if the gain ever disappears.
+
+Decision (with hysteresis, so it doesn't flap):
+    mean travel-time gain >= routing_auto_enable_gain_pct  -> enable
+    mean travel-time gain <  routing_auto_disable_gain_pct -> disable
+    in between                                             -> leave as-is
+
+Every decision is stored (auditable) in DynamicConfig under `routing_road_eval`
+and exposed via GET /api/v1/ml/road-eval. Cost is bounded: it evaluates only
+`routing_eval_sample_batches` batches once per `routing_eval_interval_hours`.
+"""
+
+import asyncio
+import json
+import logging
+import threading
+import time
+from datetime import datetime
+from typing import Any, Dict, List, Optional
+
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+
+def _total(order_idx: List[int], matrix: List[List[float]]) -> float:
+    """Total open-route cost: origin(0) -> drops in `order_idx` (1-based into matrix)."""
+    seq = [0] + [i + 1 for i in order_idx]
+    return sum(matrix[seq[k]][seq[k + 1]] for k in range(len(seq) - 1))
+
+
+class RoadSequencingAgent:
+    """Measures road-vs-aerial sequencing on real batches and auto-toggles the flag."""
+
+    def __init__(self):
+        self._opt = None  # lazy RouteOptimizer
+        self._scheduler_started = False
+        self._lock = threading.Lock()
+        self.last_decision: Dict[str, Any] = {}
+
+    def _optimizer(self):
+        if self._opt is None:
+            from app.services.routing.route_optimizer import RouteOptimizer
+            self._opt = RouteOptimizer()
+        return self._opt
+
+    async def evaluate(self, sample_batches: int = 8, days: int = 14) -> Dict[str, Any]:
+        """
+        Compare aerial-order vs road-order travel time on sampled real batches.
+        Returns mean gain % and per-batch detail. No flag changes here.
+        """
+        from app.services.routing.delivery_history_service import get_delivery_history_service
+        from app.core.arrow_utils import calculate_haversine_matrix_vectorized
+
+        opt = self._optimizer()
+        if not opt.use_google_maps:
+            return {"evaluated": 0, "reason": "no_google_key", "mean_gain_pct": 0.0}
+
+        batches = get_delivery_history_service().sample_batches(
+            days=days, limit=sample_batches
+        )
+        if not batches:
+            return {"evaluated": 0, "reason": "no_local_batches", "mean_gain_pct": 0.0}
+
+        per_batch: List[Dict[str, Any]] = []
+        for b in batches:
+            origin, drops = b["origin"], b["drops"]
+            locs = [origin] + drops
+            matrix = await opt._road_duration_matrix(locs)
+            if matrix is None:
+                continue
+            # aerial order (current production behaviour)
+            lats = np.array([p[0] for p in locs])
+            lons = np.array([p[1] for p in locs])
+            aerial = calculate_haversine_matrix_vectorized(lats, lons)
+            aer = [i - 1 for i in opt._two_opt_improve(opt._solve_greedy(locs, aerial), aerial) if i != 0]
+            # road order (OR-Tools open-TSP on the road travel-time matrix)
+            road = [i - 1 for i in opt._solve_tsp_ortools(locs, matrix) if i != 0]
+
+            # Human's ACTUAL route: drops are already in delivered order
+            # (_load_raw_rows is ORDER BY deliverytime), so identity = what the
+            # rider really drove. This is the "do we beat the humans?" baseline.
+            human = list(range(len(drops)))
+
+            t_aer = _total(aer, matrix)
+            t_road = _total(road, matrix)
+            t_human = _total(human, matrix)
+            gain = (100.0 * (t_aer - t_road) / t_aer) if t_aer > 0 else 0.0
+            gain_vs_human = (100.0 * (t_human - t_road) / t_human) if t_human > 0 else 0.0
+            per_batch.append({
+                "rider": b["rider"], "day": b["day"], "drops": len(drops),
+                "aerial_min": round(t_aer, 1), "road_min": round(t_road, 1),
+                "human_min": round(t_human, 1),
+                "gain_pct": round(gain, 1),
+                "gain_vs_human_pct": round(gain_vs_human, 1),
+                "beat_human": t_road <= t_human + 1e-9,
+            })
+
+        if not per_batch:
+            return {"evaluated": 0, "reason": "matrix_unavailable", "mean_gain_pct": 0.0}
+
+        n = len(per_batch)
+        mean_gain = sum(x["gain_pct"] for x in per_batch) / n
+        mean_vs_human = sum(x["gain_vs_human_pct"] for x in per_batch) / n
+        beats = sum(1 for x in per_batch if x["beat_human"])
+        return {
+            "evaluated": n,
+            "mean_gain_pct": round(mean_gain, 2),
+            "median_gain_pct": round(sorted(x["gain_pct"] for x in per_batch)[n // 2], 2),
+            "human_beat_rate_pct": round(100.0 * beats / n, 1),
+            "human_beat_count": f"{beats}/{n}",
+            "mean_gain_vs_human_pct": round(mean_vs_human, 2),
+            "per_batch": per_batch,
+        }
+
+    def decide_and_apply(self) -> Dict[str, Any]:
+        """Run an evaluation and autonomously enable/disable road sequencing."""
+        from app.config.dynamic_config import get_config
+        cfg = get_config()
+
+        sample = int(cfg.get("routing_eval_sample_batches", 8))
+        days = int(cfg.get("routing_eval_days", 14))
+        enable_thr = float(cfg.get("routing_auto_enable_gain_pct", 3.0))
+        disable_thr = float(cfg.get("routing_auto_disable_gain_pct", 1.0))
+        min_batches = int(cfg.get("routing_eval_min_batches", 3))
+        auto_manage = bool(cfg.get("routing_auto_manage", True))
+
+        try:
+            result = asyncio.run(self.evaluate(sample_batches=sample, days=days))
+        except RuntimeError:
+            # An event loop is already running in this thread — use a fresh one.
+            loop = asyncio.new_event_loop()
+            try:
+                result = loop.run_until_complete(self.evaluate(sample_batches=sample, days=days))
+            finally:
+                loop.close()
+
+        current = bool(cfg.get("routing_use_road_distance", False))
+        evaluated = result.get("evaluated", 0)
+        gain = result.get("mean_gain_pct", 0.0)
+        action = "kept"
+        new_state = current
+
+        if not auto_manage:
+            action = "auto_manage_off"
+        elif evaluated < min_batches:
+            action = "insufficient_data"
+        else:
+            if gain >= enable_thr and not current:
+                new_state = True
+                cfg.set("routing_use_road_distance", True, source="road_agent")
+                action = "enabled"
+            elif gain < disable_thr and current:
+                new_state = False
+                cfg.set("routing_use_road_distance", False, source="road_agent")
+                action = "disabled"
+
+        decision = {
+            "decided_at": datetime.utcnow().isoformat(),
+            "action": action,
+            "flag_before": current,
+            "flag_after": new_state,
+            "mean_gain_pct": gain,
+            "enable_threshold_pct": enable_thr,
+            "disable_threshold_pct": disable_thr,
+            "evaluation": result,
+        }
+        self.last_decision = decision
+        try:
+            # Persist a compact copy for audit (without the full per-batch list).
+            compact = {k: v for k, v in decision.items() if k != "evaluation"}
+            compact["evaluated"] = evaluated
+            compact["human_beat_rate_pct"] = result.get("human_beat_rate_pct")
+            compact["mean_gain_vs_human_pct"] = result.get("mean_gain_vs_human_pct")
+            cfg.set("routing_road_eval", compact, source="road_agent")
+        except Exception:
+            pass
+
+        logger.info(
+            f"[RoadAgent] action={action} gain_vs_aerial={gain}% evaluated={evaluated} "
+            f"flag {current}->{new_state}"
+        )
+        if evaluated:
+            logger.info(
+                f"[RoadAgent] beat humans on {result.get('human_beat_count', '?')} batches "
+                f"(avg {result.get('mean_gain_vs_human_pct', 0)}% faster than actual delivered order)"
+            )
+        return decision
+
+    def ensure_background_agent(self, interval_hours: int = 24,
+                               warmup_seconds: int = 120) -> bool:
+        """Start the autonomous decision loop once (daemon thread)."""
+        with self._lock:
+            if self._scheduler_started:
+                return False
+            self._scheduler_started = True
+
+        def _loop():
+            # Let the ETA sync agent populate the local mirror first.
+            time.sleep(max(0, warmup_seconds))
+            logger.info(f"[RoadAgent] autonomous road-sequencing decision loop started "
+                        f"(interval={interval_hours}h)")
+            while True:
+                try:
+                    self.decide_and_apply()
+                except Exception as e:
+                    logger.warning(f"[RoadAgent] decision cycle failed (will retry): {e}")
+                from app.config.dynamic_config import get_config
+                hrs = int(get_config().get("routing_eval_interval_hours", interval_hours))
+                time.sleep(max(1, hrs) * 3600)
+
+        threading.Thread(target=_loop, daemon=True, name="road-seq-agent").start()
+        return True
+
+
+_agent: Optional[RoadSequencingAgent] = None
+_agent_lock = threading.Lock()
+
+
+def get_road_agent() -> RoadSequencingAgent:
+    global _agent
+    with _agent_lock:
+        if _agent is None:
+            _agent = RoadSequencingAgent()
+        return _agent

app/services/routing/zone_service.py

@@ -0,0 +1,201 @@
+
+import logging
+from typing import List, Dict, Any
+
+logger = logging.getLogger(__name__)
+
+class ZoneService:
+    """
+    Service to classify orders and riders into geographic zones.
+    Defaulting to Coimbatore logic as per user context.
+    """
+    
+    # Approximate Center of Coimbatore (Gandhipuram/Bus Stand area)
+    CENTER_LAT = 11.0168
+    CENTER_LON = 76.9558
+    
+    def __init__(self):
+        pass
+
+    def determine_zone(self, lat: float, lon: float) -> str:
+        """
+        Determine the zone (North, South, East, West, etc.) based on coordinates.
+        """
+        if lat == 0 or lon == 0:
+            return "Unknown"
+
+        lat_diff = lat - self.CENTER_LAT
+        lon_diff = lon - self.CENTER_LON
+        
+        # Simple Quadrant Logic
+        # North: +Lat
+        # South: -Lat
+        # East: +Lon
+        # West: -Lon
+        
+        # Define a small central buffer (0.01 degrees ~ 1.1km)
+        buffer = 0.010
+        
+        is_north = lat_diff > buffer
+        is_south = lat_diff < -buffer
+        is_east = lon_diff > buffer
+        is_west = lon_diff < -buffer
+        
+        zone_parts = []
+        
+        if is_north: zone_parts.append("North")
+        elif is_south: zone_parts.append("South")
+        
+        if is_east: zone_parts.append("East")
+        elif is_west: zone_parts.append("West")
+        
+        if not zone_parts:
+            return "Central"
+            
+        return " ".join(zone_parts)
+
+    def group_by_zones(self, flat_orders: List[Dict[str, Any]], unassigned_orders: List[Dict[str, Any]] = None, fuel_charge: float = 2.5, base_pay: float = 0.0) -> Dict[str, Any]:
+        """
+        Group a flat list of optimized orders into Zones -> Riders -> Orders.
+        Calculates profit per order and per zone.
+        """
+        zones_map = {} # "North East": { "riders": { rider_id: [orders] } }
+        unassigned_orders = unassigned_orders or []
+        
+        # Merge both for initial processing if you want everything zoned
+        all_to_process = []
+        for o in flat_orders:
+            all_to_process.append((o, True))
+        for o in unassigned_orders:
+            all_to_process.append((o, False))
+
+        for order, is_assigned in all_to_process:
+            # 1. Extract Coords
+            try:
+                # Prefer Delivery location for zoning (where the customer is)
+                lat = float(order.get("deliverylat") or order.get("droplat") or 0)
+                lon = float(order.get("deliverylong") or order.get("droplon") or 0)
+                pincode = str(order.get("deliveryzip") or "")
+            except:
+                lat, lon, pincode = 0, 0, ""
+            
+            # 2. Get Zone
+            zone_name = self.determine_zone(lat, lon)
+            order["zone_name"] = zone_name
+            
+            # 3. Initialize Zone Bucket
+            if zone_name not in zones_map:
+                zones_map[zone_name] = {
+                    "riders_map": {}, 
+                    "total_orders": 0,
+                    "assigned_orders": 0,
+                    "unassigned_orders": [],
+                    "total_kms": 0.0,
+                    "total_profit": 0.0
+                }
+            
+            # 4. Add to Rider bucket within Zone
+            rider_id = order.get("userid") or order.get("_id")
+            
+            # Track kms and profit for this zone
+            try:
+                # 'actualkms' is preferred for delivery distance
+                dist = float(order.get("actualkms", order.get("previouskms", 0)))
+                zones_map[zone_name]["total_kms"] += dist
+                
+                # Individual charge for this order: Fixed Base + Variable Distance
+                order_amount = float(order.get("orderamount") or order.get("deliveryamount") or 0)
+                rider_payment = dist * fuel_charge
+                profit = order_amount - rider_payment
+                
+                order["rider_charge"] = round(rider_payment, 2)
+                order["profit"] = round(profit, 2)
+
+                # ── PROFIT CLASS (separate from ordertype) ───────────────
+                # `ordertype` is set by route_optimizer based on delivery
+                # DISTANCE (Economy ≤5km, Premium ≤12km, Risky >12km).
+                # Overwriting it here with a profit-based label broke ETA
+                # calculations and downstream routing logic.
+                # Use `profit_class` for profit-based analytics instead.
+                if profit <= 0:
+                    order["profit_class"] = "Loss"
+                elif profit <= 5:
+                    order["profit_class"] = "Marginal"
+                elif profit <= 10:
+                    order["profit_class"] = "Profitable"
+                else:
+                    order["profit_class"] = "HighMargin"
+
+                zones_map[zone_name]["total_profit"] += profit
+            except:
+                pass
+
+            # If strictly unassigned order (no rider), put in unassigned
+            if not is_assigned:
+                zones_map[zone_name]["unassigned_orders"].append(order)
+            else:
+                str_rid = str(rider_id)
+                if str_rid not in zones_map[zone_name]["riders_map"]:
+                    zones_map[zone_name]["riders_map"][str_rid] = {
+                        "rider_details": {
+                            "id": str_rid,
+                            "name": order.get("username", "Unknown")
+                        },
+                        "orders": []
+                    }
+                zones_map[zone_name]["riders_map"][str_rid]["orders"].append(order)
+                zones_map[zone_name]["assigned_orders"] += 1
+            
+            zones_map[zone_name]["total_orders"] += 1
+
+        # 5. Restructure for API Response
+        output_zones = []
+        zone_metrics = []
+        
+        sorted_zone_names = sorted(zones_map.keys())
+        
+        for z_name in sorted_zone_names:
+            z_data = zones_map[z_name]
+            
+            # Flatten riders map
+            riders_list = []
+            for r_id, r_data in z_data["riders_map"].items():
+                riders_list.append({
+                    "rider_id": r_data["rider_details"]["id"],
+                    "rider_name": r_data["rider_details"]["name"],
+                    "orders_count": len(r_data["orders"]),
+                    "orders": r_data["orders"]
+                })
+            
+            # Create the flat metric summary
+            metrics = {
+                "zone_name": z_name,
+                "total_orders": z_data["total_orders"],
+                "assigned_orders": z_data["assigned_orders"],
+                "unassigned_orders_count": len(z_data["unassigned_orders"]),
+                "active_riders_count": len(riders_list),
+                "total_delivery_kms": round(z_data["total_kms"], 2),
+                "total_profit": round(z_data["total_profit"], 2)
+            }
+            
+            zone_metrics.append(metrics)
+            
+            # Create the detailed zone object with flattened metrics
+            zone_obj = {
+                "zone_name": z_name,
+                "total_orders": metrics["total_orders"],
+                "active_riders_count": metrics["active_riders_count"],
+                "assigned_orders": metrics["assigned_orders"],
+                "unassigned_orders_count": metrics["unassigned_orders_count"],
+                "total_delivery_kms": metrics["total_delivery_kms"],
+                "total_profit": metrics["total_profit"],
+                "riders": riders_list,
+                "unassigned_orders": z_data["unassigned_orders"]
+            }
+            
+            output_zones.append(zone_obj)
+            
+        return {
+            "detailed_zones": output_zones,
+            "zone_analysis": zone_metrics
+        }