diff --git a/app/__pycache__/main.cpython-312.pyc b/app/__pycache__/main.cpython-312.pyc
index 75c5542..9f1d46b 100644
Binary files a/app/__pycache__/main.cpython-312.pyc and b/app/__pycache__/main.cpython-312.pyc differ
diff --git a/app/config/__pycache__/dynamic_config.cpython-312.pyc b/app/config/__pycache__/dynamic_config.cpython-312.pyc
index 4aa0b5a..792b859 100644
Binary files a/app/config/__pycache__/dynamic_config.cpython-312.pyc and b/app/config/__pycache__/dynamic_config.cpython-312.pyc differ
diff --git a/app/config/dynamic_config.py b/app/config/dynamic_config.py
index 08569aa..b73f18b 100644
--- a/app/config/dynamic_config.py
+++ b/app/config/dynamic_config.py
@@ -1,14 +1,14 @@
"""
Dynamic Configuration - rider-api
-Replaces all hardcoded hyperparameters with DB-backed values.
-The ML hypertuner writes optimal values here; services read from here.
+Replaces all hardcoded hyperparameters with DB-backed values. The autonomous
+agents (ETA sync, road-sequencing, rider affinity) and the ml_admin API write
+tuned values here; services read from here.
Fallback: If DB is unavailable or no tuned values exist, defaults are used.
This means zero risk - the system works day 1 with no data.
"""
-import contextvars
import json
import logging
import os
@@ -21,32 +21,9 @@ logger = logging.getLogger(__name__)
# --- DB Path ------------------------------------------------------------------
_DB_PATH = os.getenv("ML_DB_PATH", "ml_data/ml_store.db")
-# ---------------------------------------------------------------------------
-# Per-request strategy override (async-safe via contextvars).
-# Each FastAPI request/asyncio task gets its own copy — no cross-request leaks.
-# Usage:
-# set_request_strategy("fuel_saver") → override active for this request
-# clear_request_strategy() → restore to DB value
-# ---------------------------------------------------------------------------
-_strategy_override: contextvars.ContextVar[Optional[str]] = contextvars.ContextVar(
- "strategy_override", default=None
-)
-
-
-def set_request_strategy(strategy: Optional[str]) -> None:
- """Override ml_strategy for the current async task only (thread-safe)."""
- _strategy_override.set(strategy)
-
-
-def clear_request_strategy() -> None:
- """Remove the per-request strategy override for the current async task."""
- _strategy_override.set(None)
-
# --- Hard Defaults (What the system used before ML) ---------------------------
DEFAULTS: Dict[str, Any] = {
- # System Strategy / Prompt
- "ml_strategy": "balanced",
# AssignmentService
"max_pickup_distance_km": 10.0,
"max_kitchen_distance_km": 3.0,
@@ -65,9 +42,6 @@ DEFAULTS: Dict[str, Any] = {
"road_factor": 1.3,
# ClusteringService
"cluster_radius_km": 3.0,
- # KalmanFilter
- "kalman_process_noise": 1e-4,
- "kalman_measurement_noise": 0.01,
# RealisticETACalculator
"eta_pickup_time_min": 3.0,
"eta_delivery_time_min": 4.0,
@@ -101,6 +75,18 @@ DEFAULTS: Dict[str, Any] = {
"rider_affinity_enabled": True, # False -> pure curated config
"rider_affinity_min_deliveries": 10, # learned owner needs >= this many deliveries
"rider_affinity_refresh_hours": 6, # recompute cadence (piggybacks the agent)
+ # Phase-0 kitchen+zone pattern store (delivery_history_store.py): which rider
+ # historically owns a kitchen->drop-zone pair, used to pre-assign orders.
+ # "csv" -> legacy: built from delivery_details.csv, only refreshed when a
+ # human re-exports it and calls POST /ml/reload-history.
+ # "db" -> built from the already-synced nearledb mirror (delivery_raw),
+ # rebuilt automatically every eta_sync_interval_hours — no manual
+ # step, no extra DB load (reuses rows the ETA sync already pulled).
+ "pattern_source": "csv",
+ "pattern_history_days": 30, # retention floor for delivery_raw kept for pattern-
+ # matching volume; independent of eta_history_days
+ # so the (already-validated) empirical ETA window
+ # is untouched.
}
@@ -137,18 +123,8 @@ class DynamicConfig:
# --------------------------------------------------------------------------
def get(self, key: str, default: Any = None) -> Any:
- """Get a config value. Returns ML-tuned value if available, else default.
-
- For 'ml_strategy' specifically, a per-request ContextVar override takes
- precedence so that hypertuning_params requests don't mutate the shared
- singleton (thread-safe for concurrent FastAPI requests).
- """
+ """Get a config value. Returns ML-tuned value if available, else default."""
self._maybe_reload()
- # Per-request strategy override (async-safe, no cross-request leaks)
- if key == "ml_strategy":
- override = _strategy_override.get()
- if override is not None:
- return override
val = self._cache.get(key)
if val is not None:
return val
@@ -163,7 +139,7 @@ class DynamicConfig:
return result
def set(self, key: str, value: Any, source: str = "manual") -> None:
- """Write a config value to DB (used by hypertuner)."""
+ """Write a config value to DB (used by the ml_admin API and agents)."""
try:
os.makedirs(os.path.dirname(_DB_PATH) or ".", exist_ok=True)
conn = sqlite3.connect(_DB_PATH)
@@ -185,8 +161,8 @@ class DynamicConfig:
except Exception as e:
logger.error(f"[DynamicConfig] Failed to set {key}: {e}")
- def set_bulk(self, params: Dict[str, Any], source: str = "ml_hypertuner") -> None:
- """Write multiple config values at once (called after each Optuna study)."""
+ def set_bulk(self, params: Dict[str, Any], source: str = "manual") -> None:
+ """Write multiple config values at once."""
for key, value in params.items():
self.set(key, value, source=source)
logger.info(f"[DynamicConfig] Bulk update: {len(params)} params from {source}")
@@ -219,16 +195,6 @@ class DynamicConfig:
updated_at TEXT
)
""")
- conn.execute("""
- CREATE TABLE IF NOT EXISTS kitchen_encoding (
- kitchen_name TEXT PRIMARY KEY,
- label_id INTEGER NOT NULL,
- frequency REAL DEFAULT 0.0,
- avg_profit REAL DEFAULT 0.0,
- order_count INTEGER DEFAULT 0,
- updated_at TEXT
- )
- """)
conn.commit()
conn.close()
except Exception as e:
@@ -278,131 +244,4 @@ def get_config() -> DynamicConfig:
__all__ = [
"DynamicConfig",
"get_config",
- "set_request_strategy",
- "clear_request_strategy",
- "get_kitchen_label_id",
- "get_kitchen_frequency",
- "update_kitchen_stats",
- "get_kitchen_avg_profit_smoothed",
]
-
-
-# --- Kitchen Encoding Persistence ---------------------------------------------
-def get_kitchen_label_id(kitchen_name: str) -> int:
- """Get or create persistent label ID for a kitchen."""
- try:
- conn = sqlite3.connect(_DB_PATH)
- row = conn.execute(
- "SELECT label_id FROM kitchen_encoding WHERE kitchen_name = ?",
- (kitchen_name,),
- ).fetchone()
- if row:
- conn.close()
- return row[0]
-
- max_id = conn.execute(
- "SELECT COALESCE(MAX(label_id), -1) FROM kitchen_encoding"
- ).fetchone()[0]
- new_id = max_id + 1
- conn.close()
- return new_id
- except Exception as e:
- logger.warning(f"[KitchenEncoding] Failed to get label_id: {e}")
- return hash(kitchen_name.lower().strip()) % 10000
-
-
-def get_kitchen_frequency(kitchen_name: str) -> float:
- """Get frequency ratio for a kitchen from DB."""
- try:
- conn = sqlite3.connect(_DB_PATH)
- row = conn.execute(
- "SELECT frequency FROM kitchen_encoding WHERE kitchen_name = ?",
- (kitchen_name,),
- ).fetchone()
- conn.close()
- return row[0] if row else 0.0
- except Exception:
- return 0.0
-
-
-def update_kitchen_stats(kitchen_name: str, profit: float):
- """Update kitchen stats after order completion."""
- try:
- conn = sqlite3.connect(_DB_PATH)
- row = conn.execute(
- "SELECT order_count, avg_profit FROM kitchen_encoding WHERE kitchen_name = ?",
- (kitchen_name,),
- ).fetchone()
-
- if row:
- count, avg = row
- new_count = count + 1
- new_avg = ((avg * count) + profit) / new_count
- new_freq = new_count / (
- conn.execute(
- "SELECT SUM(order_count) FROM kitchen_encoding"
- ).fetchone()[0]
- or 1
- )
- conn.execute(
- "UPDATE kitchen_encoding SET order_count = ?, avg_profit = ?, frequency = ?, updated_at = ? WHERE kitchen_name = ?",
- (
- new_count,
- new_avg,
- new_freq,
- datetime.utcnow().isoformat(),
- kitchen_name,
- ),
- )
- else:
- conn.execute(
- "INSERT INTO kitchen_encoding (kitchen_name, label_id, frequency, avg_profit, order_count, updated_at) VALUES (?, ?, ?, ?, ?, ?)",
- (
- kitchen_name,
- get_kitchen_label_id(kitchen_name),
- 1.0,
- profit,
- 1,
- datetime.utcnow().isoformat(),
- ),
- )
- conn.commit()
- conn.close()
- except Exception as e:
- logger.warning(f"[KitchenEncoding] Failed to update stats: {e}")
-
-
-def get_kitchen_avg_profit_smoothed(
- kitchen_name: str, global_avg: float = 40.0, min_samples: int = 5
-) -> float:
- """
- Get smoothed average profit for a kitchen using Bayesian smoothing.
- Reduces noise for kitchens with few orders.
-
- Formula: smoothed = (kitchen_count * kitchen_avg + min_samples * global_avg) / (kitchen_count + min_samples)
-
- This means:
- - Kitchen with many samples -> uses its own avg
- - Kitchen with few samples -> pulls toward global avg
- """
- try:
- conn = sqlite3.connect(_DB_PATH)
- row = conn.execute(
- "SELECT order_count, avg_profit FROM kitchen_encoding WHERE kitchen_name = ?",
- (kitchen_name,),
- ).fetchone()
- conn.close()
-
- if row and row[0] > 0:
- count, avg = row
- if count >= min_samples:
- return avg
- # Bayesian smoothing
- smoothed = ((count * avg) + (min_samples * global_avg)) / (
- count + min_samples
- )
- return smoothed
-
- return global_avg # Unknown kitchen defaults to global avg
- except Exception:
- return global_avg
diff --git a/app/core/__pycache__/arrow_utils.cpython-312.pyc b/app/core/__pycache__/arrow_utils.cpython-312.pyc
index 18e0c82..ed44d5c 100644
Binary files a/app/core/__pycache__/arrow_utils.cpython-312.pyc and b/app/core/__pycache__/arrow_utils.cpython-312.pyc differ
diff --git a/app/core/arrow_utils.py b/app/core/arrow_utils.py
index 738c081..f077316 100644
--- a/app/core/arrow_utils.py
+++ b/app/core/arrow_utils.py
@@ -1,13 +1,9 @@
"""
-High-performance utilities using Apache Arrow and NumPy for geographic data.
-Provides vectorized operations for distances and coordinate processing.
+Vectorized NumPy utilities for geographic distance calculations.
"""
import numpy as np
-import pyarrow as pa
-import pyarrow.parquet as pq
import logging
-from typing import List, Dict, Any, Tuple
logger = logging.getLogger(__name__)
@@ -35,29 +31,3 @@ def calculate_haversine_matrix_vectorized(lats: np.ndarray, lons: np.ndarray) ->
c = 2 * np.arctan2(np.sqrt(a), np.sqrt(1 - a))
return R * c
-
-def orders_to_arrow_table(orders: List[Dict[str, Any]]) -> pa.Table:
- """
- Convert a list of order dictionaries to an Apache Arrow Table.
- This enables zero-copy operations and efficient columnar storage.
- """
- return pa.Table.from_pylist(orders)
-
-def save_optimized_route_parquet(orders: List[Dict[str, Any]], filename: str):
- """
- Save optimized route data to a Parquet file for high-speed analysis.
- Useful for logging and historical simulation replays.
- """
- try:
- table = orders_to_arrow_table(orders)
- pq.write_table(table, filename)
- logger.info(f" Saved route data to Parquet: {filename}")
- except Exception as e:
- logger.error(f" Failed to save Parquet: {e}")
-
-def load_route_parquet(filename: str) -> List[Dict[str, Any]]:
- """
- Load route data from a Parquet file and return as a list of dicts.
- """
- table = pq.read_table(filename)
- return table.to_pylist()
diff --git a/app/main.py b/app/main.py
index fa7013a..dde5008 100644
--- a/app/main.py
+++ b/app/main.py
@@ -83,19 +83,6 @@ async def lifespan(app: FastAPI):
except Exception as e:
logger.warning(f"[Analytics] DB check failed (non-fatal): {e}")
- # Warm up the Thompson Sampling bandit (bootstraps from historical DB)
- try:
- from app.services.ml.strategy_bandit import get_bandit
- bandit = get_bandit()
- stats = bandit.get_stats()
- logger.info(
- f"[Bandit] RL strategy bandit ready — "
- f"{stats['context_count']} contexts, "
- f"{stats['total_updates']} historical updates loaded."
- )
- except Exception as e:
- logger.warning(f"[Bandit] Warm-up failed (non-fatal): {e}")
-
# Start the autonomous empirical-ETA sync agent (daemon thread).
# It mirrors completed deliveries locally and rebuilds learned ETAs on a
# schedule, so the request path never touches Postgres.
diff --git a/app/routes/__pycache__/__init__.cpython-312.pyc b/app/routes/__pycache__/__init__.cpython-312.pyc
index d51954a..0c5d8ff 100644
Binary files a/app/routes/__pycache__/__init__.cpython-312.pyc and b/app/routes/__pycache__/__init__.cpython-312.pyc differ
diff --git a/app/routes/__pycache__/ml_admin.cpython-312.pyc b/app/routes/__pycache__/ml_admin.cpython-312.pyc
index d83cd6f..03420c3 100644
Binary files a/app/routes/__pycache__/ml_admin.cpython-312.pyc and b/app/routes/__pycache__/ml_admin.cpython-312.pyc differ
diff --git a/app/routes/__pycache__/optimization.cpython-312.pyc b/app/routes/__pycache__/optimization.cpython-312.pyc
index 2128f8e..769eeee 100644
Binary files a/app/routes/__pycache__/optimization.cpython-312.pyc and b/app/routes/__pycache__/optimization.cpython-312.pyc differ
diff --git a/app/routes/__pycache__/riders.cpython-312.pyc b/app/routes/__pycache__/riders.cpython-312.pyc
new file mode 100644
index 0000000..622c4f2
Binary files /dev/null and b/app/routes/__pycache__/riders.cpython-312.pyc differ
diff --git a/app/routes/ml_admin.py b/app/routes/ml_admin.py
index 28de651..be2de82 100644
--- a/app/routes/ml_admin.py
+++ b/app/routes/ml_admin.py
@@ -3,8 +3,8 @@ Analytics & ML Admin API
=========================
Exposes historical assignment quality data, the empirical-ETA pipeline, and the
autonomous road-sequencing / rider-affinity agents. (The XGBoost/Optuna hypertuner,
-ID3 risk tree, and profit predictor have all been retired — they never affected
-assignment.)
+ID3 risk tree, profit predictor, and Thompson-Sampling strategy bandit have all
+been retired — none of them ever affected assignment or routing behavior.)
Endpoints:
GET /api/v1/ml/status – quality trend, analytics DB + history stats
@@ -12,7 +12,6 @@ Endpoints:
GET /api/v1/ml/config – active config values
PATCH /api/v1/ml/config – manual config override
POST /api/v1/ml/reset – reset config to defaults
- POST /api/v1/ml/strategy – change optimization strategy
POST /api/v1/ml/refresh-eta – sync nearledb + rebuild empirical ETA stats
GET /api/v1/ml/eta-accuracy – formula vs empirical ETA backtest
GET/POST /api/v1/ml/road-eval – autonomous road-sequencing decision
@@ -103,7 +102,6 @@ def ml_analytics():
"hourly_stats": collector.get_hourly_stats(),
"zone_stats": collector.get_zone_stats(),
"quality_histogram": collector.get_quality_histogram(),
- "strategy_comparison": collector.get_strategy_comparison(),
}
except Exception as e:
logger.error(f"[ML API] analytics: {e}", exc_info=True)
@@ -163,142 +161,6 @@ def ml_reset():
raise HTTPException(status_code=500, detail=str(e))
-# ---------------------------------------------------------------------------
-# POST /strategy
-# ---------------------------------------------------------------------------
-
-@router.post("/strategy", summary="Change the optimization strategy")
-def ml_strategy(strategy: str = Body(default="balanced", embed=True)):
- """
- Choices: balanced | fuel_saver | aggressive_speed | zone_strict
- Affects how the quality score is computed in analytics only.
- """
- valid = ["balanced", "fuel_saver", "aggressive_speed", "zone_strict"]
- if strategy not in valid:
- raise HTTPException(400, f"Invalid strategy. Choose from {valid}")
- from app.config.dynamic_config import get_config
- try:
- get_config().set("ml_strategy", strategy)
- return {"status": "ok", "strategy": strategy}
- except Exception as e:
- logger.error(f"[ML API] strategy: {e}", exc_info=True)
- raise HTTPException(status_code=500, detail=str(e))
-
-
-# ---------------------------------------------------------------------------
-# POST /auto-tune
-# ---------------------------------------------------------------------------
-
-@router.post("/auto-tune", summary="Run SQL-based strategy auto-tuner")
-def ml_auto_tune():
- """
- Analyses the assignment log and picks the best-performing ml_strategy
- based on average quality score across all recorded calls.
-
- Rules:
- - A strategy needs ≥ 10 calls to be considered.
- - At least 2 strategies must have enough data to compare.
- - If a better strategy is found it is written to DynamicConfig
- immediately and takes effect on the next /riderassign call.
- - Also returns per-hour breakdown so you can see peak-hour patterns.
-
- Safe to call any time. Runs in < 100ms.
- """
- from app.services.ml.ml_data_collector import get_collector
- from app.config.dynamic_config import get_config, DEFAULTS
- try:
- collector = get_collector()
- cfg = get_config()
- comparison = collector.get_strategy_comparison()
- hourly = collector.get_hourly_stats()
- total_records = collector.count_records()
-
- if total_records == 0:
- return {
- "status": "no_data",
- "message": "No assignment events logged yet. "
- "Call /riderassign a few times first.",
- "total_records": 0,
- }
-
- qualified = [s for s in comparison if s["call_count"] >= 10]
- current_strategy = cfg.get("ml_strategy", "balanced")
- action = "no_change"
- recommendation = None
-
- if len(qualified) >= 2:
- best = max(qualified, key=lambda x: x["avg_quality"])
- recommendation = best["strategy"]
- if best["strategy"] != current_strategy:
- cfg.set("ml_strategy", best["strategy"], source="auto_tuner")
- action = "updated"
- logger.info(
- f"[AutoTune API] Strategy: '{current_strategy}' → "
- f"'{best['strategy']}' (quality={best['avg_quality']:.1f})"
- )
- elif len(comparison) > 0:
- action = "insufficient_data"
- recommendation = comparison[0]["strategy"] # best so far even if < 10 calls
-
- # Per-hour best strategy (informational — not auto-applied)
- # Shows which strategy logged the highest quality at each hour
- hour_best: list = []
- if hourly:
- for h in hourly:
- # Find which strategy performed best in this hour block
- # (simple: use the dominant strategy for that hour from comparison)
- hour_best.append({
- "hour": h["hour"],
- "avg_quality": h["avg_quality"],
- "call_count": h["call_count"],
- "sla_breaches": h["sla_breaches"],
- })
-
- return {
- "status": "ok",
- "action": action,
- "current_strategy": cfg.get("ml_strategy", "balanced"),
- "recommendation": recommendation,
- "total_records": total_records,
- "strategy_comparison": comparison,
- "hourly_quality": hour_best,
- "message": (
- f"Strategy updated to '{recommendation}'."
- if action == "updated"
- else "Current strategy is already optimal."
- if action == "no_change"
- else "More data needed (≥ 10 calls per strategy to compare)."
- ),
- }
- except Exception as e:
- logger.error(f"[ML API] auto-tune: {e}", exc_info=True)
- raise HTTPException(status_code=500, detail=str(e))
-
-
-# ---------------------------------------------------------------------------
-# GET /bandit
-# ---------------------------------------------------------------------------
-
-@router.get("/bandit", summary="Thompson Sampling bandit — posterior stats per context")
-def ml_bandit():
- """
- Shows the current state of the RL strategy bandit.
-
- Each context (time_band|load_band) has 4 arms (strategies).
- For each arm:
- mean_reward — expected quality / 100 based on posterior mean
- observations — number of observed calls (excluding prior)
- alpha / beta — Beta distribution parameters
-
- The bandit uses Thompson Sampling to select strategies automatically
- on every /riderassign call, balancing exploration vs exploitation.
- """
- try:
- from app.services.ml.strategy_bandit import get_bandit
- return {"status": "ok", **get_bandit().get_stats()}
- except Exception as e:
- logger.error(f"[ML API] bandit: {e}", exc_info=True)
- raise HTTPException(status_code=500, detail=str(e))
# ---------------------------------------------------------------------------
@@ -346,21 +208,24 @@ def ml_history(patterns: bool = False):
(all clear dominant-rider zones, sorted by pattern score).
"""
from app.services.vector.delivery_history_store import (
- get_delivery_history_store, _INDEX_PATH, _META_PATH, CSV_PATH
+ get_delivery_history_store, _paths_for, _pattern_source, CSV_PATH
)
try:
- store = get_delivery_history_store()
- meta = {}
- if os.path.isfile(_META_PATH):
- with open(_META_PATH, "r", encoding="utf-8") as f:
+ store = get_delivery_history_store()
+ source = _pattern_source()
+ vectors_path, _records_path, meta_path = _paths_for(source)
+ meta = {}
+ if os.path.isfile(meta_path):
+ with open(meta_path, "r", encoding="utf-8") as f:
meta = json.load(f)
resp = {
"status": "ok",
+ "pattern_source": source,
"record_count": store.record_count(),
"pattern_count": store.pattern_count(),
"index_ready": store.record_count() > 0,
- "disk_index": os.path.isfile(_INDEX_PATH),
+ "disk_index": os.path.isfile(vectors_path),
"csv_path": CSV_PATH,
"csv_exists": os.path.isfile(CSV_PATH),
"saved_meta": meta,
diff --git a/app/routes/optimization.py b/app/routes/optimization.py
index eac8850..ed8ee2f 100644
--- a/app/routes/optimization.py
+++ b/app/routes/optimization.py
@@ -11,7 +11,6 @@ from fastapi import APIRouter, Body, Request, Depends, status, HTTPException, Qu
from app.controllers.route_controller import RouteController
from app.core.exceptions import APIException
-from app.core.arrow_utils import save_optimized_route_parquet
logger = logging.getLogger(__name__)
@@ -85,68 +84,19 @@ def _haversine_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
return float("inf")
-def _auto_tune_strategy(collector) -> None:
- """
- SQL-based strategy auto-tuner.
-
- Reads quality_score statistics from the assignment log grouped by
- ml_strategy. If one strategy clearly outperforms the rest (minimum
- 10 calls, ≥ 2 strategies compared) it is written to DynamicConfig so
- all future requests benefit automatically.
-
- Safe to call at any time — DynamicConfig write is idempotent.
- """
- try:
- comparison = collector.get_strategy_comparison()
- if not comparison or len(comparison) < 2:
- return # not enough variety to compare
-
- # Only consider strategies with at least 10 real observations
- qualified = [s for s in comparison if s["call_count"] >= 10]
- if not qualified:
- return
-
- best = max(qualified, key=lambda x: x["avg_quality"])
- from app.config.dynamic_config import get_config as _gc
- cfg = _gc()
- current = cfg.get("ml_strategy", "balanced")
- if best["strategy"] != current:
- cfg.set("ml_strategy", best["strategy"], source="auto_tuner")
- logger.info(
- f"[AutoTune] Strategy updated: '{current}' → '{best['strategy']}' "
- f"(avg_quality={best['avg_quality']:.1f}, n={best['call_count']})"
- )
- else:
- logger.debug(
- f"[AutoTune] Strategy '{current}' already optimal "
- f"(avg_quality={best['avg_quality']:.1f})"
- )
- except Exception as _ate:
- logger.warning(f"[AutoTune] Failed (non-fatal): {_ate}")
-
-
-def _bg_log_and_maybe_retrain(
+def _bg_log_assignment(
num_orders: int,
num_riders: int,
hyperparams: dict,
assignments: dict,
unassigned_count: int,
elapsed_ms: float,
- bandit_context: "tuple | None" = None, # (time_band, load_band)
) -> None:
- """
- Background worker: log event → update RL bandit → auto-retrain ID3
- → auto-tune strategy. Runs in _ml_executor (never blocks the response).
-
- Thresholds:
- every call → bandit posterior update (Thompson Sampling RL)
- every 100 evts → greedy SQL auto-tuner (backup check, rarely fires now)
- """
+ """Background worker: log the assignment event. Runs in _ml_executor so
+ it never blocks the API response."""
try:
from app.services.ml.ml_data_collector import get_collector as _gc
collector = _gc()
-
- # ── 1. Log event, get back quality_score ──────────────────────────
quality_score = collector.log_assignment_event(
num_orders=num_orders,
num_riders=num_riders,
@@ -155,29 +105,8 @@ def _bg_log_and_maybe_retrain(
unassigned_count=unassigned_count,
elapsed_ms=elapsed_ms,
) or 50.0
-
count = collector.count_records()
logger.debug(f"[ML BG] Logged event #{count}, quality={quality_score:.1f}")
-
- # ── 2. Thompson Sampling bandit update (RL feedback) ──────────────
- if bandit_context:
- try:
- from app.services.ml.strategy_bandit import get_bandit as _gb
- _t_band, _l_band = bandit_context
- _strategy = hyperparams.get("ml_strategy", "balanced")
- get_bandit_fn = _gb
- get_bandit_fn().update(_t_band, _l_band, _strategy, quality_score)
- logger.debug(
- f"[Bandit] Updated ctx={_t_band}|{_l_band} "
- f"arm={_strategy} reward={quality_score/100:.3f}"
- )
- except Exception as _be:
- logger.debug(f"[Bandit] Update failed (non-fatal): {_be}")
-
- # ── 3. Greedy SQL auto-tune every 100 events (backup) ────────────
- if count >= 20 and count % 100 == 0:
- _auto_tune_strategy(collector)
-
except Exception as _e:
logger.warning(f"[ML BG] Background task failed (non-fatal): {_e}")
@@ -242,18 +171,6 @@ async def provider_optimize_forward(
try:
url = "https://jupiter.nearle.app/live/api/v1/deliveries/createdeliveries"
result = await controller.optimize_and_forward_provider_payload(body, url)
-
- # Parquet snapshot — offloaded so it never blocks the response
- def _snap():
- try:
- os.makedirs("data/snapshots", exist_ok=True)
- snapshot_path = f"data/snapshots/route_{int(time.time())}.parquet"
- save_optimized_route_parquet(body, snapshot_path)
- logger.info(f"Apache Arrow: Snapshot saved to {snapshot_path}")
- except Exception as e:
- logger.warning(f"Could not save Arrow snapshot: {e}")
- _ml_executor.submit(_snap)
-
return result
except APIException:
raise
@@ -554,7 +471,6 @@ async def assign_orders_to_riders(
request: Request,
body: Any = Body(default=None),
reshuffle: bool = Query(False, alias="reshuffle"),
- hypertuning_params: str = None,
):
"""
Smart assignment of orders to riders.
@@ -653,11 +569,10 @@ async def assign_orders_to_riders(
# 3. Log summary after all data is in hand
mode_str = "reshuffle" if do_reshuffle else "normal"
- tuning_str = hypertuning_params if hypertuning_params else "null"
logger.info(
f"\n━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\n"
f"[API HIT] POST /api/v1/optimization/riderassign\n"
- f"[CONFIG] Mode: {mode_str.upper()} | Hypertuning: {tuning_str} | Active Riders: {len(riders)}\n"
+ f"[CONFIG] Mode: {mode_str.upper()} | Active Riders: {len(riders)}\n"
f"━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━"
)
logger.info(f"[API] riderassign ▶ orders={len(orders)} riders={len(riders)} mode={mode_str}")
@@ -811,44 +726,10 @@ async def assign_orders_to_riders(
vrp_orders = [o for o in orders if id(o) not in history_pre_assigned_ids]
# 3. Run Assignment (AssignmentService)
- # -- Per-request strategy override (thread-safe via contextvars) --
- # Old approach mutated _cfg._cache directly — a race condition when two
- # concurrent requests set different strategies simultaneously.
- # New approach: each async task gets its OWN copy of the ContextVar,
- # so overrides never leak between concurrent requests.
- from app.config.dynamic_config import get_config, set_request_strategy
- from app.services.ml.behavior_analyzer import time_band as _tb, load_band as _lb
- from datetime import datetime as _DT
+ from app.config.dynamic_config import get_config
_cfg = get_config()
- # ── Bandit context (used both for selection + background update) ──
- _bandit_t_band = _tb(_DT.now().isoformat())
- _bandit_l_band = _lb(len(vrp_orders) / max(1, len(riders)))
- _bandit_context = (_bandit_t_band, _bandit_l_band)
-
- valid_strategies = ["balanced", "fuel_saver", "aggressive_speed", "zone_strict"]
- if hypertuning_params and hypertuning_params in valid_strategies:
- # Explicit caller override always wins
- set_request_strategy(hypertuning_params)
- logger.info(f"[HYPERTUNE] Per-request strategy override: {hypertuning_params}")
- else:
- # ── Thompson Sampling: let the bandit choose ──────────────────
- # The bandit samples from Beta posteriors for this (time, load)
- # context and picks the arm with highest sample. It naturally
- # explores when uncertain and exploits known-good strategies as
- # data accumulates. Replaces the one-size-fits-all greedy picker.
- try:
- from app.services.ml.strategy_bandit import get_bandit as _gb
- _bandit_strategy = _gb().select(_bandit_t_band, _bandit_l_band)
- set_request_strategy(_bandit_strategy)
- logger.info(
- f"[Bandit] Strategy='{_bandit_strategy}' "
- f"ctx={_bandit_t_band}|{_bandit_l_band}"
- )
- except Exception as _bse:
- logger.debug(f"[Bandit] Selection failed (non-fatal): {_bse}")
-
optimizer = RouteOptimizer()
# ── PHASE 3a: TRUE VRP (primary solver) ──────────────────────────────
@@ -863,7 +744,7 @@ async def assign_orders_to_riders(
if not do_reshuffle: # VRP not meaningful during reshuffle (intentional exploration)
try:
# Build minimal rider info for VRP
- from app.services.routing.kalman_filter import smooth_rider_locations
+ from app.services.routing.gps_smoother import smooth_rider_locations
_smooth_riders = smooth_rider_locations(list(riders))
from app.services.core.assignment_service import AssignmentService as _AS
@@ -1272,21 +1153,19 @@ async def assign_orders_to_riders(
risk_meta: dict = {"label": "n/a", "model_trained": False, "deprecated": True}
# ── BACKGROUND ML LOGGING ────────────────────────────────────────────
- # Fire-and-forget: log this assignment event to SQLite, auto-retrain
- # ID3 every 50 events, auto-tune strategy every 100 events.
+ # Fire-and-forget: log this assignment event to SQLite.
# Uses _ml_executor so the API response is never delayed.
try:
_elapsed_ms = (time.time() - _t0) * 1000
_hyp_snapshot = get_config().get_all() # frozen copy for this call
_ml_executor.submit(
- _bg_log_and_maybe_retrain,
+ _bg_log_assignment,
len(orders),
len(riders),
_hyp_snapshot,
{rid: list(ords) for rid, ords in assignments.items()},
len(unassigned_orders),
round(_elapsed_ms, 1),
- _bandit_context, # (time_band, load_band) for RL update
)
except Exception as _mle:
logger.debug(f"[ML BG] Submit failed (non-fatal): {_mle}")
@@ -1354,7 +1233,6 @@ async def assign_orders_to_riders(
},
"reshuffle_mode": do_reshuffle,
"solver_mode": "vrp_optimal" if used_vrp else "2phase_heuristic",
- "hypertuning_params": hypertuning_params or "default",
"faiss_coord_corrections": _faiss_corrected,
"faiss_customer_records": _faiss_store.record_count() if _faiss_store else 0,
"faiss_history_preassigned": _history_hits,
diff --git a/app/services/core/__pycache__/assignment_service.cpython-312.pyc b/app/services/core/__pycache__/assignment_service.cpython-312.pyc
index 12a190e..be421f6 100644
Binary files a/app/services/core/__pycache__/assignment_service.cpython-312.pyc and b/app/services/core/__pycache__/assignment_service.cpython-312.pyc differ
diff --git a/app/services/core/assignment_service.py b/app/services/core/assignment_service.py
index d68f6ac..bda2d87 100644
--- a/app/services/core/assignment_service.py
+++ b/app/services/core/assignment_service.py
@@ -6,110 +6,15 @@ from math import radians, cos, sin, asin, sqrt, ceil
from typing import List, Dict, Any, Optional, Set
from collections import defaultdict
from app.config.rider_preferences import RIDER_PREFERRED_KITCHENS, BLOCKED_RIDERS as _BLOCKED_RIDERS_SET
-from app.services.routing.kalman_filter import (
+from app.services.routing.gps_smoother import (
smooth_rider_locations,
smooth_order_coordinates,
)
-from app.config.dynamic_config import (
- get_config,
- get_kitchen_label_id as _get_kitchen_label_id,
- get_kitchen_frequency as _get_kitchen_frequency,
- update_kitchen_stats,
-)
-from app.services.ml.ml_data_collector import get_collector
+from app.config.dynamic_config import get_config
logger = logging.getLogger(__name__)
-class DataEncoder:
- """
- Data Encoding Utilities for ML-ready feature engineering.
- Implements techniques from idea.txt for categorical, spatial, and temporal data.
- """
-
- EARTH_RADIUS_KM = 6371
-
- @staticmethod
- def haversine(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
- """Calculate great circle distance between two points."""
- try:
- 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 * DataEncoder.EARTH_RADIUS_KM
- except:
- return 0.0
-
- @staticmethod
- def cyclic_encode_hour(hour: int) -> tuple[float, float]:
- """
- Cyclic time encoding - captures traffic patterns better than discrete buckets.
- hour_sin = sin(2π * hour / 24)
- hour_cos = cos(2π * hour / 24)
- """
- hour_sin = math.sin(2 * math.pi * hour / 24)
- hour_cos = math.cos(2 * math.pi * hour / 24)
- return hour_sin, hour_cos
-
- @staticmethod
- def cyclic_encode_day(day_of_week: int) -> tuple[float, float]:
- """
- Cyclic day encoding for weekly patterns.
- """
- day_sin = math.sin(2 * math.pi * day_of_week / 7)
- day_cos = math.cos(2 * math.pi * day_of_week / 7)
- return day_sin, day_cos
-
- @staticmethod
- def geohash_encode(lat: float, lon: float, precision: int = 7) -> str:
- """
- Geohash encoding for spatial data.
- Converts lat/lon to grid cell string for locality capture.
- precision=7 gives ~153m x 153m cells (good for delivery zones)
- """
- try:
- return _simple_geohash(lat, lon, precision)
- except:
- return "unknown"
-
-
-def _simple_geohash(lat: float, lon: float, precision: int = 7) -> str:
- """Standard geohash encoding — 5 bits per character, BASE32 output."""
- if lat == 0 or lon == 0:
- return "unknown"
- BASE32 = "0123456789bcdefghjkmnpqrstuvwxyz"
- 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(BASE32[char_bits])
- return "".join(hash_chars)
-
-
def _zone_key(lat: float, lon: float) -> str:
"""O(1) ~5 km grid cell key for zone proximity matching."""
if lat == 0 or lon == 0:
@@ -117,26 +22,6 @@ def _zone_key(lat: float, lon: float) -> str:
return f"{int(lat / 0.044)},{int(lon / 0.044)}"
-# Kitchen encoding now uses persistent DB storage from dynamic_config
-
-
-def update_kitchen_encoding(kitchen_name: str, profit: float = None):
- """Update kitchen stats in DB when orders are processed."""
- if kitchen_name and kitchen_name != "Unknown":
- if profit is not None:
- update_kitchen_stats(kitchen_name, profit)
-
-
-def get_kitchen_label_id(kitchen_name: str) -> int:
- """Get persistent label ID for a kitchen."""
- return _get_kitchen_label_id(kitchen_name)
-
-
-def get_kitchen_frequency(kitchen_name: str) -> float:
- """Get persistent frequency ratio for a kitchen."""
- return _get_kitchen_frequency(kitchen_name)
-
-
class AssignmentService:
def __init__(self):
# Curated config drives HARD kitchen ownership. Copy so substitution
@@ -196,35 +81,19 @@ class AssignmentService:
self.earth_radius_km = 6371
self._cfg = get_config()
- self._encoder = DataEncoder()
# Cost parameters for composite cost function
self._fuel_rate = 2.5 # Per km
self._base_rider_cost = 0.0
self._merchant_margin_avg = 5.0 # Default average margin
- # Profit encoding cache
- self._kitchen_profit_cache: Dict[str, List[float]] = defaultdict(list)
- self._profit_mean = 0.0
- self._profit_std = 1.0
-
def calculate_order_profit_features(
self, order: Dict[str, Any], distance_km: float
) -> Dict[str, float]:
"""
- Calculate engineered profit features for ML-ready data.
-
- Features:
- - profit: order amount - rider cost
- - profit_density: profit per kilometer (key signal!)
- - cost_efficiency: rider cost per estimated time
- - route_score: profit - composite cost (maximize this!)
- - encoded_geohash: spatial encoding
- - cyclic_time: hour_sin, hour_cos
+ Calculate profit and profit-density for one order, used as a scoring
+ signal (profit_bonus) when picking which rider gets a cluster.
"""
- features = {}
-
- # Extract order values
try:
order_amount = float(
order.get("orderamount") or order.get("deliveryamount") or 0
@@ -232,38 +101,17 @@ class AssignmentService:
except:
order_amount = 0.0
- # Calculate rider cost: base + (distance * fuel_rate)
+ # Rider cost: base + (distance * fuel_rate)
rider_cost = self._base_rider_cost + (distance_km * self._fuel_rate)
- features["rider_cost"] = rider_cost
# Profit = revenue - cost
profit = order_amount - rider_cost
- features["profit"] = profit
- # Profit density = profit / distance (HIGH SIGNAL feature!)
- # High density = profitable short deliveries
- if distance_km > 0:
- features["profit_density"] = profit / distance_km
- else:
- features["profit_density"] = 0.0
+ # Profit density = profit / distance — high density means profitable
+ # short deliveries; used to prioritise clusters worth serving.
+ profit_density = profit / distance_km if distance_km > 0 else 0.0
- # Cost efficiency = cost / time estimate (assuming 15 min per order average)
- estimated_time_min = max(15, distance_km * 4) # Rough estimate
- features["cost_efficiency"] = (
- rider_cost / estimated_time_min if estimated_time_min > 0 else 0
- )
-
- # Route score = profit - distance_cost (what we want to MAXIMIZE)
- # This is the core optimization target
- features["route_score"] = profit - (
- distance_km * 0.5
- ) # 0.5 = opportunity cost per km
-
- # Composite edge weight for optimizer (MINIMIZE this)
- # weight = cost - profit_margin_bonus
- features["composite_weight"] = rider_cost - (profit * 0.3) # 30% profit bonus
-
- return features
+ return {"profit": profit, "profit_density": profit_density}
def _load_config(self):
"""Load ML-tuned hyperparams fresh on every assignment call."""
@@ -478,7 +326,6 @@ class AssignmentService:
# Use caller-supplied pricing so dynamic API rates flow into scoring
self._fuel_rate = fuel_charge
self._base_rider_cost = base_pay
- _call_start = time.time()
# 0. Prep
assignments: Dict[int, List[Dict[str, Any]]] = defaultdict(list)
@@ -588,36 +435,14 @@ class AssignmentService:
orders, max_cluster_radius_km=self.MAX_KITCHEN_DISTANCE_KM
)
- # 2b. ENRICH CLUSTERS WITH PROFIT FEATURES (Data Encoding: Target + Frequency)
+ # 2b. Tag each cluster with the set of kitchen names it covers (used
+ # below for hard kitchen-ownership matching).
for cluster in clusters:
cluster["kitchen_names"] = set()
for order in cluster["orders"]:
- k_name = self.get_order_kitchen(order)
- cluster["kitchen_names"].add(k_name)
- update_kitchen_encoding(k_name)
+ cluster["kitchen_names"].add(self.get_order_kitchen(order))
- # Update kitchen profit cache for target encoding
- for order in cluster["orders"]:
- k_name = self.get_order_kitchen(order)
- profit = (
- float(order.get("orderamount") or order.get("deliveryamount") or 50)
- - 40
- )
- self._kitchen_profit_cache[k_name].append(profit)
-
- # Calculate profit statistics for normalization
- all_profits = [
- p for profits in self._kitchen_profit_cache.values() for p in profits
- ]
- if all_profits:
- self._profit_mean = sum(all_profits) / len(all_profits)
- if len(all_profits) > 1:
- variance = sum((p - self._profit_mean) ** 2 for p in all_profits) / len(
- all_profits
- )
- self._profit_std = variance**0.5
-
- logger.info(f"Created {len(clusters)} order clusters with profit encoding")
+ logger.info(f"Created {len(clusters)} order clusters")
# 2c. MINIMAL RIDER PRE-SELECTION
# Calculate the theoretical minimum number of riders needed so we don't
@@ -659,15 +484,6 @@ class AssignmentService:
cluster_geohash = _zone_key(centroid_lat, centroid_lon)
for order in cluster_orders:
- k_name = self.get_order_kitchen(order)
- # Target encoding: use average profit for this kitchen
- kitchen_profits = self._kitchen_profit_cache.get(k_name, [0])
- avg_kitchen_profit = (
- sum(kitchen_profits) / len(kitchen_profits)
- if kitchen_profits
- else 0
- )
-
o_lat = float(order.get("pickuplat", 0))
o_lon = float(order.get("pickuplon", 0))
dist = (
@@ -1004,19 +820,10 @@ class AssignmentService:
# 6. Commit State and History
self._post_process(assignments, rider_states, state_mgr)
- # 7. -- ML DATA COLLECTION -----------------------------------------
- try:
- elapsed_ms = (time.time() - _call_start) * 1000
- get_collector().log_assignment_event(
- num_orders=len(orders),
- num_riders=len(riders),
- hyperparams=self._cfg.get_all(),
- assignments=assignments,
- unassigned_count=len(unassigned_orders),
- elapsed_ms=elapsed_ms,
- )
- except Exception as _ml_err:
- logger.debug(f"ML logging skipped: {_ml_err}")
+ # ML event logging happens once, in the /riderassign endpoint after
+ # Phase-0 history merge + solo consolidation — not here — so every
+ # request produces exactly one assignment_ml_log row (see
+ # optimization.py::_bg_log_assignment).
# Log final distribution (use r_orders to avoid shadowing the outer `orders` list)
logger.info("=" * 50)
diff --git a/app/services/ml/__pycache__/ml_data_collector.cpython-312.pyc b/app/services/ml/__pycache__/ml_data_collector.cpython-312.pyc
index 63f726b..ea46a5c 100644
Binary files a/app/services/ml/__pycache__/ml_data_collector.cpython-312.pyc and b/app/services/ml/__pycache__/ml_data_collector.cpython-312.pyc differ
diff --git a/app/services/ml/behavior_analyzer.py b/app/services/ml/behavior_analyzer.py
deleted file mode 100644
index 0ade27b..0000000
--- a/app/services/ml/behavior_analyzer.py
+++ /dev/null
@@ -1,55 +0,0 @@
-"""
-Feature band encoders
-======================
-Discrete bucketers for assignment-context features (distance, time-of-day, load,
-order density). These are small pure helpers used by the Thompson-sampling
-strategy bandit (`strategy_bandit.py`) and the /riderassign bandit context.
-
-NOTE: The ID3 SUCCESS/RISK decision tree that used to live here has been retired —
-it only ever produced response metadata and never affected any assignment. Only
-the generic feature-band encoders remain.
-"""
-
-from datetime import datetime
-
-
-def distance_band(km: float) -> str:
- """Total route distance -> discrete band."""
- if km <= 5.0: return "SHORT"
- if km <= 15.0: return "MID"
- if km <= 30.0: return "LONG"
- return "VERY_LONG"
-
-
-def time_band(ts_str: str) -> str:
- """ISO timestamp -> time-of-day band."""
- try:
- hour = datetime.fromisoformat(ts_str).hour
- if 6 <= hour < 10: return "MORNING_RUSH"
- if 10 <= hour < 12: return "LATE_MORNING"
- if 12 <= hour < 14: return "LUNCH_RUSH"
- if 14 <= hour < 17: return "AFTERNOON"
- if 17 <= hour < 20: return "EVENING_RUSH"
- if 20 <= hour < 23: return "NIGHT"
- return "LATE_NIGHT"
- except Exception:
- return "UNKNOWN"
-
-
-def load_band(avg_load: float) -> str:
- """Average orders-per-rider -> load band."""
- if avg_load <= 2.0: return "LIGHT"
- if avg_load <= 5.0: return "MODERATE"
- if avg_load <= 8.0: return "HEAVY"
- return "OVERLOADED"
-
-
-def order_density_band(num_orders: int, num_riders: int) -> str:
- """Orders per available rider -> density band."""
- if num_riders == 0:
- return "NO_RIDERS"
- ratio = num_orders / num_riders
- if ratio <= 2.0: return "SPARSE"
- if ratio <= 5.0: return "NORMAL"
- if ratio <= 9.0: return "DENSE"
- return "OVERLOADED"
diff --git a/app/services/ml/ml_data_collector.py b/app/services/ml/ml_data_collector.py
index f83646d..79fd933 100644
--- a/app/services/ml/ml_data_collector.py
+++ b/app/services/ml/ml_data_collector.py
@@ -7,13 +7,12 @@ Key upgrades over the original
--------------------------------
1. FROZEN historical scores - quality_score is written ONCE at log time.
get_training_data() returns scores as-is from the DB (no retroactive mutation).
-2. Rich schema - zone_id, city_id, is_peak, weather_code,
- sla_breached, avg_delivery_time_min for richer features.
-3. SLA tracking - logs whether delivery SLA was breached.
-4. Analytics API - get_hourly_stats(), get_strategy_comparison(),
- get_quality_histogram(), get_zone_stats() for dashboard consumption.
-5. Thread-safe writes - connection-per-write pattern for FastAPI workers.
-6. Indexed columns - timestamp, ml_strategy, zone_id for fast queries.
+2. Rich schema - zone_id, city_id, is_peak, weather_code for
+ richer features.
+3. Analytics API - get_hourly_stats(), get_quality_histogram(),
+ get_zone_stats() for dashboard consumption.
+4. Thread-safe writes - connection-per-write pattern for FastAPI workers.
+5. Indexed columns - timestamp, zone_id for fast queries.
"""
import csv
@@ -45,7 +44,7 @@ class MLDataCollector:
Each log_assignment_event() call writes one row capturing:
- Operating context (time, orders, riders, zone, city)
- Active hyperparams (exact config snapshot for this call)
- - Measured outcomes (quality score, SLA, latency, distances)
+ - Measured outcomes (quality score, latency, distances)
quality_score is computed once and FROZEN - never retroactively changed.
"""
@@ -70,8 +69,6 @@ class MLDataCollector:
zone_id: str = "default",
city_id: str = "default",
weather_code: str = "CLEAR",
- sla_minutes: Optional[float] = None,
- avg_delivery_time_min: Optional[float] = None,
) -> None:
"""
Log one assignment event.
@@ -95,13 +92,8 @@ class MLDataCollector:
o for orders in assignments.values() if orders for o in orders
]
total_distance_km = sum(self._get_km(o) for o in all_orders)
- ml_strategy = hyperparams.get("ml_strategy", "balanced")
max_opr = hyperparams.get("max_orders_per_rider", 12)
- sla_breached = 0
- if sla_minutes and avg_delivery_time_min:
- sla_breached = int(avg_delivery_time_min > sla_minutes)
-
# Quality score - FROZEN at log time
quality_score = self._compute_quality_score(
num_orders=num_orders,
@@ -111,7 +103,6 @@ class MLDataCollector:
num_riders=num_riders,
total_distance_km=total_distance_km,
max_orders_per_rider=max_opr,
- ml_strategy=ml_strategy,
)
row = {
@@ -146,7 +137,6 @@ class MLDataCollector:
"search_time_limit_seconds", 5
),
"road_factor": hyperparams.get("road_factor", 1.3),
- "ml_strategy": ml_strategy,
"riders_used": riders_used,
"total_assigned": total_assigned,
"unassigned_count": unassigned_count,
@@ -154,8 +144,6 @@ class MLDataCollector:
"load_std": round(load_std, 3),
"total_distance_km": round(total_distance_km, 2),
"elapsed_ms": round(elapsed_ms, 1),
- "sla_breached": sla_breached,
- "avg_delivery_time_min": round(avg_delivery_time_min or 0.0, 2),
"quality_score": round(quality_score, 2),
}
@@ -171,7 +159,7 @@ class MLDataCollector:
except Exception as e:
logger.warning(f"[MLCollector] Logging failed (non-fatal): {e}")
- return 50.0 # neutral fallback so bandit update still fires
+ return 50.0 # neutral fallback
# ------------------------------------------------------------------
# Data retrieval for training
@@ -180,7 +168,6 @@ class MLDataCollector:
def get_training_data(
self,
min_records: int = 30,
- strategy_filter: Optional[str] = None,
since_hours: Optional[int] = None,
) -> Optional[List[Dict[str, Any]]]:
"""
@@ -195,9 +182,6 @@ class MLDataCollector:
params: list = []
clauses: list = []
- if strategy_filter:
- clauses.append("ml_strategy = ?")
- params.append(strategy_filter)
if since_hours:
cutoff = (datetime.utcnow() - timedelta(hours=since_hours)).isoformat()
clauses.append("timestamp >= ?")
@@ -253,7 +237,7 @@ class MLDataCollector:
return {"avg_quality": 0.0, "sample_size": 0, "history": []}
def get_hourly_stats(self, last_days: int = 7) -> List[Dict[str, Any]]:
- """Quality, SLA, and call volume aggregated by hour-of-day."""
+ """Quality and call volume aggregated by hour-of-day."""
try:
conn = sqlite3.connect(self._db_path)
cutoff = (datetime.utcnow() - timedelta(days=last_days)).isoformat()
@@ -263,8 +247,7 @@ class MLDataCollector:
COUNT(*) AS call_count,
AVG(quality_score) AS avg_quality,
AVG(unassigned_count) AS avg_unassigned,
- AVG(elapsed_ms) AS avg_latency_ms,
- SUM(CASE WHEN sla_breached=1 THEN 1 ELSE 0 END) AS sla_breaches
+ AVG(elapsed_ms) AS avg_latency_ms
FROM assignment_ml_log WHERE timestamp >= ?
GROUP BY hour ORDER BY hour
""",
@@ -278,7 +261,6 @@ class MLDataCollector:
"avg_quality": round(r[2] or 0.0, 2),
"avg_unassigned": round(r[3] or 0.0, 2),
"avg_latency_ms": round(r[4] or 0.0, 1),
- "sla_breaches": r[5],
}
for r in rows
]
@@ -286,42 +268,6 @@ class MLDataCollector:
logger.error(f"[MLCollector] get_hourly_stats: {e}")
return []
- def get_strategy_comparison(self) -> List[Dict[str, Any]]:
- """Compare quality metrics across ml_strategy values."""
- try:
- conn = sqlite3.connect(self._db_path)
- rows = conn.execute(
- """
- SELECT ml_strategy,
- COUNT(*) AS call_count,
- AVG(quality_score) AS avg_quality,
- MIN(quality_score) AS min_quality,
- MAX(quality_score) AS max_quality,
- AVG(unassigned_count) AS avg_unassigned,
- AVG(total_distance_km) AS avg_distance_km,
- AVG(elapsed_ms) AS avg_latency_ms
- FROM assignment_ml_log
- GROUP BY ml_strategy ORDER BY avg_quality DESC
- """
- ).fetchall()
- conn.close()
- return [
- {
- "strategy": r[0],
- "call_count": r[1],
- "avg_quality": round(r[2] or 0.0, 2),
- "min_quality": round(r[3] or 0.0, 2),
- "max_quality": round(r[4] or 0.0, 2),
- "avg_unassigned": round(r[5] or 0.0, 2),
- "avg_distance_km": round(r[6] or 0.0, 2),
- "avg_latency_ms": round(r[7] or 0.0, 1),
- }
- for r in rows
- ]
- except Exception as e:
- logger.error(f"[MLCollector] get_strategy_comparison: {e}")
- return []
-
def get_quality_histogram(self, bins: int = 10) -> List[Dict[str, Any]]:
"""Quality score distribution for histogram chart."""
try:
@@ -348,14 +294,13 @@ class MLDataCollector:
return []
def get_zone_stats(self) -> List[Dict[str, Any]]:
- """Quality and SLA stats grouped by zone."""
+ """Quality stats grouped by zone."""
try:
conn = sqlite3.connect(self._db_path)
rows = conn.execute(
"""
SELECT zone_id, COUNT(*) AS call_count,
AVG(quality_score) AS avg_quality,
- SUM(sla_breached) AS sla_breaches,
AVG(total_distance_km) AS avg_distance_km
FROM assignment_ml_log
GROUP BY zone_id ORDER BY avg_quality DESC
@@ -367,8 +312,7 @@ class MLDataCollector:
"zone_id": r[0],
"call_count": r[1],
"avg_quality": round(r[2] or 0.0, 2),
- "sla_breaches": r[3],
- "avg_distance_km": round(r[4] or 0.0, 2),
+ "avg_distance_km": round(r[3] or 0.0, 2),
}
for r in rows
]
@@ -385,17 +329,6 @@ class MLDataCollector:
except Exception:
return 0
- def count_by_strategy(self) -> Dict[str, int]:
- try:
- conn = sqlite3.connect(self._db_path)
- rows = conn.execute(
- "SELECT ml_strategy, COUNT(*) FROM assignment_ml_log GROUP BY ml_strategy"
- ).fetchall()
- conn.close()
- return {r[0]: r[1] for r in rows}
- except Exception:
- return {}
-
def export_csv(self) -> str:
"""Export all records as CSV string."""
try:
@@ -448,7 +381,6 @@ class MLDataCollector:
num_riders: int,
total_distance_km: float,
max_orders_per_rider: int,
- ml_strategy: str = "balanced",
) -> float:
"""
Multi-dimensional quality score (0–100, higher = better).
@@ -461,11 +393,7 @@ class MLDataCollector:
│ rider_efficiency │ reward using minimal riders for the batch size │
└──────────────────────┴────────────────────────────────────────────────┘
- Strategy weights (w_assign, w_dist, w_balance, w_efficiency):
- - balanced : (45, 20, 20, 15)
- - aggressive_speed: (70, 15, 0, 15) — care about assignment + efficiency
- - fuel_saver : (25, 60, 0, 15) — heavily penalise long routes
- - zone_strict : (35, 25, 25, 15) — balanced with zone awareness
+ One fixed weighting (45, 20, 20, 15) is used for every call.
"""
import math
if num_orders == 0:
@@ -488,14 +416,7 @@ class MLDataCollector:
min_riders_needed = max(1, math.ceil(num_orders / max_orders_per_rider))
rider_efficiency = min(1.0, min_riders_needed / max(1, riders_used))
- weights = {
- # assign dist balance efficiency
- "aggressive_speed": (70.0, 15.0, 0.0, 15.0),
- "fuel_saver": (25.0, 60.0, 0.0, 15.0),
- "zone_strict": (35.0, 25.0, 25.0, 15.0),
- "balanced": (45.0, 20.0, 20.0, 15.0),
- }
- w_comp, w_dist, w_bal, w_eff = weights.get(ml_strategy, (45.0, 20.0, 20.0, 15.0))
+ w_comp, w_dist, w_bal, w_eff = (45.0, 20.0, 20.0, 15.0)
return min(
assigned_ratio * w_comp
@@ -542,7 +463,6 @@ class MLDataCollector:
cluster_radius_km REAL,
search_time_limit_seconds INTEGER,
road_factor REAL,
- ml_strategy TEXT DEFAULT 'balanced',
riders_used INTEGER,
total_assigned INTEGER,
unassigned_count INTEGER,
@@ -550,8 +470,6 @@ class MLDataCollector:
load_std REAL,
total_distance_km REAL DEFAULT 0.0,
elapsed_ms REAL,
- sla_breached INTEGER DEFAULT 0,
- avg_delivery_time_min REAL DEFAULT 0.0,
quality_score REAL
)
""")
@@ -560,9 +478,6 @@ class MLDataCollector:
"ALTER TABLE assignment_ml_log ADD COLUMN zone_id TEXT DEFAULT 'default'",
"ALTER TABLE assignment_ml_log ADD COLUMN city_id TEXT DEFAULT 'default'",
"ALTER TABLE assignment_ml_log ADD COLUMN weather_code TEXT DEFAULT 'CLEAR'",
- "ALTER TABLE assignment_ml_log ADD COLUMN sla_breached INTEGER DEFAULT 0",
- "ALTER TABLE assignment_ml_log ADD COLUMN avg_delivery_time_min REAL DEFAULT 0.0",
- "ALTER TABLE assignment_ml_log ADD COLUMN ml_strategy TEXT DEFAULT 'balanced'",
"ALTER TABLE assignment_ml_log ADD COLUMN total_distance_km REAL DEFAULT 0.0",
]
for ddl in migrations:
@@ -572,7 +487,6 @@ class MLDataCollector:
pass
for idx in [
"CREATE INDEX IF NOT EXISTS idx_timestamp ON assignment_ml_log(timestamp)",
- "CREATE INDEX IF NOT EXISTS idx_strategy ON assignment_ml_log(ml_strategy)",
"CREATE INDEX IF NOT EXISTS idx_zone ON assignment_ml_log(zone_id)",
]:
conn.execute(idx)
diff --git a/app/services/ml/strategy_bandit.py b/app/services/ml/strategy_bandit.py
deleted file mode 100644
index 2c3cb51..0000000
--- a/app/services/ml/strategy_bandit.py
+++ /dev/null
@@ -1,277 +0,0 @@
-"""
-Thompson Sampling Contextual Bandit — Strategy Selector
-=========================================================
-Replaces the greedy SQL auto-tuner with proper online RL.
-
-Problem
--------
-The old greedy tuner picks the strategy with the highest *average* quality
-across all history. It never explores alternatives once one strategy leads,
-and it ignores context (the same strategy isn't best at all times of day
-and all load levels).
-
-Solution
---------
-A contextual multi-armed bandit with Thompson Sampling:
-
- Context = (time_band, load_band) — up to 7 × 4 = 28 states
- Arms = 4 strategies — balanced / fuel_saver / aggressive_speed / zone_strict
- Reward = quality_score / 100 — 0..1 float, already logged by MLDataCollector
-
-How Thompson Sampling works
----------------------------
-Each (context, arm) pair has a Beta(α, β) posterior where:
- α = sum of rewards seen so far (high quality calls push α up)
- β = sum of "anti-rewards" (low quality calls push β up)
-
-To SELECT a strategy:
- 1. For each arm, sample θ ~ Beta(α, β)
- 2. Pick arm with highest θ
- → Naturally balances exploration (uncertain arms get sampled often)
- with exploitation (well-known good arms dominate when confident)
-
-To UPDATE after an assignment:
- reward = quality_score / 100
- α += reward
- β += (1 - reward)
-
-Bootstrap
----------
-On first startup the entire historical SQLite DB is replayed to warm up
-posteriors so the bandit starts informed, not blank.
-If saved state already exists it is loaded from disk instead.
-
-Persistence
------------
-ml_data/strategy_bandit.json — saved every 10 updates.
-"""
-
-import json
-import logging
-import os
-import threading
-from typing import Dict, List, Optional, Tuple
-
-import numpy as np
-
-logger = logging.getLogger(__name__)
-
-_SAVE_PATH = os.getenv("BANDIT_PATH", "ml_data/strategy_bandit.json")
-_DB_PATH = os.getenv("ML_DB_PATH", "ml_data/ml_store.db")
-
-ARMS = ["balanced", "fuel_saver", "aggressive_speed", "zone_strict"]
-
-
-class ContextualBandit:
- """
- Thompson Sampling bandit for ml_strategy selection.
-
- Context key : "{time_band}|{load_band}"
- Arms : ARMS list (4 strategies)
- Prior : Beta(1, 1) — uniform, no initial preference
- """
-
- def __init__(self):
- self._lock = threading.Lock()
- # _posteriors[ctx_key][arm] = [alpha, beta]
- self._posteriors: Dict[str, Dict[str, List[float]]] = {}
- self._update_count = 0
- self._total_pulls = 0
- self._load()
- self._bootstrap_from_db()
-
- # ── Public API ───────────────────────────────────────────────────────────
-
- def select(self, time_band: str, load_band: str) -> str:
- """
- Sample from Beta posteriors and return the strategy with the highest
- sample. Explores uncertain arms naturally; exploits known-good arms
- as confidence grows.
- """
- ctx = self._ctx(time_band, load_band)
- with self._lock:
- samples = {
- arm: float(np.random.beta(*self._ab(ctx, arm)))
- for arm in ARMS
- }
- self._total_pulls += 1
- chosen = max(samples, key=samples.__getitem__)
- logger.debug(
- f"[Bandit] ctx={ctx} "
- f"samples={{{', '.join(f'{k}:{v:.3f}' for k, v in samples.items())}}} "
- f"→ {chosen}"
- )
- return chosen
-
- def update(self, time_band: str, load_band: str,
- strategy: str, quality_score: float) -> None:
- """Update the Beta posterior for (context, arm) after observing quality."""
- if strategy not in ARMS:
- return
- ctx = self._ctx(time_band, load_band)
- reward = min(1.0, max(0.0, float(quality_score) / 100.0))
- with self._lock:
- ab = self._ab(ctx, strategy)
- ab[0] += reward # alpha ← quality adds to success mass
- ab[1] += (1.0 - reward) # beta ← (1-quality) adds to failure mass
- self._update_count += 1
- should_save = self._update_count % 10 == 0
- if should_save:
- self._save()
-
- def best_arm(self, time_band: str, load_band: str) -> Tuple[str, float]:
- """
- Return the arm with the highest posterior mean (pure exploitation, no
- sampling noise). Used for logging and dashboard, not for live selection.
- """
- ctx = self._ctx(time_band, load_band)
- with self._lock:
- means = {
- arm: self._ab(ctx, arm)[0] / sum(self._ab(ctx, arm))
- for arm in ARMS
- }
- best = max(means, key=means.__getitem__)
- return best, round(means[best], 4)
-
- def get_stats(self) -> Dict:
- """Return per-context arm statistics for the ML admin dashboard."""
- with self._lock:
- stats: Dict[str, Dict] = {}
- for ctx, arms in sorted(self._posteriors.items()):
- stats[ctx] = {}
- for arm in ARMS:
- if arm not in arms:
- stats[ctx][arm] = {"mean_reward": 0.5, "observations": 0,
- "alpha": 1.0, "beta": 1.0}
- continue
- alpha, beta = arms[arm]
- n = alpha + beta - 2.0 # subtract the Beta(1,1) prior mass
- mean = alpha / (alpha + beta)
- stats[ctx][arm] = {
- "mean_reward": round(mean, 4),
- "observations": round(max(0, n), 1),
- "alpha": round(alpha, 2),
- "beta": round(beta, 2),
- }
- return {
- "total_pulls": self._total_pulls,
- "total_updates": self._update_count,
- "context_count": len(self._posteriors),
- "arms": ARMS,
- "contexts": stats,
- }
-
- # ── Persistence ──────────────────────────────────────────────────────────
-
- def _save(self) -> None:
- try:
- with self._lock:
- snapshot = {
- "posteriors": {
- ctx: {arm: list(ab) for arm, ab in arms.items()}
- for ctx, arms in self._posteriors.items()
- },
- "update_count": self._update_count,
- "total_pulls": self._total_pulls,
- }
- os.makedirs(os.path.dirname(_SAVE_PATH) or ".", exist_ok=True)
- with open(_SAVE_PATH, "w", encoding="utf-8") as f:
- json.dump(snapshot, f, indent=2)
- except Exception as e:
- logger.warning(f"[Bandit] Save failed: {e}")
-
- def _load(self) -> None:
- try:
- if not os.path.exists(_SAVE_PATH):
- logger.info("[Bandit] No saved state — will bootstrap from DB.")
- return
- with open(_SAVE_PATH, "r", encoding="utf-8") as f:
- data = json.load(f)
- self._posteriors = data.get("posteriors", {})
- self._update_count = data.get("update_count", 0)
- self._total_pulls = data.get("total_pulls", 0)
- logger.info(
- f"[Bandit] Loaded from disk — "
- f"{len(self._posteriors)} contexts, {self._update_count} updates"
- )
- except Exception as e:
- logger.warning(f"[Bandit] Load failed (starting fresh): {e}")
-
- # ── Bootstrap ────────────────────────────────────────────────────────────
-
- def _bootstrap_from_db(self) -> None:
- """
- Warm-up posteriors by replaying every historical assignment event.
- Skipped if the saved JSON already reflects current DB data.
- """
- if self._update_count > 0:
- logger.info(
- f"[Bandit] Already warmed ({self._update_count} updates) — "
- "skipping DB bootstrap."
- )
- return
- try:
- import sqlite3
- from app.services.ml.behavior_analyzer import (
- time_band as _tb,
- load_band as _lb,
- )
- conn = sqlite3.connect(_DB_PATH)
- rows = conn.execute(
- "SELECT timestamp, avg_load, ml_strategy, quality_score "
- "FROM assignment_ml_log "
- "WHERE ml_strategy IS NOT NULL AND quality_score IS NOT NULL "
- "ORDER BY id ASC"
- ).fetchall()
- conn.close()
-
- if not rows:
- logger.info("[Bandit] No historical data — starting with uniform priors.")
- return
-
- for ts, avg_load, strategy, quality in rows:
- t_band = _tb(str(ts)) if ts else "UNKNOWN"
- l_band = _lb(float(avg_load or 0))
- self.update(t_band, l_band, strategy, float(quality or 50))
-
- # Reset counter so we don't confuse bootstrap updates with live ones
- with self._lock:
- self._update_count = len(rows)
-
- logger.info(
- f"[Bandit] Bootstrapped from {len(rows)} historical events — "
- f"{len(self._posteriors)} contexts, {len(ARMS)} arms."
- )
- self._save()
- except Exception as e:
- logger.warning(f"[Bandit] Bootstrap failed (non-fatal): {e}")
-
- # ── Helpers ──────────────────────────────────────────────────────────────
-
- @staticmethod
- def _ctx(time_band: str, load_band: str) -> str:
- return f"{time_band}|{load_band}"
-
- def _ab(self, ctx: str, arm: str) -> List[float]:
- """Get or create Beta(1, 1) uniform prior for (ctx, arm). NOT thread-safe alone."""
- if ctx not in self._posteriors:
- self._posteriors[ctx] = {}
- if arm not in self._posteriors[ctx]:
- self._posteriors[ctx][arm] = [1.0, 1.0]
- return self._posteriors[ctx][arm]
-
-
-# ── Singleton ────────────────────────────────────────────────────────────────
-
-_bandit_instance: Optional[ContextualBandit] = None
-_bandit_lock = threading.Lock()
-
-
-def get_bandit() -> ContextualBandit:
- """Return the process-level singleton ContextualBandit."""
- global _bandit_instance
- if _bandit_instance is None:
- with _bandit_lock:
- if _bandit_instance is None:
- _bandit_instance = ContextualBandit()
- return _bandit_instance
diff --git a/app/services/rider/__pycache__/get_active_riders.cpython-312.pyc b/app/services/rider/__pycache__/get_active_riders.cpython-312.pyc
new file mode 100644
index 0000000..a759432
Binary files /dev/null and b/app/services/rider/__pycache__/get_active_riders.cpython-312.pyc differ
diff --git a/app/services/rider/__pycache__/rider_history_service.cpython-312.pyc b/app/services/rider/__pycache__/rider_history_service.cpython-312.pyc
new file mode 100644
index 0000000..e609b61
Binary files /dev/null and b/app/services/rider/__pycache__/rider_history_service.cpython-312.pyc differ
diff --git a/app/services/rider/__pycache__/rider_state_manager.cpython-312.pyc b/app/services/rider/__pycache__/rider_state_manager.cpython-312.pyc
new file mode 100644
index 0000000..5ae4a62
Binary files /dev/null and b/app/services/rider/__pycache__/rider_state_manager.cpython-312.pyc differ
diff --git a/app/services/rider/__pycache__/substitution_service.cpython-312.pyc b/app/services/rider/__pycache__/substitution_service.cpython-312.pyc
new file mode 100644
index 0000000..8a81abb
Binary files /dev/null and b/app/services/rider/__pycache__/substitution_service.cpython-312.pyc differ
diff --git a/app/services/routing/__pycache__/clustering_service.cpython-312.pyc b/app/services/routing/__pycache__/clustering_service.cpython-312.pyc
index 0760122..cd3b9ac 100644
Binary files a/app/services/routing/__pycache__/clustering_service.cpython-312.pyc and b/app/services/routing/__pycache__/clustering_service.cpython-312.pyc differ
diff --git a/app/services/routing/__pycache__/delivery_history_service.cpython-312.pyc b/app/services/routing/__pycache__/delivery_history_service.cpython-312.pyc
index 4952567..cc84f04 100644
Binary files a/app/services/routing/__pycache__/delivery_history_service.cpython-312.pyc and b/app/services/routing/__pycache__/delivery_history_service.cpython-312.pyc differ
diff --git a/app/services/routing/__pycache__/gps_smoother.cpython-312.pyc b/app/services/routing/__pycache__/gps_smoother.cpython-312.pyc
new file mode 100644
index 0000000..979b062
Binary files /dev/null and b/app/services/routing/__pycache__/gps_smoother.cpython-312.pyc differ
diff --git a/app/services/routing/__pycache__/realistic_eta_calculator.cpython-312.pyc b/app/services/routing/__pycache__/realistic_eta_calculator.cpython-312.pyc
index 29d666e..c5af52c 100644
Binary files a/app/services/routing/__pycache__/realistic_eta_calculator.cpython-312.pyc and b/app/services/routing/__pycache__/realistic_eta_calculator.cpython-312.pyc differ
diff --git a/app/services/routing/__pycache__/route_optimizer.cpython-312.pyc b/app/services/routing/__pycache__/route_optimizer.cpython-312.pyc
index e0a1a4a..5353906 100644
Binary files a/app/services/routing/__pycache__/route_optimizer.cpython-312.pyc and b/app/services/routing/__pycache__/route_optimizer.cpython-312.pyc differ
diff --git a/app/services/routing/__pycache__/zone_service.cpython-312.pyc b/app/services/routing/__pycache__/zone_service.cpython-312.pyc
index 6d63667..98ea58b 100644
Binary files a/app/services/routing/__pycache__/zone_service.cpython-312.pyc and b/app/services/routing/__pycache__/zone_service.cpython-312.pyc differ
diff --git a/app/services/routing/delivery_history_service.py b/app/services/routing/delivery_history_service.py
index cc67927..5a7810e 100644
--- a/app/services/routing/delivery_history_service.py
+++ b/app/services/routing/delivery_history_service.py
@@ -444,6 +444,37 @@ class DeliveryHistoryService:
cols = ["deliveryid", "userid", "pickupcustomer", "pickuptime", "deliverytime", "dlat", "dlon", "plat", "plon"]
return [dict(zip(cols, r)) for r in rows]
+ def get_pattern_records(self, days: int) -> List[Dict[str, Any]]:
+ """
+ Shape the local nearledb mirror into the record format the delivery-
+ history pattern store (Phase-0 kitchen+zone rider lookup) expects:
+ kitchen/pickuplat/pickuplon/deliverylat/deliverylong/userid/ridername.
+
+ `ridername` is always "" — delivery_raw doesn't store it (it's
+ display-only in the /ml/history debug endpoint; every actual matching
+ decision is keyed on userid, so this doesn't affect assignment).
+ """
+ records: List[Dict[str, Any]] = []
+ for r in self._load_raw_rows(days):
+ try:
+ plat = float(r.get("plat") or 0)
+ plon = float(r.get("plon") or 0)
+ dlat = float(r.get("dlat") or 0)
+ dlon = float(r.get("dlon") or 0)
+ uid = int(float(r.get("userid") or 0))
+ if not plat or not dlat or uid == 0:
+ continue
+ records.append({
+ "kitchen": (r.get("pickupcustomer") or "").strip().lower(),
+ "pickuplat": plat, "pickuplon": plon,
+ "deliverylat": dlat, "deliverylong": dlon,
+ "userid": uid,
+ "ridername": "",
+ })
+ except (TypeError, ValueError):
+ continue
+ return records
+
def sample_batches(self, days: int = 14, min_drops: int = 4, max_drops: int = 15,
limit: int = 10) -> List[Dict[str, Any]]:
"""
@@ -623,12 +654,24 @@ class DeliveryHistoryService:
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).
+ aggregates locally -> (if pattern_source=db) rebuild the Phase-0
+ pattern store from the same synced rows. This is what the scheduler
+ and the admin endpoint call. The DB is touched only by the sync step
+ (new rows only).
"""
+ from app.config.dynamic_config import get_config
+ cfg = get_config()
+ pattern_on = str(cfg.get("pattern_source", "csv")) == "db"
+ pattern_days = int(cfg.get("pattern_history_days", 30))
+ # Local retention must cover whichever consumer needs more history
+ # (the ETA window vs. the pattern-store window) without changing the
+ # ETA computation's own `days` window below — that backtest result is
+ # already validated at 14 days and this must not perturb it.
+ retain_days = max(days, pattern_days) if pattern_on else days
+
with _WRITE_LOCK:
try:
- sync = self.sync_from_db(days=days, tenant_id=tenant_id, full=full)
+ sync = self.sync_from_db(days=retain_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.
@@ -636,9 +679,23 @@ class DeliveryHistoryService:
self._last_summary = {"status": "sync_failed", "error": str(e), "rebuild": rebuilt}
return self._last_summary
- self._prune_raw(days)
+ self._prune_raw(retain_days)
rebuilt = self.rebuild_aggregates(days)
- self._last_summary = {"status": "ok", "sync": sync, "rebuild": rebuilt}
+
+ pattern_rebuild = None
+ if pattern_on:
+ try:
+ from app.services.vector.delivery_history_store import get_delivery_history_store
+ records = self.get_pattern_records(pattern_days)
+ n = get_delivery_history_store().rebuild_from_records(records)
+ pattern_rebuild = {"records": n, "days": pattern_days}
+ except Exception as e:
+ logger.warning(f"[DeliveryHistory] pattern-store rebuild skipped: {e}")
+
+ self._last_summary = {
+ "status": "ok", "sync": sync, "rebuild": rebuilt,
+ "pattern_rebuild": pattern_rebuild,
+ }
return self._last_summary
# -- cache + lookup -----------------------------------------------------
diff --git a/app/services/routing/gps_smoother.py b/app/services/routing/gps_smoother.py
new file mode 100644
index 0000000..e71eae4
--- /dev/null
+++ b/app/services/routing/gps_smoother.py
@@ -0,0 +1,150 @@
+"""
+GPS location smoothing — rider-api
+
+Smooths noisy rider GPS pings (typical error +-5-15m, worse on poor signal,
+occasional bad-fix "jumps") using a per-rider exponential moving average
+(EMA): each new reading is blended with the running estimate so a single bad
+ping can't yank the rider's position, while the estimate still tracks real
+movement.
+
+This used to be implemented as a full Kalman filter (per-coordinate process/
+measurement covariance, gain computed every update). For a "constant
+position" state model with no velocity term — which is what this is, since
+we're smoothing noisy pings, not tracking motion — the Kalman update reduces
+mathematically to an EMA once the gain reaches steady state, which happens
+within the first couple of updates. The EMA below is the same behavior with
+one constant instead of two, and no covariance bookkeeping to explain to the
+next person reading this file.
+
+Only two things are actually used elsewhere in the app:
+ smooth_rider_locations(riders) — per-rider EMA, stateful across calls
+ smooth_order_coordinates(orders) — validates/normalises delivery coords
+ (NOT smoothed — see its docstring)
+"""
+
+import logging
+import time
+from typing import Dict, Optional, Tuple
+
+logger = logging.getLogger(__name__)
+
+# Smoothing factor: how much weight a new GPS reading gets against the
+# running estimate. Lower = smoother/slower to react, higher = trusts each
+# new ping more. 0.1 matches the steady-state behavior of the Kalman filter
+# this replaced (process_noise=1e-4, measurement_noise=0.01).
+_ALPHA = 0.1
+_STALE_SECONDS = 1800.0 # reset a rider's running estimate after 30 min silence
+
+
+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
+
+
+class _RiderEstimate:
+ """Running EMA estimate of one rider's position."""
+
+ def __init__(self):
+ self.lat: Optional[float] = None
+ self.lon: Optional[float] = None
+ self.last_updated: float = time.time()
+
+ def update(self, lat: float, lon: float) -> Tuple[float, float]:
+ if not _is_valid_coord(lat, lon):
+ return (self.lat, self.lon) if self.lat is not None else (lat, lon)
+
+ if time.time() - self.last_updated > _STALE_SECONDS:
+ self.lat = self.lon = None # stale — start fresh
+
+ if self.lat is None:
+ self.lat, self.lon = lat, lon
+ else:
+ self.lat += _ALPHA * (lat - self.lat)
+ self.lon += _ALPHA * (lon - self.lon)
+
+ self.last_updated = time.time()
+ return self.lat, self.lon
+
+
+_rider_estimates: Dict[str, _RiderEstimate] = {}
+
+
+def smooth_rider_locations(riders: list) -> list:
+ """
+ Apply EMA smoothing to a list of rider dicts in-place, keyed by rider id
+ (history preserved across calls via a process-level registry).
+
+ Reads/writes: latitude, longitude (and currentlat/currentlong if present).
+ Adds: _location_smoothed = True on each processed rider.
+ """
+ for rider in riders:
+ try:
+ rider_id = str(
+ 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
+
+ estimate = _rider_estimates.setdefault(rider_id, _RiderEstimate())
+ smooth_lat, smooth_lon = estimate.update(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["_location_smoothed"] = True
+ except Exception as e:
+ logger.debug(f"Rider location 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 these are NOT smoothed:
+ Smoothing blends successive measurements from the same source over time.
+ Delivery coordinates are a single static point (one measurement) — there
+ is nothing to blend. 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 _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
diff --git a/app/services/routing/kalman_filter.py b/app/services/routing/kalman_filter.py
deleted file mode 100644
index 608ac6f..0000000
--- a/app/services/routing/kalman_filter.py
+++ /dev/null
@@ -1,327 +0,0 @@
-"""
-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
diff --git a/app/services/routing/route_optimizer.py b/app/services/routing/route_optimizer.py
index fdc6580..1223b54 100644
--- a/app/services/routing/route_optimizer.py
+++ b/app/services/routing/route_optimizer.py
@@ -10,7 +10,6 @@ FEATURES:
- Automatic outlier detection and coordinate correction
- Hybrid distance calculation (Google Maps + Haversine fallback)
- Robust error handling for invalid inputs
-- Composite cost function (idea.txt: distance + profit - margin)
"""
import math
@@ -21,7 +20,7 @@ import asyncio
from typing import Dict, Any, List as _List, Optional, Tuple, Union
from datetime import datetime, timedelta
import httpx
-from app.services.routing.kalman_filter import smooth_order_coordinates
+from app.services.routing.gps_smoother import smooth_order_coordinates
import numpy as np
from app.core.arrow_utils import calculate_haversine_matrix_vectorized
from app.config.dynamic_config import get_config
@@ -38,191 +37,6 @@ except ImportError:
logger = logging.getLogger(__name__)
-class CompositeCostCalculator:
- """
- Composite Cost Function for Profit-Aware Route Optimization.
-
- Based on idea.txt data encoding techniques:
- - Total Cost = Distance Cost + Rider Cost - Merchant Profit
- - Edge Cost = (distance * fuel_rate) + rider_cost - merchant_margin
- - route_score = profit - composite_cost (what we want to MAXIMIZE)
-
- This transforms the problem from:
- "Find shortest route" -> "Find most profitable route"
- """
-
- def __init__(self):
- # Cost parameters (can be ML-tuned via DynamicConfig)
- self.fuel_rate_per_km = 2.5
- self.base_rider_cost = 0.0
- self.opportunity_cost_per_km = 0.5 # Cost of rider's time per km
- self.profit_weight = 0.3 # How much profit influences routing (0-1)
- self.traffic_multiplier_peak = 1.5 # Peak hour traffic penalty
- self.traffic_multiplier_normal = 1.2 # Normal traffic multiplier
-
- # Defaults for orders without profit data
- self.default_order_amount = 80.0
- self.default_merchant_margin = 5.0
-
- def calculate_composite_cost(
- self,
- distance_km: float,
- order_amount: float = None,
- merchant_margin: float = None,
- traffic_factor: float = 1.0,
- time_of_day: str = "NORMAL",
- ) -> Dict[str, float]:
- """
- Calculate composite cost for a route edge.
-
- Args:
- distance_km: Distance for this leg
- order_amount: Revenue from this order (if known)
- merchant_margin: Merchant's margin for this order (if known)
- traffic_factor: Traffic multiplier (1.0 = normal)
- time_of_day: Traffic time category ("PEAK", "NORMAL", "OFF_PEAK")
-
- Returns:
- Dict with:
- - distance_cost: Raw distance cost
- - rider_cost: Total rider cost for this leg
- - gross_profit: Revenue - rider cost
- - net_cost: Cost after profit adjustment (MINIMIZE THIS)
- - route_score: Profitability score (MAXIMIZE THIS)
- """
- # Distance cost = fuel + opportunity cost
- distance_cost = distance_km * self.fuel_rate_per_km
-
- # Rider cost = base + distance cost
- rider_cost = self.base_rider_cost + distance_cost
-
- # Apply traffic penalty
- if time_of_day == "PEAK":
- rider_cost *= self.traffic_multiplier_peak
- elif time_of_day == "NORMAL":
- rider_cost *= self.traffic_multiplier_normal
-
- # Apply custom traffic factor
- rider_cost *= traffic_factor
-
- # Profit calculation (target encoding: use order data if available)
- if order_amount is None:
- order_amount = self.default_order_amount
- if merchant_margin is None:
- merchant_margin = self.default_merchant_margin
-
- gross_profit = order_amount - rider_cost
-
- # Net cost = rider cost - profit contribution
- # This means high-profit orders have LOWER cost (more desirable)
- profit_adjustment = gross_profit * self.profit_weight
- net_cost = rider_cost - profit_adjustment
-
- # Route score = profit - opportunity cost (for route planning)
- route_score = gross_profit - (distance_km * self.opportunity_cost_per_km)
-
- # Ensure net_cost is never negative (minimum cost for any delivery)
- net_cost = max(net_cost, 5.0) # Minimum 5 km equivalent cost
-
- return {
- "distance_cost": round(distance_cost, 2),
- "rider_cost": round(rider_cost, 2),
- "gross_profit": round(gross_profit, 2),
- "net_cost": round(net_cost, 2),
- "route_score": round(route_score, 2),
- }
-
- def calculate_cost_matrix(
- self,
- dist_matrix: np.ndarray,
- orders: _List[Dict[str, Any]] = None,
- traffic_condition: str = "NORMAL",
- ) -> Tuple[np.ndarray, Dict[str, Any]]:
- """
- Calculate composite cost matrix for all node pairs.
-
- Args:
- dist_matrix: Distance matrix (N x N)
- orders: List of orders (for profit data)
- traffic_condition: Traffic condition ("PEAK", "NORMAL", "OFF_PEAK")
-
- Returns:
- Tuple of (cost_matrix, summary_stats)
- """
- n = len(dist_matrix)
- cost_matrix = np.zeros((n, n))
-
- # Extract order data for profit encoding
- order_amounts = []
- merchant_margins = []
-
- if orders:
- for o in orders:
- try:
- amount = float(
- o.get("orderamount")
- or o.get("deliveryamount")
- or self.default_order_amount
- )
- margin = float(
- o.get("merchant_margin") or self.default_merchant_margin
- )
- except:
- amount = self.default_order_amount
- margin = self.default_merchant_margin
- order_amounts.append(amount)
- merchant_margins.append(margin)
- else:
- order_amounts = [self.default_order_amount] * n
- merchant_margins = [self.default_merchant_margin] * n
-
- # Calculate costs for each pair
- total_cost = 0.0
- total_profit = 0.0
- high_cost_count = 0
-
- for i in range(n):
- for j in range(n):
- if i == j:
- cost_matrix[i][j] = 0
- continue
-
- dist = dist_matrix[i][j]
-
- # Use order j's profit data (destination)
- order_amount = (
- order_amounts[j - 1] if j > 0 else self.default_order_amount
- )
- merchant_margin = (
- merchant_margins[j - 1] if j > 0 else self.default_merchant_margin
- )
-
- cost_data = self.calculate_composite_cost(
- distance_km=dist,
- order_amount=order_amount,
- merchant_margin=merchant_margin,
- time_of_day=traffic_condition,
- )
-
- cost_matrix[i][j] = cost_data["net_cost"]
- total_cost += cost_data["net_cost"]
- total_profit += cost_data["gross_profit"]
-
- if cost_data["net_cost"] > 50:
- high_cost_count += 1
-
- summary = {
- "total_cost": round(total_cost, 2),
- "total_profit": round(total_profit, 2),
- "avg_cost": round(total_cost / (n * n) if n > 0 else 0, 2),
- "avg_profit": round(total_profit / (n * n) if n > 0 else 0, 2),
- "high_cost_legs": high_cost_count,
- "traffic_condition": traffic_condition,
- }
-
- return cost_matrix, summary
-
-
class RouteOptimizer:
"""Route optimization using Google OR-Tools (Async)."""
@@ -253,8 +67,6 @@ class RouteOptimizer:
# Solver time limit (ML-tuned)
self.search_time_limit_seconds = int(_cfg.get("search_time_limit_seconds"))
- self.cost_calculator = CompositeCostCalculator()
-
def haversine_distance(
self, lat1: float, lon1: float, lat2: float, lon2: float
) -> float:
@@ -274,153 +86,6 @@ class RouteOptimizer:
except Exception:
return 0.0
- async def _get_google_maps_distances_batch(
- self, origin_lat: float, origin_lon: float, destinations: _List[tuple]
- ) -> Dict[tuple, float]:
- """Get road distances for multiple destinations from Google Maps API. (Async, Parallel)"""
- if not self.use_google_maps or not destinations:
- return {}
-
- results = {}
- batch_size = 25
- chunks = [
- destinations[i : i + batch_size]
- for i in range(0, len(destinations), batch_size)
- ]
-
- async def process_batch(batch):
- batch_result = {}
- try:
- dest_str = "|".join([f"{lat},{lon}" for lat, lon in batch])
- url = "https://maps.googleapis.com/maps/api/distancematrix/json"
- params = {
- "origins": f"{origin_lat},{origin_lon}",
- "destinations": dest_str,
- "key": self.google_maps_api_key,
- "units": "metric",
- }
- async with httpx.AsyncClient(timeout=10.0) as client:
- response = await client.get(url, params=params)
- response.raise_for_status()
- data = response.json()
-
- if data.get("status") == "OK":
- rows = data.get("rows", [])
- if rows:
- elements = rows[0].get("elements", [])
- for idx, element in enumerate(elements):
- if idx < len(batch):
- dest_coord = batch[idx]
- if element.get("status") == "OK":
- dist = element.get("distance", {}).get("value")
- dur = element.get("duration", {}).get("value")
- if dist is not None:
- batch_result[dest_coord] = {
- "distance": dist / 1000.0,
- "duration": dur / 60.0 if dur else None,
- }
- except Exception as e:
- logger.warning(f"Google Maps batch call failed: {e}")
- return batch_result
-
- batch_results_list = await asyncio.gather(
- *[process_batch(chunk) for chunk in chunks]
- )
- for res in batch_results_list:
- results.update(res)
- return results
-
- # ------------------------------------------------------------------
- # GOOGLE DIRECTIONS - WAYPOINT OPTIMISATION
- # ------------------------------------------------------------------
-
- async def _optimize_waypoints_google(
- self,
- origin_lat: float,
- origin_lon: float,
- waypoints: _List[Tuple[float, float]],
- ) -> Tuple[Optional[_List[int]], Optional[_List[float]]]:
- """
- Ask Google Directions API to find the optimal visiting order for a set
- of delivery points starting from a kitchen/pickup location.
-
- Uses `optimize:true` in the waypoints parameter - Google solves the TSP
- internally using actual road geometry (turn restrictions, one-way
- streets, real distances) rather than Haversine approximation.
-
- Returns
- -------
- (waypoint_order, leg_km)
- waypoint_order 0-based indices into `waypoints` in optimal order.
- e.g. [2, 0, 1] means visit wp[2] -> wp[0] -> wp[1].
- leg_km Actual road distance (km) for each leg in order:
- leg_km[0] = kitchen -> wp[order[0]],
- leg_km[1] = wp[order[0]] -> wp[order[1]], etc.
- Both are None on any failure - caller falls back to OR-Tools.
-
- Notes
- -----
- - Supports up to 25 intermediate waypoints (Google's standard limit).
- - destination = origin (closed-loop TSP); the return leg is discarded.
- - One API call per rider per assignment - cheap at delivery scale.
- """
- if not self.use_google_maps or not waypoints or len(waypoints) < 2:
- return None, None
- if len(waypoints) > 25:
- return None, None # fall back to OR-Tools for unusually large routes
-
- try:
- wp_str = "optimize:true|" + "|".join(
- f"{lat},{lon}" for lat, lon in waypoints
- )
- params = {
- "origin": f"{origin_lat},{origin_lon}",
- "destination": f"{origin_lat},{origin_lon}", # closed loop
- "waypoints": wp_str,
- "key": self.google_maps_api_key,
- }
- async with httpx.AsyncClient(timeout=8.0) as client:
- resp = await client.get(
- "https://maps.googleapis.com/maps/api/directions/json",
- params=params,
- )
- resp.raise_for_status()
- data = resp.json()
-
- status_code = data.get("status")
- if status_code != "OK":
- logger.debug(
- f"[GoogleWaypoints] status={status_code} "
- f"error='{data.get('error_message', '')}'"
- )
- return None, None
-
- routes = data.get("routes", [])
- if not routes:
- return None, None
-
- route = routes[0]
- wp_order = route.get("waypoint_order")
- legs = route.get("legs", [])
-
- if wp_order is None or len(wp_order) != len(waypoints):
- return None, None
-
- # Extract leg distances (metres -> km), skip the return-to-origin leg
- leg_km: _List[float] = []
- for leg in legs[: len(waypoints)]: # first N legs only
- dist_m = leg.get("distance", {}).get("value")
- leg_km.append(dist_m / 1000.0 if dist_m is not None else 0.0)
-
- logger.debug(
- f"[GoogleWaypoints] Optimised {len(waypoints)} stops -> order={wp_order}"
- )
- return wp_order, leg_km
-
- except Exception as _e:
- logger.debug(f"[GoogleWaypoints] Failed (non-fatal): {_e}")
- return None, None
-
# ------------------------------------------------------------------
# ROAD-AWARE VISITING ORDER (Phase 2 - opt-in, cached)
# ------------------------------------------------------------------
@@ -698,8 +363,8 @@ class RouteOptimizer:
routing_enums_pb2.LocalSearchMetaheuristic.GUIDED_LOCAL_SEARCH
)
# TSP time limit hard-capped at 2 seconds per kitchen.
- # The ML hypertuner may push search_time_limit_seconds up to 8-10s
- # chasing marginally better routes, but at delivery scale (< 15 stops)
+ # search_time_limit_seconds can be tuned up to 8-10s via config, but at
+ # delivery scale (< 15 stops)
# OR-Tools finds a near-optimal solution in < 200ms. Waiting 8-10s
# per kitchen x 3 kitchens x 4 riders = 96s of unnecessary waiting.
# The VRP already has its own 3s cap. This cap applies to per-rider
diff --git a/app/services/vector/__pycache__/delivery_history_store.cpython-312.pyc b/app/services/vector/__pycache__/delivery_history_store.cpython-312.pyc
index b184cb3..91dba3e 100644
Binary files a/app/services/vector/__pycache__/delivery_history_store.cpython-312.pyc and b/app/services/vector/__pycache__/delivery_history_store.cpython-312.pyc differ
diff --git a/app/services/vector/delivery_history_store.py b/app/services/vector/delivery_history_store.py
index 0a6519d..359819b 100644
--- a/app/services/vector/delivery_history_store.py
+++ b/app/services/vector/delivery_history_store.py
@@ -1,12 +1,21 @@
"""
Delivery History Store
======================
-Pattern-first vector rider lookup using 30-day delivery history.
+Pattern-first vector rider lookup using recent delivery history.
Works on any platform — no faiss-cpu dependency.
+Data source (config `pattern_source`)
+--------------------------------------
+ "csv" (legacy) — built from delivery_details.csv, only refreshed when a
+ human re-exports it and calls POST /ml/reload-history.
+ "db" (default going forward) — built from the nearledb mirror the ETA-sync
+ agent already maintains (delivery_raw), rebuilt
+ automatically every sync cycle via rebuild_from_records().
+ No manual step, no extra database load.
+
How it works
------------
-At startup the CSV is parsed once and two structures are built:
+Records (however sourced) are turned into two structures:
1. PATTERN TABLE (primary, O(1) lookup)
City divided into ~1.1 km grid cells (round coords to 2 d.p.).
@@ -65,9 +74,21 @@ CORRECTIONS_PATH = os.getenv("DELIVERY_CORRECTIONS_CSV", "delivery_corrections.c
_CORRECTION_WEIGHT = int(os.getenv("CORRECTION_WEIGHT", "1"))
_STORE_DIR = os.getenv("FAISS_HISTORY_DIR", "ml_data/faiss_history")
-_VECTORS_PATH = os.path.join(_STORE_DIR, "delivery_history_vectors.npy")
-_RECORDS_PATH = os.path.join(_STORE_DIR, "delivery_history_records.pkl")
-_META_PATH = os.path.join(_STORE_DIR, "delivery_history.meta")
+
+
+def _paths_for(source: str) -> Tuple[str, str, str]:
+ """
+ Disk cache paths, namespaced by source ("csv" or "db"). Namespaced so that
+ flipping `pattern_source` at runtime (e.g. to debug, or to roll back)
+ never clobbers the other mode's cached snapshot — each mode keeps its own
+ independent copy on disk.
+ """
+ suffix = "" if source == "csv" else f".{source}"
+ return (
+ os.path.join(_STORE_DIR, f"delivery_history_vectors{suffix}.npy"),
+ os.path.join(_STORE_DIR, f"delivery_history_records{suffix}.pkl"),
+ os.path.join(_STORE_DIR, f"delivery_history{suffix}.meta"),
+ )
# ---------------------------------------------------------------------------
# Thresholds
@@ -78,6 +99,15 @@ _MIN_PATTERN_DOMINANCE = 0.60 # pattern table: rider owns ≥ 60 % of zone
_MIN_PATTERN_VOLUME = 3 # pattern table: at least 3 deliveries in zone
+def _pattern_source() -> str:
+ """'csv' (legacy, manual refresh) or 'db' (auto-refreshed from nearledb mirror)."""
+ try:
+ from app.config.dynamic_config import get_config
+ return str(get_config().get("pattern_source", "csv"))
+ except Exception:
+ return "csv"
+
+
# ---------------------------------------------------------------------------
# Pure-NumPy L2 index — drop-in replacement for faiss.IndexFlatL2
# ---------------------------------------------------------------------------
@@ -152,6 +182,7 @@ class DeliveryHistoryStore:
store.pattern_count() -> int
store.get_pattern_stats() -> list
store.reload_from_csv() -> int
+ store.rebuild_from_records(records) -> int
"""
def __init__(self):
@@ -174,8 +205,21 @@ class DeliveryHistoryStore:
def _load(self) -> None:
os.makedirs(_STORE_DIR, exist_ok=True)
+ if _pattern_source() == "db":
+ # DB-driven: the ETA-sync agent keeps this fresh on its own cadence
+ # by calling rebuild_from_records() after every sync cycle. At
+ # startup we just load whatever was last persisted — no CSV
+ # freshness check applies in this mode.
+ if self._load_from_disk("db"):
+ return
+ logger.info(
+ "[DeliveryHistory] pattern_source=db, no persisted snapshot yet "
+ "— will populate on the next ETA-sync cycle."
+ )
+ return
+
if self._saved_files_are_current():
- if self._load_from_disk():
+ if self._load_from_disk("csv"):
return
logger.warning(
"[DeliveryHistory] Saved files corrupt — rebuilding from CSV."
@@ -184,14 +228,15 @@ class DeliveryHistoryStore:
records = self._parse_csv()
if not records:
return
- self._build_and_save(records)
+ self._build_and_save(records, source="csv")
def _saved_files_are_current(self) -> bool:
- for path in (_VECTORS_PATH, _RECORDS_PATH, _META_PATH):
+ vectors_path, records_path, meta_path = _paths_for("csv")
+ for path in (vectors_path, records_path, meta_path):
if not os.path.isfile(path):
return False
try:
- with open(_META_PATH, "r", encoding="utf-8") as f:
+ with open(meta_path, "r", encoding="utf-8") as f:
meta = json.load(f)
if not os.path.isfile(CSV_PATH):
return True
@@ -199,10 +244,11 @@ class DeliveryHistoryStore:
except Exception:
return False
- def _load_from_disk(self) -> bool:
+ def _load_from_disk(self, source: str) -> bool:
try:
- vectors = np.load(_VECTORS_PATH) # (N, 4) float32
- with open(_RECORDS_PATH, "rb") as f:
+ vectors_path, records_path, _ = _paths_for(source)
+ vectors = np.load(vectors_path) # (N, 4) float32
+ with open(records_path, "rb") as f:
records = pickle.load(f)
if not records or len(vectors) != len(records):
@@ -342,7 +388,7 @@ class DeliveryHistoryStore:
return patterns, zone_index
- def _build_and_save(self, records: List[Dict]) -> None:
+ def _build_and_save(self, records: List[Dict], source: str = "csv") -> None:
vectors = np.array(
[[r["pickuplat"], r["pickuplon"], r["deliverylat"], r["deliverylong"]]
for r in records],
@@ -373,19 +419,28 @@ class DeliveryHistoryStore:
)
try:
- np.save(_VECTORS_PATH, vectors)
- with open(_RECORDS_PATH, "wb") as f:
+ vectors_path, records_path, meta_path = _paths_for(source)
+ np.save(vectors_path, vectors)
+ with open(records_path, "wb") as f:
pickle.dump(records, f, protocol=pickle.HIGHEST_PROTOCOL)
- csv_mtime = os.path.getmtime(CSV_PATH) if os.path.isfile(CSV_PATH) else 0
- with open(_META_PATH, "w", encoding="utf-8") as f:
+ # csv_mtime only means something for the CSV path's freshness check
+ # (_saved_files_are_current). DB-built snapshots are refreshed by the
+ # ETA-sync agent's own schedule, not a file-mtime comparison.
+ csv_mtime = (
+ os.path.getmtime(CSV_PATH)
+ if source == "csv" and os.path.isfile(CSV_PATH)
+ else 0
+ )
+ with open(meta_path, "w", encoding="utf-8") as f:
json.dump({
+ "source": source,
"csv_mtime": csv_mtime,
"record_count": len(records),
"pattern_zones": len(patterns),
"clear_patterns": clear,
}, f, indent=2)
logger.info(
- f"[DeliveryHistory] Saved to '{_STORE_DIR}'. "
+ f"[DeliveryHistory] Saved to '{_STORE_DIR}' (source={source}). "
"Next startup loads from disk."
)
except Exception as e:
@@ -665,7 +720,34 @@ class DeliveryHistoryStore:
records = self._parse_csv() # already merges corrections internally
if not records:
return 0
- self._build_and_save(records)
+ self._build_and_save(records, source="csv")
+ return len(records)
+
+ def rebuild_from_records(self, records: List[Dict]) -> int:
+ """
+ Rebuild the pattern table + vector index directly from pre-shaped
+ records (kitchen/pickuplat/pickuplon/deliverylat/deliverylong/userid/
+ ridername), bypassing CSV parsing entirely.
+
+ Called by the ETA-sync agent (delivery_history_service.py) after each
+ sync cycle when pattern_source=db, so this store stays as fresh as the
+ nearledb mirror — no manual CSV re-export/reload needed.
+ """
+ if not records:
+ logger.warning(
+ "[DeliveryHistory] rebuild_from_records got 0 records — "
+ "keeping the existing store as-is."
+ )
+ return 0
+
+ # Merge manually-verified corrections on top, same as the CSV path,
+ # so the human-override mechanism still works in DB mode.
+ if os.path.isfile(CORRECTIONS_PATH):
+ corr = self._parse_csv_file(CORRECTIONS_PATH, label="Corrections CSV")
+ if corr:
+ records = records + (corr * _CORRECTION_WEIGHT)
+
+ self._build_and_save(records, source="db")
return len(records)
def inject_corrections(self, corrections_path: str = CORRECTIONS_PATH,
@@ -720,14 +802,22 @@ class DeliveryHistoryStore:
self._patterns = patterns
self._zone_index = zone_index
- # Save to disk so next restart includes corrections
+ # Save to disk (under whichever source is currently active) so next
+ # restart includes corrections
try:
- np.save(_VECTORS_PATH, vectors)
- with open(_RECORDS_PATH, "wb") as f:
+ active_source = _pattern_source()
+ vectors_path, records_path, meta_path = _paths_for(active_source)
+ np.save(vectors_path, vectors)
+ with open(records_path, "wb") as f:
pickle.dump(merged, f)
- csv_mtime = os.path.getmtime(CSV_PATH) if os.path.isfile(CSV_PATH) else 0
- with open(_META_PATH, "w", encoding="utf-8") as f:
- json.dump({"csv_mtime": csv_mtime, "record_count": len(merged),
+ csv_mtime = (
+ os.path.getmtime(CSV_PATH)
+ if active_source == "csv" and os.path.isfile(CSV_PATH)
+ else 0
+ )
+ with open(meta_path, "w", encoding="utf-8") as f:
+ json.dump({"source": active_source, "csv_mtime": csv_mtime,
+ "record_count": len(merged),
"pattern_count": len(patterns)}, f)
logger.info(
f"[DeliveryHistory] Corrections injected and saved — "
diff --git a/app/templates/ml_dashboard.html b/app/templates/ml_dashboard.html
index b30f3c9..31df1cd 100644
--- a/app/templates/ml_dashboard.html
+++ b/app/templates/ml_dashboard.html
@@ -452,43 +452,6 @@
transform: scaleY(1.1)
}
- /* ── STRATEGY SWITCHER ── */
- .strategy-grid {
- display: grid;
- grid-template-columns: 1fr 1fr;
- gap: .5rem
- }
-
- .strategy-card {
- padding: .75rem;
- border: 1px solid var(--border);
- border-radius: 2px;
- cursor: pointer;
- transition: all .15s;
- }
-
- .strategy-card:hover {
- border-color: var(--border2)
- }
-
- .strategy-card.active {
- border-color: var(--accent);
- background: rgba(0, 212, 255, .06)
- }
-
- .strategy-name {
- font-size: .75rem;
- font-weight: 700;
- letter-spacing: .05em;
- margin-bottom: .2rem
- }
-
- .strategy-desc {
- font-family: var(--mono);
- font-size: .62rem;
- color: var(--muted)
- }
-
/* ── BADGE ── */
.badge {
display: inline-block;
@@ -718,11 +681,6 @@
—
—
-
-
SLA Breaches
-
—
-
Recent window
-
Avg Latency
—
@@ -758,17 +716,8 @@
-
- Strategy Mode
- —
-
-
-
-
-
-
Multi-Objective Pareto
-
-
+
Multi-Objective Pareto
+
@@ -794,28 +743,6 @@
Quality Distribution
-
-
Strategy Comparison
-
-
@@ -830,13 +757,12 @@
Zone |
Calls |
Avg Q |
- SLA Breaches |
Avg km |
- | Loading...
+ | Loading...
|
@@ -900,14 +826,6 @@