feat(simple_metrics): add threshold-based alerting via send_notification

Compares each computed metric against optional per-model thresholds
from model_config.simple_metrics_thresholds. Convention:
- {metric}_max: breach when value > threshold (rmse, mse, mae)
- {metric}_min: breach when value < threshold (r2)

Fires WARNING notification on breach. Missing thresholds = no alerting.
Schema designed to be extensible for Card 2 (Drift) thresholds.

SIENTIAPDE-1986

laborious/activities/model_metrics.py

@@ -6,7 +6,6 @@ with workflow.unsafe.imports_passed_through():
     import warnings
     from typing import Any
 
-    import numpy as np
     import pandas as pd
     from pandas import DataFrame, Index, Series, to_datetime
     from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
@@ -16,6 +15,7 @@ with workflow.unsafe.imports_passed_through():
     from sientia_do.observability.sientia_monitoring import SientiaMonitoring
     from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
     from sientia_model.analytics.drift_analysis import DriftAnalysis, DriftInsufficientDataError
+    from sientia_model.metrics.regression import RegressionMetrics
 
     from laborious import metrics
     from laborious.utils.dataframe_debug import build_dataframe_debug_message
@@ -365,62 +365,62 @@ class ModelMetrics(SientiaMonitoring):
     @activity.defn(name='calculate_simple_metrics')
     def calculate_simple_metrics(self, input_data: dict[str, Any]) -> list[dict]:
         """
-        Calculate simple metrics for a model. Metrics available are:
+        Calculate simple regression metrics for a model using RegressionMetrics.
-            - rmse
+
-            - mse
-            - mae
-            - r2
-            - accuracy
-            - precision
-            - recall
-            - f1
         Args:
-            input_data (dict[str, Any]): Input data containing:
+            input_data: Input data containing:
                 - metadata (dict): Workflow execution metadata
                 - model_id (str): ID of the MLFlow model
-                - target_data (pd.DataFrame): Target data for calculating metrics, containing target and prediction columns
+                - target_data (list[dict]): Target data with target, prediction, timestamp columns
-                - metrics (list[str]): List of metrics to calculate
+                - metrics (list[str]): Metric names to calculate
+                - interval_minutes (int): Window interval in minutes
+                - model_type (str | None): Model algorithm type (for r2 lock)
         Returns:
-            dict[Hashable, Any]: Dictionary containing the calculated metrics
+            list[dict]: Records with metric, value, model_id, timestamp, data_size, interval_minutes
         """
-
         metadata = input_data['metadata']
         model_id = input_data['model_id']
         target_data = DataFrame(input_data['target_data'])
-        metric_names = input_data['metrics']
+        metric_names = list(input_data['metrics'])
         interval_minutes = input_data['interval_minutes']
+        model_type = input_data.get('model_type')
 
         data_size = target_data.shape[0]
 
-        output_data = []
+        # Filter r2 when model_type is known and unsupported
+        if (
+            model_type
+            and 'r2' in metric_names
+            and not RegressionMetrics.is_r2_supported(model_type)
+        ):
+            metric_names = [m for m in metric_names if m != 'r2']
+            self.warning(
+                f'r2 excluded for model {model_id}: not supported for model_type={model_type}',
+                metadata,
+            )
+            self.send_notification(
+                metadata=metadata,
+                notification_id='SIMPLE_METRICS_R2_UNSUPPORTED',
+                message=f'r2 excluded: not a valid metric for model_type={model_type}',
+                block='model_metrics',
+                level=NotificationLevel.WARNING,
+            )
 
-        diff = target_data['target'] - target_data['prediction']
+        if not metric_names:
-        diff_squared = diff**2
+            self.info(f'No metrics to calculate for model {model_id} after filtering', metadata)
+            return []
 
         self.info(f'Calculating simple metrics for model {model_id}: {metric_names}', metadata)
 
