diff --git a/laborious/workflows/simple_metrics.py b/laborious/workflows/simple_metrics.py index 8819d7e..475f871 100644 --- a/laborious/workflows/simple_metrics.py +++ b/laborious/workflows/simple_metrics.py @@ -33,6 +33,7 @@ class SimpleMetrics: target_name = model_config['target'] model_type = model_config.get('model_type') thresholds = model_config.get('simple_metrics_thresholds') + min_samples = model_config.get('simple_metrics_min_samples') # Built from internal Temporal workflow config, not user input (bandit B608). query = f""" @@ -65,6 +66,15 @@ class SimpleMetrics: if not target_data: return + # Too few rows for a reliable value: metrics are still persisted, but a noisy short + # window must not alert (SIENTIAPDE-1995). + enough_samples = not min_samples or len(target_data) >= min_samples + if not enough_samples: + workflow.logger.info( + f'{len(target_data)} rows < simple_metrics_min_samples={min_samples}: ' + 'threshold alerting skipped' + ) + simple_metrics = await workflow.execute_local_activity_method( Activities.calculate_simple_metrics, { @@ -74,7 +84,7 @@ class SimpleMetrics: 'metrics': input_data.get('metrics', ['rmse', 'mse', 'mae', 'r2']), 'interval_minutes': interval_minutes, 'model_type': model_type, - 'thresholds': thresholds, + 'thresholds': thresholds if enough_samples else None, }, retry_policy=retry_policy, start_to_close_timeout=timedelta(seconds=300), diff --git a/tests/laborious/workflows/test_simple_metrics.py b/tests/laborious/workflows/test_simple_metrics.py index 5d94f60..c293685 100644 --- a/tests/laborious/workflows/test_simple_metrics.py +++ b/tests/laborious/workflows/test_simple_metrics.py @@ -1,4 +1,4 @@ -from unittest.mock import ANY, AsyncMock, call, patch +from unittest.mock import ANY, AsyncMock, MagicMock, call, patch from pytest import fixture, mark from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ @@ -217,3 +217,56 @@ async def test_run_default_metrics(workflow_mock: AsyncMock, simple_metrics: Sim retry_policy=ANY, start_to_close_timeout=ANY, ) + + +def _input(**model_config) -> dict: + return { + 'schedule_name': 'test_schedule', + 'model_name': 'test_model', + 'model_id': 'test_model_id', + 'interval_minutes': 60, + 'model_config': {'target': 'test_target', **model_config}, + 'schema': 'test_schema', + 'predictions_table_name': 'test_predictions_table', + 'data_table_name': 'test_data_table', + 'target_table_name': 'test_target_table', + 'metrics': ['rmse'], + } + + +_ROW = {'timestamp': '2026-10-06 10:00:00+0000', 'prediction': 1.0, 'target': 2.0} + + +@mark.asyncio +@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock) +async def test_run_below_min_samples_drops_thresholds(workflow_mock, simple_metrics): + workflow_mock.logger = MagicMock() + # load, export, and the adaptive-state call S2 (Task 8) adds. + workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 2, None, None]) + workflow_mock.execute_local_activity_method = AsyncMock( + return_value=[{'metric': 'rmse', 'value': 9.0}] + ) + + await simple_metrics.run( + _input(simple_metrics_thresholds={'rmse_max': 1.0}, simple_metrics_min_samples=5) + ) + + activity_input = workflow_mock.execute_local_activity_method.call_args.args[1] + assert activity_input['thresholds'] is None + + +@mark.asyncio +@patch('laborious.workflows.simple_metrics.workflow', new_callable=AsyncMock) +async def test_run_at_min_samples_keeps_thresholds(workflow_mock, simple_metrics): + workflow_mock.logger = MagicMock() + workflow_mock.execute_activity_method = AsyncMock(side_effect=[[_ROW] * 5, None, None]) + workflow_mock.execute_local_activity_method = AsyncMock( + return_value=[{'metric': 'rmse', 'value': 9.0}] + ) + + await simple_metrics.run( + _input(simple_metrics_thresholds={'rmse_max': 1.0}, simple_metrics_min_samples=5) + ) + + activity_input = workflow_mock.execute_local_activity_method.call_args.args[1] + assert activity_input['thresholds'] == {'rmse_max': 1.0}