diff --git a/laborious/activities/model_metrics.py b/laborious/activities/model_metrics.py index 8b59f00..d2271b9 100644 --- a/laborious/activities/model_metrics.py +++ b/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 `''` 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. diff --git a/tests/laborious/activities/test_model_metrics.py b/tests/laborious/activities/test_model_metrics.py index e849757..c6490ad 100644 --- a/tests/laborious/activities/test_model_metrics.py +++ b/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