feat(ml): B1 drift monitoring live — apples-to-apples on canonical features

- MLScoringEngine feeds the model its rain features at inference
  (bundle rainfall + api_30 → same aggregates as training; they were
  silently 0.0 before) and distance_to_iffi_m; feature_row() exposes
  the named vector
- ShadowChallengerExecutor persists that vector in breakdown.features
  so the drift job compares training vs live on identical keys/scales
- drift job: monitored feature configurable (MONITORING__DRIFT_FEATURE,
  default rain.rain_72h_mm), require_features filter in the repo,
  enabled by default
- cache_cleanup purges model_runs older than
  SCORING__MODEL_RUNS_RETENTION_DAYS (national shadow ≈ 1 GB/day)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: gzileni

.env.example

@@ -177,6 +177,22 @@ LOG_JSON=false
 #REPORT__ENABLED=true
 #REPORT__HOUR_UTC=6
 
+# -----------------------------------------------------------------------------
+# ML drift monitoring (B1) — daily PSI/KS on one canonical model feature,
+# training vs the live vectors the shadow persists in model_runs
+# (breakdown.features — identical keys and scales by construction).
+# Informational: the retrain trigger only logs; promotion stays manual.
+# Note: with a case-control training set vs full-grid scoring, PSI on a
+# weather feature alarms whenever live weather differs from event-time
+# weather — read it as regime info, not as an automatic mandate.
+# -----------------------------------------------------------------------------
+#MONITORING__ENABLE_DRIFT_MONITORING=true
+#MONITORING__DRIFT_FEATURE=rain.rain_72h_mm
+#MONITORING__DRIFT_CHECK_HOURS=24
+# Shadow writes ~1M rows/day nationally (~1 GB/day): the cleanup job
+# purges model_runs older than this. 0 = keep everything.
+#SCORING__MODEL_RUNS_RETENTION_DAYS=30
+
 # --- Dispatch rules ---
 ALERT__MIN_LEVEL=Moderate         # Low | Moderate | High | VeryHigh
 ALERT__MIN_STATIC_S=0.5           # below-High cells alert only if S >= this (susceptible slope)

README.md

@@ -81,7 +81,7 @@ La stessa interfaccia `CellFeatureBundle` accetta anche il motore ML V2.
 | **Workflow MAF (V1)** | AreaResolver → StaticFactors → MeteoFetch → SeismicCheck → FireCheck → \[SensorFetch?\] → RiskScoring → EscalationGate → RiskAnalyst → Briefing → PersistResult → AlertDispatch | one-shot CLI | `agents/` + `limen monitor-once` |
 | **Provider LLM** | precedenza `LLM__PROVIDER` > Anthropic > OpenAI > Foundry > Ollama; il resolver salta i provider cloud senza SDK e cade su Ollama (solo httpx). Briefing in italiano; RiskAnalyst restituisce JSON tipizzato. | risolto all'avvio | `agents/llm_factory/resolve_llm_factory` |
 | **API HTTP** | `/health` + `/ready`, `POST /api/monitor/{aoi}`, `GET /api/aoi/{id}/risk/latest`, `GET /api/cell/{id}/breakdown`, `GET /api/aoi`, `GET /api/alerts`, `/api/tiles/...`, OpenAPI su `/docs` e `/redoc` | FastAPI / uvicorn | `api/` + `limen serve` |