-        for metric in metric_names:
+        # Build Series with DatetimeIndex for RegressionMetrics
-            if metric == 'rmse':
+        timestamps = pd.to_datetime(target_data['timestamp'])
-                output_data.append({'metric': 'rmse', 'value': np.sqrt(np.mean(diff_squared))})
+        real_data = Series(target_data['target'].values, index=timestamps, dtype=float)
-            elif metric == 'mse':
+        predictions = Series(target_data['prediction'].values, index=timestamps, dtype=float)
-                output_data.append({'metric': 'mse', 'value': np.mean(diff_squared)})
-            elif metric == 'mae':
-                output_data.append({'metric': 'mae', 'value': np.mean(np.abs(diff))})
-            elif metric == 'r2':
-                y_true = target_data['target']
-                y_mean = np.mean(y_true)
 
-                ss_res = np.sum(diff_squared)
+        regression = RegressionMetrics(real_data, predictions)
-                ss_tot = np.sum((y_true - y_mean) ** 2)
+        output_data = regression.calculate(metric_names)
-
-                # Evita divisão por zero
-                if ss_tot == 0:
-                    r2_score = 0.0
-                else:
-                    r2_score = 1 - (ss_res / ss_tot)
-
-                output_data.append({'metric': 'r2', 'value': r2_score})
 
+        # Wrap with metadata columns matching the existing output schema
         data = DataFrame(output_data)
         data['model_id'] = model_id
         data['timestamp'] = target_data['timestamp'].max()
@@ -429,4 +429,34 @@ class ModelMetrics(SientiaMonitoring):
 
         self._debug_dataframe(f'Simple metrics dataframe: Size {data.shape}', data, metadata)
 
+        # Threshold alerting (optional — no crash when absent)
+        thresholds = input_data.get('thresholds')
+        if thresholds:
+            # Convention: _max thresholds breach when value > threshold,
+            #             _min thresholds breach when value < threshold.
+            for row in output_data:
+                metric_name = row['metric']
+                value = row['value']
+                max_key = f'{metric_name}_max'
+                min_key = f'{metric_name}_min'
+
+                breach_msg = None
+                if max_key in thresholds and value > thresholds[max_key]:
+                    breach_msg = f'{metric_name}={value} exceeds {max_key}={thresholds[max_key]}'
+                elif min_key in thresholds and value < thresholds[min_key]:
+                    breach_msg = f'{metric_name}={value} below {min_key}={thresholds[min_key]}'
+
+                if breach_msg:
+                    self.warning(
+                        f'Threshold breach for model {model_id}: {breach_msg}',
+                        metadata,
+                    )
+                    self.send_notification(
+                        metadata=metadata,
+                        notification_id='SIMPLE_METRICS_THRESHOLD_BREACH',
+                        message=f'Threshold breach for model {model_id}: {breach_msg}',
+                        block='model_metrics',
+                        level=NotificationLevel.WARNING,
+                    )
+
         return data.to_dict(orient='records')

laborious/workflows/simple_metrics.py

@@ -31,6 +31,8 @@ class SimpleMetrics:
 
         model_config = input_data['model_config']
         target_name = model_config['target']
