Stage 193: drift forecast — projected severity

core/forecast.py

@@ -0,0 +1,184 @@
+"""
+core.forecast — Stage 193 drift trajectory projection.
+
+The existing engine answers "what is the drift score TODAY?". This
+module answers "what will it likely be NEXT WEEK if the recent
+trajectory continues?" — useful for operators to triage the
+"still-medium-but-trending-critical" entities before they cross
+the threshold.
+
+Math (intentionally simple — a transparent linear projection beats
+an opaque ML model for an operator who needs to act on the number):
+
+  1. Pull the entity's recent drift_score history (most-recent first).
+  2. Fit a least-squares linear trend on the last N points.
+  3. Project the score `horizon_days` ahead at the same slope.
+  4. Clamp to [0, 1].
+  5. Map to severity using the entity_type's thresholds.
+
+Confidence is data-driven: with only 2 points (n=2) the fit is
+brittle; with 10+ points the trend is meaningful. We use a soft
+saturation: confidence = min(1.0, n_points / 8.0), so 8 history
+points yields full confidence and fewer yields proportionally less.
+
+The trend label is "worsening" / "stable" / "improving" based on
+the slope's magnitude:
+  |slope| < 0.005/day → stable
+  slope > 0           → worsening (drift score rising)
+  slope < 0           → improving (drift score falling)
+"""
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Dict, List, Optional
+
+from .signals import _linear_slope
+
+
+_DEFAULT_HORIZON_DAYS = 7
+_STABLE_SLOPE_THRESHOLD = 0.005  # |slope| below this is "stable"
+
+
+@dataclass
+class ForecastResult:
+    entity_id: str
+    n_history_points: int
+    current_score: float
+    current_severity: str
+    horizon_days: int
+    projected_score: float
+    projected_severity: str
+    trend: str            # worsening | stable | improving
+    slope_per_day: float  # raw slope, for explanation
+    confidence: float     # 0.0 - 1.0
+
+    def to_dict(self) -> Dict:
+        return {
+            "entity_id": self.entity_id,
+            "n_history_points": self.n_history_points,
+            "current_score": round(self.current_score, 4),
+            "current_severity": self.current_severity,
+            "horizon_days": self.horizon_days,
+            "projected_score": round(self.projected_score, 4),
+            "projected_severity": self.projected_severity,
+            "trend": self.trend,
+            "slope_per_day": round(self.slope_per_day, 6),
+            "confidence": round(self.confidence, 3),
+        }
+
+
+def severity_from_score(score: float,
+                         thresholds: Dict[str, float]) -> str:
+    """Tiny mirror of core.drift.severity_from_score that takes a
+    plain dict — kept inline so the forecast module doesn't drag
+    in the full drift module just for one lookup."""
+    if score >= thresholds.get("critical", 0.75):
+        return "critical"
+    if score >= thresholds.get("high", 0.55):
+        return "high"
+    if score >= thresholds.get("medium", 0.35):
+        return "medium"
+    return "low"
+
+
+def forecast(
+    entity_id: str,
+    history: List[Dict],
+    *,
+    horizon_days: int = _DEFAULT_HORIZON_DAYS,
+    severity_thresholds: Optional[Dict[str, float]] = None,
+) -> ForecastResult:
+    """Project the entity's drift score ``horizon_days`` ahead.
+
+    ``history`` is a list of dicts with at least ``score`` and
+    ``severity`` keys, ordered MOST-RECENT FIRST (the same shape
+    returned by ``RunRepository.entity_score_history``). The first
+    element is treated as the current state. The slope is fit on
+    the chronological order (newest last in math, even though the
+    input is reversed).
+
+    Empty history → returns a "no data" forecast with confidence 0
+    and trend "stable" so the caller doesn't crash.
+    """
+    if severity_thresholds is None:
+        # Reasonable defaults matching the engine's defaults.
+        severity_thresholds = {
+            "critical": 0.75, "high": 0.55, "medium": 0.35,
+        }
+    if not history:
+        return ForecastResult(
+            entity_id=entity_id,
+            n_history_points=0,
+            current_score=0.0,
+            current_severity="low",
+            horizon_days=horizon_days,
+            projected_score=0.0,
+            projected_severity="low",
+            trend="stable",
+            slope_per_day=0.0,
+            confidence=0.0,
+        )
+    # Reverse so we work in chronological order (oldest → newest).
+    chrono = list(reversed(history))
+    scores = [float(h["score"]) for h in chrono
+              if h.get("score") is not None]
+    if not scores:
+        return ForecastResult(
+            entity_id=entity_id,
+            n_history_points=0,
+            current_score=0.0,
+            current_severity="low",
+            horizon_days=horizon_days,
+            projected_score=0.0,
+            projected_severity="low",
+            trend="stable",
+            slope_per_day=0.0,
+            confidence=0.0,
+        )
+    current_score = scores[-1]
+    current_severity = (history[0].get("severity")
+                        or severity_from_score(current_score,
+                                                severity_thresholds))
+    if len(scores) < 2:
+        # One point — no trend to extract. Project flat with low
+        # confidence so the dashboard shows "(not enough data)".
+        return ForecastResult(
+            entity_id=entity_id,
+            n_history_points=len(scores),
+            current_score=current_score,
+            current_severity=current_severity,
+            horizon_days=horizon_days,
+            projected_score=current_score,
+            projected_severity=current_severity,
+            trend="stable",
+            slope_per_day=0.0,
+            confidence=0.1,
+        )
+    # Slope is "score units per index step". The history points are
+    # successive runs which are typically daily; we treat one step
+    # as one day. A future stage could pull actual wall-clock deltas
+    # if customers run hourly or weekly schedules.
+    slope = _linear_slope(scores)
+    projected_raw = current_score + slope * horizon_days
+    projected_clamped = max(0.0, min(1.0, projected_raw))
+    if abs(slope) < _STABLE_SLOPE_THRESHOLD:
+        trend = "stable"
+    elif slope > 0:
+        trend = "worsening"
+    else:
+        trend = "improving"
+    # Soft saturation on history depth. 8 points = full confidence.
+    confidence = min(1.0, len(scores) / 8.0)
+    return ForecastResult(
+        entity_id=entity_id,
+        n_history_points=len(scores),
+        current_score=current_score,
+        current_severity=current_severity,
+        horizon_days=horizon_days,
+        projected_score=projected_clamped,
+        projected_severity=severity_from_score(
+            projected_clamped, severity_thresholds),
+        trend=trend,
+        slope_per_day=slope,
+        confidence=confidence,
+    )