-| **Job periodici** | workflow MAF orario (con shadow ML), **sweep previsionale** ogni 6 h, **nowcast radar DPC** ogni 15 min, report nazionale giornaliero, sync ISPRA settimanale, drift monitor, cache cleanup | APScheduler in-process | `api/jobs/` |
+| **Job periodici** | workflow MAF orario (con shadow ML), **sweep previsionale** ogni 6 h, **nowcast radar DPC** ogni 15 min, report nazionale giornaliero, sync ISPRA settimanale, **drift monitor ML** (PSI/KS training-vs-live sulle feature canoniche che lo shadow persiste), cache cleanup + retention di `model_runs` (default 30 gg) | APScheduler in-process | `api/jobs/` |
 | **Radar DPC (nowcast)** | SRI nazionale 1 km / 5 min (piattaforma radar DPC, CC-BY-SA): pioggia ≥ `NOWCAST__MIN_INTENSITY_MMH` su una regione ⇒ il workflow di quella AOI parte subito invece di aspettare il tick orario (cooldown 45 min; alert dal percorso operativo normale) | poll ogni `NOWCAST__INTERVAL_MINUTES` | `integrations/dpc/` + `api/jobs/nowcast_monitoring.py` |
 | **Forecast previsionale** | scoring a `now+H` ore con pioggia prevista Open-Meteo (osservata+prevista nella stessa finestra); champion + challenger ML sulle stesse celle; report on-demand o alert "PREVISIONE" schedulato con dedup (AOI, orizzonte) | `limen forecast` / job ogni `FORECAST__INTERVAL_HOURS` | `agents/workflows/forecast.py` + `api/jobs/forecast_monitoring.py` |
 | **MCP `limen-ops`** | `tool_risk_summary`, `tool_top_risk_cells`, `tool_cell_breakdown`, `tool_recent_alerts`, `tool_national_report`, `tool_run_monitor` (admin, fail-closed su `MCP_ADMIN_TOKEN`) per gateway agentici (OpenClaw, Claude Desktop) | servizio compose `mcp`, HTTP `127.0.0.1:8766/mcp` | `mcp/` + `limen mcp-serve` |

src/limen/agents/executors/shadow_challenger.py

@@ -40,9 +40,15 @@ async def run(self, ctx: MonitoringContext) -> MonitoringContext:
         model_version = getattr(self._challenger, "model_version", "v1-deterministic")
 
         bundles = assemble_bundles(ctx)
+        feature_row_fn = getattr(self._challenger, "feature_row", None)
         rows: list[ModelRunRow] = []
         for bundle in bundles:
             scored = self._challenger.score(bundle)
+            breakdown = scored.breakdown.model_dump(mode="json")
+            if feature_row_fn is not None:
+                # Canonical model inputs → drift monitoring compares
+                # training vs live on identical keys and scales.
+                breakdown["features"] = feature_row_fn(bundle)
             rows.append(
                 ModelRunRow(
                     cell_id=bundle.cell_id,
@@ -53,7 +59,7 @@ async def run(self, ctx: MonitoringContext) -> MonitoringContext:
                     role="challenger",
                     probability=scored.score,
                     risk_class=scored.level.value,
-                    breakdown=scored.breakdown.model_dump(mode="json"),
+                    breakdown=breakdown,
                 )
             )
 

src/limen/api/jobs/cache_cleanup.py

@@ -10,9 +10,33 @@
 
 from limen.api.dependencies import AppDependencies
 from limen.core.logging import get_logger
+from limen.data.db import acquire
 
 log = get_logger(__name__)
 
+# Per-tick cap: the job runs every few minutes, so retention deletes in
+# small chronological batches (id is bigserial) instead of one huge sweep.
+_RETENTION_BATCH = 50_000
+
+
+async def _purge_old_model_runs(retention_days: int) -> int:
+    if retention_days <= 0:
+        return 0
+    async with acquire() as conn:
+        tag = await conn.execute(
+            """
+            DELETE FROM model_runs WHERE id IN (
+                SELECT id FROM model_runs
+                WHERE computed_at < now() - make_interval(days => $1)
+                ORDER BY id
+                LIMIT $2
+            )
+            """,
+            retention_days,
+            _RETENTION_BATCH,
+        )
+    return int(tag.split()[-1])
+
 
 async def run_cache_cleanup_job(deps: AppDependencies) -> int:
     """Delete expired ``app_cache`` rows; returns the number removed."""
@@ -29,6 +53,15 @@ async def run_cache_cleanup_job(deps: AppDependencies) -> int:
             error_type=type(exc).__name__,
         )
         return 0