+        model_type = model_config.get('model_type')
+        thresholds = model_config.get('simple_metrics_thresholds')
 
         query = f"""
             select p."timestamp", p.prediction, ld.value as "target"
@@ -70,6 +72,8 @@ class SimpleMetrics:
                 'target_data': target_data,
                 'metrics': input_data.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
                 'interval_minutes': interval_minutes,
+                'model_type': model_type,
+                'thresholds': thresholds,
             },
             retry_policy=retry_policy,
             start_to_close_timeout=timedelta(seconds=300),

requirements-local.txt

@@ -4,7 +4,7 @@ sqlalchemy
 asyncua==1.0.6
 redis
 git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.1
-git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.10.0
+git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.13.1
 prometheus-client   
 botocore
 boto3

requirements.txt

@@ -4,7 +4,7 @@ sqlalchemy
 asyncua==1.0.6
 redis
 sientia_do>=1.12.1
-sientia_model>=0.8.2
+sientia_model>=0.12.0
 prometheus-client
 botocore
 boto3

tests/laborious/activities/test_model_metrics.py

@@ -45,6 +45,23 @@ metadata = {
 }
 
 
+def _mock_regression_metrics_class(calculate_return):
+    """Returns a patch context manager that mocks RegressionMetrics."""
+    mock_instance = MagicMock()
+    mock_instance.calculate.return_value = calculate_return
+    mock_class = MagicMock(return_value=mock_instance)
+    mock_class.is_r2_supported = MagicMock(return_value=True)
+    mock_class.supported_metrics = MagicMock(return_value=['rmse', 'mse', 'mae', 'r2'])
+    return (
+        patch(
+            'laborious.activities.model_metrics.RegressionMetrics',
+            mock_class,
+        ),
+        mock_class,
+        mock_instance,
+    )
+
+
 def test_calculate_drift_invalid_chunk_period(model_metrics_activity):
     # Arrange
     input_data = {
@@ -693,7 +710,6 @@ def test_get_drift_metrics_dataframe_error(
 
 
 def test_calculate_simple_metrics_success_all_metrics(model_metrics_activity):
-    # Arrange
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'],
@@ -710,28 +726,28 @@ def test_calculate_simple_metrics_success_all_metrics(model_metrics_activity):
         'interval_minutes': 5,
     }
 
-    # Act
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+        [
-
+            {'metric': 'rmse', 'value': 0.1},
-    # Assert
+            {'metric': 'mse', 'value': 0.01},
-    assert len(result['metric']) == 4
+            {'metric': 'mae', 'value': 0.1},
-    assert 'rmse' in result['metric'].values
+            {'metric': 'r2', 'value': 0.99},
-    assert 'mse' in result['metric'].values
+        ]
-    assert 'mae' in result['metric'].values
-    assert 'r2' in result['metric'].values
-    assert all(model_id == 'test_model_id' for model_id in result['model_id'].values)
-    assert all(timestamp == '2023-05-26 11:12:29' for timestamp in result['timestamp'].values)
-    assert all(data_size == 3 for data_size in result['data_size'].values)
-    assert all(interval_minutes == 5 for interval_minutes in result['interval_minutes'].values)
-    model_metrics_activity.info.assert_called_once_with(
-        "Calculating simple metrics for model test_model_id: ['rmse', 'mse', 'mae', 'r2']",
-        metadata['metadata'],
     )
-    model_metrics_activity.debug.assert_called_once()
+
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 4
+    assert set(result['metric'].values) == {'rmse', 'mse', 'mae', 'r2'}
+    assert all(mid == 'test_model_id' for mid in result['model_id'].values)
+    assert all(ts == '2023-05-26 11:12:29' for ts in result['timestamp'].values)
+    assert all(ds == 3 for ds in result['data_size'].values)
+    assert all(im == 5 for im in result['interval_minutes'].values)
+    mock_instance.calculate.assert_called_once_with(['rmse', 'mse', 'mae', 'r2'])
 
 
 def test_calculate_simple_metrics_success_rmse_only(model_metrics_activity):
-    # Arrange
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -748,23 +764,23 @@ def test_calculate_simple_metrics_success_rmse_only(model_metrics_activity):
         'interval_minutes': 5,
     }
 
-    # Act
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+        [{'metric': 'rmse', 'value': 0.1}]
+    )
 
-    # Assert
+    with patcher:
-    assert len(result['metric']) == 1
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 1
     assert result['metric'].values[0] == 'rmse'
     assert result['model_id'].values[0] == 'test_model_id'
     assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
     assert result['data_size'].values[0] == 2
     assert result['interval_minutes'].values[0] == 5
-    model_metrics_activity.info.assert_called_once_with(
+    mock_instance.calculate.assert_called_once_with(['rmse'])
-        "Calculating simple metrics for model test_model_id: ['rmse']", metadata['metadata']
-    )
 
 
 def test_calculate_simple_metrics_success_mse_only(model_metrics_activity):
-    # Arrange
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -781,23 +797,23 @@ def test_calculate_simple_metrics_success_mse_only(model_metrics_activity):
         'interval_minutes': 5,
     }
 
-    # Act
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+        [{'metric': 'mse', 'value': 0.01}]
+    )
 
-    # Assert
+    with patcher:
-    assert len(result['metric']) == 1
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 1
     assert result['metric'].values[0] == 'mse'
     assert result['model_id'].values[0] == 'test_model_id'
     assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
     assert result['data_size'].values[0] == 2
     assert result['interval_minutes'].values[0] == 5
-    model_metrics_activity.info.assert_called_once_with(
+    mock_instance.calculate.assert_called_once_with(['mse'])
-        "Calculating simple metrics for model test_model_id: ['mse']", metadata['metadata']
-    )
 
 
 def test_calculate_simple_metrics_success_mae_only(model_metrics_activity):
-    # Arrange
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -814,23 +830,23 @@ def test_calculate_simple_metrics_success_mae_only(model_metrics_activity):
         'interval_minutes': 5,
     }
 
-    # Act
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+        [{'metric': 'mae', 'value': 0.1}]
+    )
 
-    # Assert
+    with patcher:
-    assert len(result['metric']) == 1
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 1
     assert result['metric'].values[0] == 'mae'
     assert result['model_id'].values[0] == 'test_model_id'
     assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
     assert result['data_size'].values[0] == 2
     assert result['interval_minutes'].values[0] == 5
-    model_metrics_activity.info.assert_called_once_with(
+    mock_instance.calculate.assert_called_once_with(['mae'])
-        "Calculating simple metrics for model test_model_id: ['mae']", metadata['metadata']
-    )
 
 
 def test_calculate_simple_metrics_success_r2_only(model_metrics_activity):
-    # Arrange
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -847,24 +863,24 @@ def test_calculate_simple_metrics_success_r2_only(model_metrics_activity):
         'interval_minutes': 5,
     }
 
-    # Act
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+        [{'metric': 'r2', 'value': 0.95}]
+    )
 
-    # Assert
+    with patcher:
-    assert len(result['metric']) == 1
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 1
     assert result['metric'].values[0] == 'r2'
     assert result['model_id'].values[0] == 'test_model_id'
     assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
     assert result['data_size'].values[0] == 2
     assert result['interval_minutes'].values[0] == 5
-    model_metrics_activity.info.assert_called_once_with(
+    mock_instance.calculate.assert_called_once_with(['r2'])
-        "Calculating simple metrics for model test_model_id: ['r2']", metadata['metadata']
-    )
 
 
-def test_calculate_simple_metrics_r2_zero_ss_tot(model_metrics_activity):
+def test_calculate_simple_metrics_r2_zero_variance_delegates_to_lib(model_metrics_activity):
-    # Arrange
+    """r2 zero-variance is now the lib's responsibility. Activity just passes through."""
-    # All target values are the same, so ss_tot will be 0
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -881,24 +897,18 @@ def test_calculate_simple_metrics_r2_zero_ss_tot(model_metrics_activity):
         'interval_minutes': 5,
     }
 
