File size: 2,155 Bytes
6835659 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | from __future__ import annotations
from typing import Any, Dict, Optional
from src.coherence.thresholds import AdaptiveThresholds
def detect_drift(
msci: Optional[float],
st_i: Optional[float],
st_a: Optional[float],
si_a: Optional[float] = None,
) -> Dict[str, Any]:
"""
Drift logic (Phase-3C):
- global_drift is True if msci is FAIL OR >=2 metrics are FAIL.
- includes drift_score (fail_count / metrics_count) for debugging.
"""
drift: Dict[str, Any] = {
"visual_drift": False,
"audio_drift": False,
"global_drift": False,
}
if st_i is not None and st_a is not None and st_i + 0.15 < st_a:
drift["visual_drift"] = True
if st_i is not None and st_a is not None and st_a + 0.15 < st_i:
drift["audio_drift"] = True
metrics: Dict[str, float] = {}
if msci is not None:
metrics["msci"] = msci
if st_i is not None:
metrics["st_i"] = st_i
if st_a is not None:
metrics["st_a"] = st_a
if si_a is not None:
metrics["si_a"] = si_a
try:
if metrics:
thresholds = AdaptiveThresholds()
statuses = {k: thresholds.classify_value(k, v) for k, v in metrics.items()}
fail_count = sum(1 for s in statuses.values() if s == "FAIL")
msci_fail = statuses.get("msci") == "FAIL"
drift["global_drift"] = bool(msci_fail or fail_count >= 2)
drift["global_drift_score"] = float(fail_count / max(len(statuses), 1))
drift["fail_count"] = fail_count
drift["msci_fail"] = msci_fail
drift["status"] = statuses
else:
drift["global_drift"] = False
drift["global_drift_score"] = 0.0
drift["fail_count"] = 0
drift["msci_fail"] = False
drift["status"] = {}
except Exception:
if msci is not None:
drift["global_drift"] = msci < 0.35
drift["global_drift_score"] = 1.0 if drift["global_drift"] else 0.0
else:
drift["global_drift"] = False
drift["global_drift_score"] = 0.0
return drift
|