feat(simple_metrics): add adaptive-state hysteresis and recovery-run rules (SIENTIAPDE-2191)

laborious/activities/model_metrics.py

@@ -4,6 +4,7 @@ with workflow.unsafe.imports_passed_through():
     import time
     import traceback
     import warnings
+    from math import ceil
     from typing import Any
 
     import pandas as pd
@@ -48,6 +49,48 @@ def threshold_breaches(
     return breaches
 
 
+_MAX_RECOVERY_RUNS = 12
+_FREQUENCY_UNIT_SECONDS = {'s': 1, 'm': 60, 'h': 3600, 'd': 86400}
+
+
+def frequency_to_seconds(frequency: str) -> int:
+    """`'5m'` -> 300. Same `'<int><s|m|h|d>'` format the orchestrator schedules with."""
+    # ponytail: copy of orchestrator/utils/converters.py:parse_frequency (no shared helper in
+    # sientia_do), same int() parsing so both repos accept the same strings, plus a > 0 guard;
+    # move it to sientia_do if a third repo needs it.
+    unit = _FREQUENCY_UNIT_SECONDS.get(frequency[-1:])
+    if unit is None:
+        raise ValueError(f'Invalid frequency: {frequency!r}')
+    value = int(frequency[:-1])
+    if value <= 0:
+        raise ValueError(f'Invalid frequency: {frequency!r}')
+    return value * unit
+
+
+def recovery_runs(interval_minutes: int, frequency_alert: str) -> int:
+    """
+    N of the hysteresis rule: healthy runs needed to renew the whole window without a breach,
+    capped at 12 (SIENTIAPDE-1995 decision 3).
+    """
+    runs = ceil(interval_minutes * 60 / frequency_to_seconds(frequency_alert))
+    return min(_MAX_RECOVERY_RUNS, max(1, runs))
+
+
+def next_adaptive_state(state: dict[str, Any], breached: bool, runs: int) -> dict[str, Any]:
+    """
+    Hysteresis rule of the adaptive frequency (SIENTIAPDE-1995): escalate on the first breach,
+    recover after `runs` consecutive non-breach runs.
+    """
+    if breached:
+        return {'escalated': True, 'healthy_streak': 0}
+    if not state.get('escalated'):
+        return {'escalated': False, 'healthy_streak': 0}
+    streak = int(state.get('healthy_streak', 0)) + 1
+    if streak >= runs:
+        return {'escalated': False, 'healthy_streak': 0}
+    return {'escalated': True, 'healthy_streak': streak}
+
+
 class ModelMetrics(SientiaMonitoring):
     """
     Metrics and statistical analysis activities for the Laborious pipeline.

tests/laborious/activities/test_model_metrics.py

@@ -7,7 +7,13 @@ from sientia_do.observability.metrics_controller import MetricsController
 from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
 from sientia_model.analytics.drift_analysis import DriftInsufficientDataError
 
-from laborious.activities.model_metrics import ModelMetrics, threshold_breaches
+from laborious.activities.model_metrics import (
+    ModelMetrics,
+    frequency_to_seconds,
+    next_adaptive_state,
+    recovery_runs,
+    threshold_breaches,
+)
 
 
 @fixture
@@ -1230,6 +1236,8 @@ def test_to_naive_utc_keeps_naive_timestamps():
         ([{'metric': 'rmse', 'value': 0.5}], {'rmse_max': 1.0}, 0),
         ([{'metric': 'r2', 'value': 0.5}], {'r2_min': 0.9}, 1),
         ([{'metric': 'r2', 'value': 0.95}], {'r2_min': 0.9}, 0),
+        ([{'metric': 'rmse', 'value': 1.0}], {'rmse_max': 1.0}, 0),  # equal is not a breach
+        ([{'metric': 'r2', 'value': 0.9}], {'r2_min': 0.9}, 0),
         ([{'metric': 'mae', 'value': 99.0}], {'rmse_max': 1.0}, 0),
         ([{'metric': 'rmse', 'value': 99.0}], None, 0),
         ([], {'rmse_max': 1.0}, 0),
@@ -1242,3 +1250,53 @@ def test_to_naive_utc_keeps_naive_timestamps():
 )
 def test_threshold_breaches(rows, thresholds, expected_count):
     assert len(threshold_breaches(rows, thresholds)) == expected_count
+
+
+def _state(escalated: bool, streak: int) -> dict:
+    return {'escalated': escalated, 'healthy_streak': streak}
+
+
+@mark.parametrize(
+    'state, breached, runs, expected',
+    [
+        ({}, True, 3, _state(True, 0)),  # first breach escalates
+        ({}, False, 3, _state(False, 0)),  # missing state = healthy
+        (_state(False, 0), False, 3, _state(False, 0)),  # healthy stays healthy
+        (_state(True, 0), False, 3, _state(True, 1)),  # healthy run while escalated
+        (_state(True, 1), False, 3, _state(True, 2)),
+        (_state(True, 2), False, 3, _state(False, 0)),  # N-th healthy run recovers
+        (_state(True, 2), True, 3, _state(True, 0)),  # breach resets the streak
+        (_state(True, 0), False, 1, _state(False, 0)),  # N=1 recovers on first healthy run
+    ],
+)
+def test_next_adaptive_state(state, breached, runs, expected):
+    assert next_adaptive_state(state, breached, runs) == expected
+
+
+@mark.parametrize(
+    'frequency, seconds',
+    [('30s', 30), ('5m', 300), ('2h', 7200), ('1d', 86400)],
+)
+def test_frequency_to_seconds(frequency, seconds):
+    assert frequency_to_seconds(frequency) == seconds
+
+
+@mark.parametrize('frequency', ['', '5', 'm', '5x', 'abc', '0m'])
+def test_frequency_to_seconds_rejects_invalid(frequency):
+    with raises(ValueError):
+        frequency_to_seconds(frequency)
+
+
+@mark.parametrize(
+    'interval_minutes, frequency_alert, expected',
+    [
+        (60, '5m', 12),  # 60/5 = 12, at the cap
+        (60, '15m', 4),
+        (60, '7m', 9),  # ceil(8.57)
+        (60, '30s', 12),  # 120 capped at 12
+        (1440, '1h', 12),  # 24 capped at 12
+        (5, '1h', 1),  # window shorter than one run: at least 1
+    ],
+)
+def test_recovery_runs(interval_minutes, frequency_alert, expected):
+    assert recovery_runs(interval_minutes, frequency_alert) == expected