-    if removed:
-        log.info("job.cache_cleanup.done", removed=removed)
+    try:
+        purged = await _purge_old_model_runs(deps.settings.scoring.model_runs_retention_days)
+    except Exception as exc:
+        log.error(
+            "job.cache_cleanup.model_runs_error",
+            error=str(exc),
+            error_type=type(exc).__name__,
+        )
+        purged = 0
+    if removed or purged:
+        log.info("job.cache_cleanup.done", removed=removed, model_runs_purged=purged)
     return int(removed)

src/limen/api/jobs/drift_monitor.py

@@ -31,24 +31,31 @@ async def run_drift_monitor_job(deps: AppDependencies) -> int:
         log.info("job.drift.skip_no_training_samples")
         return 0
 
-    # Use the static.susc_ispra column as the reference univariate
-    # distribution — cheap to compute and a good proxy for distributional
-    # change. Production deployments will rotate which feature is
-    # monitored each cycle.
-    reference = [
-        float(s.features.get("static", {}).get("susc_ispra") or 0.0) for s in training_samples
-    ]
+    # Compare one canonical model feature training-vs-live. The shadow
+    # persists the exact vector the model consumed under
+    # breakdown["features"] (same keys and scales as training_samples),
+    # so the comparison is apples-to-apples by construction.
+    feature = deps.settings.monitoring.drift_feature
+    top, _, sub = feature.partition(".")
+    reference = [float((s.features.get(top) or {}).get(sub) or 0.0) for s in training_samples]
     reference_labels = [float(s.label) for s in training_samples]
 
     window_start = datetime.now(UTC) - timedelta(days=7)
-    recent = await recent_for_role("challenger", since=window_start, limit=10_000)
+    recent = await recent_for_role(
+        "challenger", since=window_start, limit=10_000, require_features=True
+    )
     if not recent:
         log.info("job.drift.no_recent_challenger_runs")
         return 0
 
     candidate = [
-        float(r.breakdown.get("static_terms", {}).get("susc_ispra") or 0.0) for r in recent
+        float(r.breakdown["features"][feature])
+        for r in recent
+        if feature in (r.breakdown.get("features") or {})
     ]