-    # Act
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+        [{'metric': 'r2', 'value': 0.0}]
-
-    # Assert
-    assert len(result['metric']) == 1
-    assert result['metric'].values[0] == 'r2'
-    assert result['value'].values[0] == 0.0  # Should return 0.0 when ss_tot == 0
-    assert result['model_id'].values[0] == 'test_model_id'
-    assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
-    assert result['data_size'].values[0] == 2
-    assert result['interval_minutes'].values[0] == 5
-    model_metrics_activity.info.assert_called_once_with(
-        "Calculating simple metrics for model test_model_id: ['r2']", metadata['metadata']
     )
 
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert result['value'].values[0] == 0.0
+    mock_instance.calculate.assert_called_once_with(['r2'])
+
 
 def test_calculate_simple_metrics_success_multiple_metrics_subset(model_metrics_activity):
-    # Arrange
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'],
@@ -915,23 +925,26 @@ def test_calculate_simple_metrics_success_multiple_metrics_subset(model_metrics_
         'interval_minutes': 5,
     }
 
-    # Act
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+        [
-
+            {'metric': 'rmse', 'value': 0.1},
-    # Assert
+            {'metric': 'mae', 'value': 0.1},
-    assert len(result['metric']) == 2
+        ]
-    assert 'rmse' in result['metric'].values
-    assert 'mae' in result['metric'].values
-    assert all(model_id == 'test_model_id' for model_id in result['model_id'].values)
-    assert all(timestamp == '2023-05-26 11:12:29' for timestamp in result['timestamp'].values)
-    assert all(data_size == 3 for data_size in result['data_size'].values)
-    assert all(interval_minutes == 5 for interval_minutes in result['interval_minutes'].values)
-    model_metrics_activity.info.assert_called_once_with(
-        "Calculating simple metrics for model test_model_id: ['rmse', 'mae']", metadata['metadata']
     )
 
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
 
