core/drift.py

from dataclasses import dataclass, field
from datetime import date
from typing import Dict, List


@dataclass
class DriftIssue:
    issue_id: str
    entity_id: str
    entity_type: str
    detected_at: date
    score: float
    severity: str
    title: str
    explanation: str
    evidence: List[Dict] = field(default_factory=list)


# Documented defaults. Stage 1 makes these overridable per entity-type via
# config (see core/config.py), but the defaults stay pinned here so existing
# behaviour — and the tests that pin it — never change silently.
DEFAULT_DRIFT_WEIGHTS = {
    "delta": 0.25,
    "psi": 0.20,
    "xi": 0.25,
    "gamma": 0.15,
    "kappa": 0.15,
    # Stage 187 — α (acceleration / second derivative). Default
    # weight 0 so the signal is fully opt-in via entity_type
    # config — adding the signal must not silently shift drift
    # scores on any pre-existing tenant.
    "alpha": 0.0,
}
DEFAULT_SEVERITY_THRESHOLDS = {
    "critical": 0.75,
    "high": 0.55,
    "medium": 0.35,
}


def drift_score(signals: Dict[str, float], weights: Dict[str, float] = None) -> float:
    """Weighted aggregate of the five signals into a 0..1 drift score.

    ``psi`` and ``kappa`` are "health" signals (1.0 = healthy) so they enter as
    ``(1 - value)``; ``delta``, ``xi`` and ``gamma`` are "stress" signals.
    ``weights`` defaults to ``DEFAULT_DRIFT_WEIGHTS`` — pass a config-derived
    dict to override per entity-type. Missing keys fall back to the default.
    """
    w = DEFAULT_DRIFT_WEIGHTS if weights is None else weights
    delta = signals.get("delta", 0.0)
    psi = signals.get("psi", 1.0)
    xi = signals.get("xi", 0.0)
    gamma = signals.get("gamma", 0.0)
    kappa = signals.get("kappa", 1.0)
    alpha = signals.get("alpha", 0.0)         # Stage 187
    return (
        w.get("delta", DEFAULT_DRIFT_WEIGHTS["delta"]) * delta +
        w.get("psi", DEFAULT_DRIFT_WEIGHTS["psi"]) * (1.0 - psi) +
        w.get("xi", DEFAULT_DRIFT_WEIGHTS["xi"]) * xi +
        w.get("gamma", DEFAULT_DRIFT_WEIGHTS["gamma"]) * gamma +
        w.get("kappa", DEFAULT_DRIFT_WEIGHTS["kappa"]) * (1.0 - kappa) +
        w.get("alpha", DEFAULT_DRIFT_WEIGHTS["alpha"]) * alpha
    )


def severity_from_score(score: float, thresholds: Dict[str, float] = None) -> str:
    """Map a drift score to a severity label.

    ``thresholds`` defaults to ``DEFAULT_SEVERITY_THRESHOLDS`` — pass a
    config-derived dict to override per entity-type.
    """
    t = DEFAULT_SEVERITY_THRESHOLDS if thresholds is None else thresholds
    if score >= t.get("critical", DEFAULT_SEVERITY_THRESHOLDS["critical"]):
        return "critical"
    if score >= t.get("high", DEFAULT_SEVERITY_THRESHOLDS["high"]):
        return "high"
    if score >= t.get("medium", DEFAULT_SEVERITY_THRESHOLDS["medium"]):
        return "medium"
    return "low"