+    if not candidate:
+        log.info("job.drift.no_feature_rows", feature=feature)
+        return 0
     candidate_probs = [r.probability for r in recent]
     report = make_report(
         reference=reference,
@@ -63,6 +70,7 @@ async def run_drift_monitor_job(deps: AppDependencies) -> int:
     trigger = RetrainingTrigger.from_inputs(drift=report, new_iffi_since_last_train=0)
     log.info(
         "job.drift.done",
+        feature=feature,
         psi=report.psi,
         ks=report.ks,
         pred_drift=report.pred_drift,

src/limen/config/settings.py

@@ -315,6 +315,9 @@ class ScoringSettings(BaseSettings):
     mlflow_tracking_uri: str = "sqlite:///mlflow.db"
     mlflow_experiment: str = "limen-landslide"
     mlflow_registered_model: str = "limen-landslide-ml"
+    # Shadow rows older than this are purged by the cleanup job — the
+    # national hourly sweep writes ~1M rows/day (~1 GB/day). 0 = keep all.
+    model_runs_retention_days: int = Field(default=30, ge=0)
     mlflow_model_stage: Literal["Staging", "Production", "Archived"] = "Production"
     # Promotion gate — the ML model is blocked from champion until it
     # clears these floors on the same backtest the V1 baseline ran on.
@@ -368,7 +371,11 @@ class MonitoringSettings(BaseSettings):
     prediction_drift_alert: float = Field(default=0.15, ge=0.0)
     # APScheduler cadence — coarse, drift checks aren't a hot path.
     drift_check_hours: int = Field(default=24, ge=1)
-    enable_drift_monitoring: bool = False
+    enable_drift_monitoring: bool = True
+    # Canonical feature whose distribution is compared training-vs-live.
+    # Must exist both in training_samples.features and in the shadow's
+    # persisted breakdown["features"] (same keys, same scales).
+    drift_feature: str = "rain.rain_72h_mm"
 
 
 class GeodataSettings(BaseSettings):

src/limen/core/scoring/ml_engine.py

@@ -147,6 +147,15 @@ def from_registry(
     # ------------------------------------------------------------------
     # ScoringEngine interface
     # ------------------------------------------------------------------
+    def feature_row(self, bundle: CellFeatureBundle) -> dict[str, float]:
+        """Named canonical feature vector — persisted by the shadow so the
+        drift monitor compares training vs live on identical keys/scales."""
+        names = self._artefacts.feature_names
+        if not names:
+            return {}
+        row = _bundle_to_feature_row(bundle, names=names)
+        return dict(zip(names, row, strict=True))
+
     def score(self, bundle: CellFeatureBundle) -> RiskScore:
         """Predict the cell's calibrated probability + component breakdown.
 
@@ -216,14 +225,28 @@ def _clamp01_scaled(x: float | None, cap: float) -> float:
 
 def _bundle_to_feature_row(bundle: CellFeatureBundle, *, names: list[str]) -> list[float]:
     """Project a bundle onto the ordered feature vector the model expects."""
+    from limen.ml.rain_features import compute_rain_aggregates
+
+    rain = compute_rain_aggregates(
+        [(s.timestamp, s.precipitation_mm) for s in bundle.dynamic.rainfall.samples],
+        as_of=bundle.dynamic.valuation_time,
+    )
     flat: dict[str, float] = {
         "static.susc_ispra": _clamp01(bundle.static.susc_ispra),
         "static.iffi_density_500": float(bundle.static.iffi_density_500 or 0.0),
+        "static.distance_to_iffi_m": float(bundle.static.distance_to_iffi_m or 0.0),
         "static.slope_deg": float(bundle.static.slope_deg or 0.0),
         "static.pai_class_norm": _clamp01(bundle.static.pai_class_norm),
         "static.litho_weight": _clamp01(bundle.static.litho_weight),
         "static.twi": float(bundle.static.twi or 0.0),
         "static.curvature": float(bundle.static.curvature or 0.0),
+        # Same antecedent-rain aggregates the model trained on (CERRA
+        # replay). The monitoring window is 48 h, so rain_72h is a lower
+        # bound; the 30-day total comes from the API_30 archive lookup.
+        "rain.rain_24h_mm": rain["rain_24h_mm"],
+        "rain.rain_72h_mm": rain["rain_72h_mm"],
+        "rain.max_i_24h_mmh": rain["max_i_24h_mmh"],
+        "rain.rain_30d_mm": float(bundle.dynamic.api_30_mm or rain["rain_30d_mm"]),
     }
     if not names:
         # Resolver fallback path: no feature schema → uniform vector,

src/limen/data/repos/model_runs_repo.py

@@ -65,21 +65,29 @@ async def insert_many(rows: Iterable[ModelRunRow]) -> int:
 
 
 async def recent_for_role(
-    role: ModelRole, *, since: datetime, limit: int = 10_000
+    role: ModelRole,
+    *,
+    since: datetime,
+    limit: int = 10_000,
+    require_features: bool = False,
 ) -> list[ModelRunRow]:
+    """Newest runs for a role. ``require_features`` keeps only rows whose
+    breakdown carries the canonical feature vector (drift monitoring)."""
     async with acquire() as conn:
         rows = await conn.fetch(
             """
             SELECT cell_id, valuation_time, aoi_id, model_uri, model_version,
                    role, probability, risk_class, breakdown
             FROM model_runs
             WHERE role = $1 AND computed_at >= $2
+              AND (NOT $4 OR breakdown ? 'features')
             ORDER BY computed_at DESC
             LIMIT $3
             """,
             role,
             since,
             limit,
+            require_features,
         )
     return [_to_row(r) for r in rows]