-def test_calculate_simple_metrics_unknown_metric_ignored(model_metrics_activity):
+    assert len(result) == 2
+    assert set(result['metric'].values) == {'rmse', 'mae'}
+    assert all(mid == 'test_model_id' for mid in result['model_id'].values)
+    assert all(ts == '2023-05-26 11:12:29' for ts in result['timestamp'].values)
+    assert all(ds == 3 for ds in result['data_size'].values)
+    assert all(im == 5 for im in result['interval_minutes'].values)
+    mock_instance.calculate.assert_called_once_with(['rmse', 'mae'])
+
+
+def test_calculate_simple_metrics_unknown_metric_raises(model_metrics_activity):
     target_data = DataFrame(
         {
             'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -948,7 +961,211 @@ def test_calculate_simple_metrics_unknown_metric_ignored(model_metrics_activity)
         'interval_minutes': 5,
     }
 
-    result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+    mock_instance = MagicMock()
+    mock_instance.calculate.side_effect = ValueError('Unknown metric: unknown_metric')
+    mock_class = MagicMock(return_value=mock_instance)
+    mock_class.is_r2_supported = MagicMock(return_value=True)
 
-    assert len(result['metric']) == 1
+    with patch('laborious.activities.model_metrics.RegressionMetrics', mock_class):
+        with raises(ValueError, match='Unknown metric'):
+            model_metrics_activity.calculate_simple_metrics(input_data)
+
+
+def test_calculate_simple_metrics_r2_skipped_for_nonlinear_model(model_metrics_activity):
+    """When model_type is non-linear, r2 is excluded and a warning notification fires."""
+    target_data = DataFrame(
+        {
+            'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
+            'target': [1.0, 2.0],
+            'prediction': [1.1, 2.1],
+        }
+    )
+
+    input_data = {
+        **metadata,
+        'model_id': 'test_model_id',
+        'target_data': target_data.to_dict(),
+        'metrics': ['rmse', 'r2'],
+        'interval_minutes': 5,
+        'model_type': 'XGBoost',
+    }
+
+    mock_instance = MagicMock()
+    mock_instance.calculate.return_value = [{'metric': 'rmse', 'value': 0.1}]
+    mock_class = MagicMock(return_value=mock_instance)
+    mock_class.is_r2_supported = MagicMock(return_value=False)
+
+    with patch('laborious.activities.model_metrics.RegressionMetrics', mock_class):
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 1
     assert result['metric'].values[0] == 'rmse'
+    mock_class.is_r2_supported.assert_called_once_with('XGBoost')
+    mock_instance.calculate.assert_called_once_with(['rmse'])
+    model_metrics_activity.warning.assert_called_once()
+    model_metrics_activity.send_notification.assert_called_once()
+    call_kwargs = model_metrics_activity.send_notification.call_args.kwargs
+    assert call_kwargs['notification_id'] == 'SIMPLE_METRICS_R2_UNSUPPORTED'
+    assert call_kwargs['level'] == NotificationLevel.WARNING
+
+
+def test_calculate_simple_metrics_no_model_type_includes_r2(model_metrics_activity):
+    """When model_type is None (legacy input), r2 is included without check."""
+    target_data = DataFrame(
+        {
+            'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
+            'target': [1.0, 2.0],
+            'prediction': [1.1, 2.1],
+        }
+    )
+
+    input_data = {
+        **metadata,
+        'model_id': 'test_model_id',
+        'target_data': target_data.to_dict(),
+        'metrics': ['r2'],
+        'interval_minutes': 5,
+        # no model_type key
+    }
+
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [{'metric': 'r2', 'value': 0.95}]
+    )
+
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert result['metric'].values[0] == 'r2'
+    mock_class.is_r2_supported.assert_not_called()
+
+
+def test_calculate_simple_metrics_threshold_breach_rmse(model_metrics_activity):
+    """When rmse exceeds rmse_max, a WARNING notification fires."""
+    target_data = DataFrame(
+        {
+            'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
+            'target': [1.0, 2.0],
+            'prediction': [1.1, 2.1],
+        }
+    )
+
+    input_data = {
+        **metadata,
+        'model_id': 'test_model_id',
+        'target_data': target_data.to_dict(),
+        'metrics': ['rmse'],
+        'interval_minutes': 5,
+        'thresholds': {'rmse_max': 0.05},
+    }
+
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [
+            {'metric': 'rmse', 'value': 0.1},
+        ]
+    )
+
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 1
+    model_metrics_activity.send_notification.assert_called_once()
+    call_kwargs = model_metrics_activity.send_notification.call_args.kwargs
+    assert call_kwargs['notification_id'] == 'SIMPLE_METRICS_THRESHOLD_BREACH'
+    assert call_kwargs['level'] == NotificationLevel.WARNING
+    assert 'rmse' in call_kwargs['message']
+
+
+def test_calculate_simple_metrics_threshold_no_breach(model_metrics_activity):
+    """When rmse is below rmse_max, no notification fires."""
+    target_data = DataFrame(
+        {
+            'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
+            'target': [1.0, 2.0],
+            'prediction': [1.1, 2.1],
+        }
+    )
+
+    input_data = {
+        **metadata,
+        'model_id': 'test_model_id',
+        'target_data': target_data.to_dict(),
+        'metrics': ['rmse'],
+        'interval_minutes': 5,
+        'thresholds': {'rmse_max': 1.0},
+    }
+
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [
+            {'metric': 'rmse', 'value': 0.1},
+        ]
+    )
+
+    with patcher:
+        model_metrics_activity.calculate_simple_metrics(input_data)
+
+    model_metrics_activity.send_notification.assert_not_called()
+
+
+def test_calculate_simple_metrics_threshold_r2_below_min(model_metrics_activity):
+    """When r2 drops below r2_min, a WARNING notification fires."""
+    target_data = DataFrame(
+        {
+            'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
+            'target': [1.0, 2.0],
+            'prediction': [1.1, 2.1],
+        }
+    )
+
+    input_data = {
+        **metadata,
+        'model_id': 'test_model_id',
+        'target_data': target_data.to_dict(),
+        'metrics': ['r2'],
+        'interval_minutes': 5,
+        'thresholds': {'r2_min': 0.95},
+    }
+
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [
+            {'metric': 'r2', 'value': 0.8},
+        ]
+    )
+
+    with patcher:
+        model_metrics_activity.calculate_simple_metrics(input_data)
+
+    model_metrics_activity.send_notification.assert_called_once()
+    call_kwargs = model_metrics_activity.send_notification.call_args.kwargs
+    assert 'r2' in call_kwargs['message']
+
+
+def test_calculate_simple_metrics_no_thresholds_no_alert(model_metrics_activity):
+    """When thresholds is None (not configured), no alerting, no crash."""
+    target_data = DataFrame(
+        {
+            'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
+            'target': [1.0, 2.0],
+            'prediction': [1.1, 2.1],
+        }
+    )
+
+    input_data = {
+        **metadata,
+        'model_id': 'test_model_id',
+        'target_data': target_data.to_dict(),
+        'metrics': ['rmse'],
+        'interval_minutes': 5,
+        # no thresholds key
+    }
+
+    patcher, mock_class, mock_instance = _mock_regression_metrics_class(
+        [
+            {'metric': 'rmse', 'value': 999.0},
+        ]
+    )
+
+    with patcher:
+        result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
+
+    assert len(result) == 1
+    model_metrics_activity.send_notification.assert_not_called()

tests/laborious/workflows/test_simple_metrics.py

@@ -103,6 +103,8 @@ async def test_run(workflow_mock: AsyncMock, simple_metrics: SimpleMetrics):
             'target_data': target_data,
             'metrics': input_data['metrics'],
             'interval_minutes': input_data['interval_minutes'],
+            'model_type': None,
+            'thresholds': None,
         },
         retry_policy=ANY,
         start_to_close_timeout=ANY,
@@ -209,6 +211,8 @@ async def test_run_default_metrics(workflow_mock: AsyncMock, simple_metrics: Sim
             'target_data': target_data,
             'metrics': ['rmse', 'mse', 'mae', 'r2'],  # Default value
             'interval_minutes': input_data['interval_minutes'],
+            'model_type': None,
+            'thresholds': None,
         },
         retry_policy=ANY,
         start_to_close_timeout=ANY,