infra/api/app.py

@@ -2025,6 +2025,29 @@ def create_app(db_path: Optional[str] = None,
             decision_limit=decision_limit,
         )
 
+    # Stage 193 — drift forecast. "If the recent trajectory keeps
+    # going, what severity will this entity be at in N days?"
+    # Readonly tier — pure projection from existing run history.
+    @app.get("/tenants/{tenant_id}/entities/{entity_id}/forecast",
+              tags=["runs"])
+    async def get_entity_forecast_route(
+        tenant_id: str,
+        entity_id: str,
+        horizon_days: int = Query(default=7, ge=1, le=90),
+        history_limit: int = Query(default=14, ge=2, le=100),
+        key: ApiKey = Depends(auth_dep),
+    ):
+        require_tenant_access(key, tenant_id)
+        require_role(key, ROLE_READONLY)
+        try:
+            return svc.get_entity_forecast(
+                tenant_id, entity_id,
+                horizon_days=horizon_days,
+                history_limit=history_limit,
+            )
+        except ValueError as e:
+            raise ApiError("bad_request", str(e), status=400) from e
+
     # Stage 192 — peer benchmarking. How does this entity rank
     # against other entities of the same entity_type within the
     # tenant? Per-metric percentile + median + IQR + range. Empty

infra/service.py

@@ -1075,6 +1075,45 @@ class OrgStateService:
         )
         return row
 
+    # --- Stage 193 — drift forecast ------------------------------------
+
+    def get_entity_forecast(
+        self,
+        tenant_id: str,
+        entity_id: str,
+        *,
+        horizon_days: int = 7,
+        history_limit: int = 14,
+    ) -> dict:
+        """Stage 193 — project the entity's drift score `horizon_days`
+        ahead based on the recent trajectory. Returns the
+        ForecastResult.to_dict() shape.
+
+        Defaults: 7-day horizon, 14-point history window. The 14
+        matches the engine's default baseline_window so the forecast
+        sees roughly the same recent context the next run would.
+
+        Read-only — derives entirely from existing run history. Cheap,
+        polite to call on every drilldown render.
+        """
+        from core.forecast import forecast as _forecast
+
+        self._require_tenant(tenant_id)
+        if horizon_days < 1 or horizon_days > 90:
+            raise ValueError(
+                f"horizon_days must be in [1, 90], got {horizon_days!r}",
+            )
+        history = self.runs.entity_score_history(
+            tenant_id, entity_id, limit=history_limit,
+        )
+        # entity_score_history returns most-recent first; forecast()
+        # expects that shape.
+        result = _forecast(
+            entity_id, history,
+            horizon_days=horizon_days,
+        )
+        return result.to_dict()
+
     # --- Stage 192 — peer benchmarking ---------------------------------
 
     def get_entity_peer_comparison(