from unittest.mock import ANY, MagicMock, patch from pandas import DataFrame, Series, Timestamp from pytest import fixture, mark, raises from sientia_do.notifications.models import NotificationLevel 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 @fixture def model_metrics_activity(): model_metrics = ModelMetrics( logger=MagicMock(), notification_handler=MagicMock(), metrics_controller=MagicMock(spec=MetricsController), ) model_metrics.error = MagicMock() model_metrics.debug = MagicMock() model_metrics.info = MagicMock() model_metrics.warning = MagicMock() model_metrics.critical = MagicMock() model_metrics.send_notification = MagicMock() model_metrics.emit_metric_sync = MagicMock() model_metrics.get_core_labels = MagicMock( return_value={ 'pod_id': 'test_pod', 'model_name': 'test_model', 'workflow_name': 'test_workflow', } ) model_metrics.observe_lag_sync = MagicMock() model_metrics.pod_id = 'test_pod' return model_metrics metadata = { 'metadata': { 'model_id': 'test_model', 'model_name': 'test_model', 'workflow_name': 'test_workflow', 'schema_name': 'test_schedule', }, } 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 = { **metadata, 'model_name': 'test_model', 'model_id': 'test_model_id', 'reference_data': None, 'target_data': { 'timestamp': ['2023-05-26 11:12:27'], 'variable': ['feature1'], 'value': [1.0], }, 'target_name': 'target', 'drift_metrics': ['ks_test'], 'chunk_period': 'invalid', } # Act & Assert try: model_metrics_activity.calculate_drift(input_data) except ValueError as e: assert str(e) == 'Invalid chunk period: invalid, must be "min" or "s"' model_metrics_activity.error.assert_called_once_with( 'Invalid chunk period: invalid', metadata['metadata'] ) else: raise AssertionError('Expected ValueError') def _sample_drift_metrics_df(ts: Timestamp) -> DataFrame: """Minimal analyzer-shaped dataframe (univariate row + columns the activity expects).""" return DataFrame( { 'timestamp': [ts], 'feature': ['feature1'], 'method': ['ks_test'], 'value': [0.5], 'alert': [False], 'chunk_index': [0], 'chunk_start_date': [ts], 'chunk_end_date': [ts], 'threshold': [0.1], 'drift_type': ['univariate'], } ) def test_calculate_drift_with_reference_data(model_metrics_activity): ts = Timestamp('2023-05-26 11:12:27') drift_df = _sample_drift_metrics_df(ts) model_metrics_activity.get_drift_metrics = MagicMock(return_value=drift_df) reference_data = DataFrame( { 'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0], } ) input_data = { **metadata, 'model_name': 'test_model', 'model_id': 'test_model_id', 'reference_data': reference_data.to_dict('list'), 'target_data': { 'timestamp': ['2023-05-26 11:12:27'], 'variable': ['feature1'], 'value': [1.0], }, 'target_name': 'target', 'drift_metrics': ['ks_test'], 'chunk_period': 'min', } result = model_metrics_activity.calculate_drift(input_data) expected_timestamp = ts.tz_localize('UTC').strftime(DATETIME_FORMAT_WITH_TZ) assert result == [ { 'timestamp': expected_timestamp, 'feature': 'feature1', 'method': 'ks_test', 'value': 0.5, 'alert': False, 'chunk_index': 0, 'chunk_start_date': ts.isoformat(), 'chunk_end_date': ts.isoformat(), 'model_id': 'test_model_id', 'accurate': True, } ] model_metrics_activity.info.assert_called() model_metrics_activity.get_drift_metrics.assert_called_once() def test_calculate_drift_without_reference_data(model_metrics_activity): # Ten rows so int(len * 0.3) >= 1 for the built-in reference slice. ts_last = Timestamp('2023-05-26 11:12:36') drift_df = _sample_drift_metrics_df(ts_last) model_metrics_activity.get_drift_metrics = MagicMock(return_value=drift_df) timestamps = [f'2023-05-26 11:12:{27 + i:02d}' for i in range(10)] target_data_dict = { 'timestamp': timestamps, 'variable': ['feature1'] * 10, 'value': [float(i) for i in range(10)], } input_data = { **metadata, 'model_name': 'test_model', 'model_id': 'test_model_id', 'reference_data': None, 'target_data': target_data_dict, 'target_name': 'target', 'drift_metrics': ['ks_test'], 'chunk_period': 's', } result = model_metrics_activity.calculate_drift(input_data) expected_timestamp = ts_last.tz_localize('UTC').strftime(DATETIME_FORMAT_WITH_TZ) assert result == [ { 'timestamp': expected_timestamp, 'feature': 'feature1', 'method': 'ks_test', 'value': 0.5, 'alert': False, 'chunk_index': 0, 'chunk_start_date': ts_last.isoformat(), 'chunk_end_date': ts_last.isoformat(), 'model_id': 'test_model_id', 'accurate': False, } ] model_metrics_activity.warning.assert_called() model_metrics_activity.send_notification.assert_called_once_with( metadata=metadata['metadata'], notification_id='MODEL_METRICS_REFERENCE_DATA_WARNING', message='Using 30% first rows of target data as reference data', block='model_metrics', level=NotificationLevel.WARNING, attachment_content=ANY, ) def test_calculate_drift_empty_drift_df(model_metrics_activity): """Empty analyzer merge yields no rows and no insufficient-data alert (lib owns that failure mode).""" ts = Timestamp('2023-05-26 11:12:27') empty_df = _sample_drift_metrics_df(ts).iloc[0:0] model_metrics_activity.get_drift_metrics = MagicMock(return_value=empty_df) reference_data = DataFrame( { 'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0], } ) input_data = { **metadata, 'model_name': 'test_model', 'model_id': 'test_model_id', 'reference_data': reference_data.to_dict(), 'target_data': { 'timestamp': ['2023-05-26 11:12:27'], 'variable': ['feature1'], 'value': [1.0], }, 'target_name': 'target', 'drift_metrics': ['ks_test'], 'chunk_period': 'min', } result = model_metrics_activity.calculate_drift(input_data) assert result == [] model_metrics_activity.send_notification.assert_not_called() def test_calculate_drift_empty_after_timestamp_filter(model_metrics_activity): """Rows dropped by target-window alignment yield an empty export list, not an insufficient-data error.""" drift_df = _sample_drift_metrics_df(Timestamp('2020-01-01')) model_metrics_activity.get_drift_metrics = MagicMock(return_value=drift_df) reference_data = DataFrame( { 'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0], } ) input_data = { **metadata, 'model_name': 'test_model', 'model_id': 'test_model_id', 'reference_data': reference_data.to_dict(), 'target_data': { 'timestamp': ['2023-05-26 11:12:27'], 'variable': ['feature1'], 'value': [1.0], }, 'target_name': 'target', 'drift_metrics': ['ks_test'], 'chunk_period': 'min', } result = model_metrics_activity.calculate_drift(input_data) assert result == [] model_metrics_activity.send_notification.assert_not_called() @patch('laborious.activities.model_metrics.DataFrame') @patch('laborious.activities.model_metrics.to_datetime') def test_calculate_drift_drift_insufficient_data_error_from_lib( mock_to_datetime, mock_dataframe, model_metrics_activity ): """``DriftInsufficientDataError`` maps to MODEL_METRICS_DRIFT_INSUFFICIENT_DATA, not GET error.""" mock_to_datetime.return_value.dt.strftime.return_value = '2023-05-26 11:12:27' lib_msg = ( '[MODEL_METRICS_DRIFT_INSUFFICIENT_DATA] Drift analysis produced no time chunks ' "(chunk_period='min', analysis_rows=1)." ) model_metrics_activity.get_drift_metrics = MagicMock( side_effect=DriftInsufficientDataError(lib_msg, analysis_rows=1) ) reference_data = DataFrame( { 'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0], } ) mock_target_df = MagicMock() mock_target_df.pivot.return_value = mock_target_df mock_target_df.index = ['2023-05-26 11:12:27'] mock_target_df.reset_index.return_value = mock_target_df mock_target_df.dropna.return_value = mock_target_df mock_target_df.__getitem__.return_value.apply.return_value = ['2023-05-26 11:12:27'] mock_target_df.drop.return_value.columns = ['feature1'] mock_dataframe.return_value = mock_target_df input_data = { **metadata, 'model_name': 'test_model', 'model_id': 'test_model_id', 'reference_data': reference_data.to_dict(), 'target_data': { 'timestamp': ['2023-05-26 11:12:27'], 'variable': ['feature1'], 'value': [1.0], }, 'target_name': 'target', 'drift_metrics': ['ks_test'], 'chunk_period': 'min', } with raises(DriftInsufficientDataError, match='MODEL_METRICS_DRIFT_INSUFFICIENT_DATA'): model_metrics_activity.calculate_drift(input_data) model_metrics_activity.error.assert_called_once_with(lib_msg, metadata['metadata']) model_metrics_activity.send_notification.assert_called_once_with( metadata=metadata['metadata'], notification_id='MODEL_METRICS_DRIFT_INSUFFICIENT_DATA', message=lib_msg, block='model_metrics', level=NotificationLevel.ERROR, attachment_content=ANY, ) def test_calculate_drift_success_min(model_metrics_activity): ts = Timestamp('2023-05-26 11:12:27') drift_df = _sample_drift_metrics_df(ts) model_metrics_activity.get_drift_metrics = MagicMock(return_value=drift_df) reference_data = DataFrame( { 'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0], } ) input_data = { **metadata, 'model_name': 'test_model', 'model_id': 'test_model_id', 'reference_data': reference_data.to_dict('list'), 'target_data': { 'timestamp': ['2023-05-26 11:12:27'], 'variable': ['feature1'], 'value': [1.0], }, 'target_name': 'target', 'drift_metrics': ['ks_test'], 'chunk_period': 'min', } result = model_metrics_activity.calculate_drift(input_data) expected_timestamp = ts.tz_localize('UTC').strftime(DATETIME_FORMAT_WITH_TZ) assert result == [ { 'timestamp': expected_timestamp, 'feature': 'feature1', 'method': 'ks_test', 'value': 0.5, 'alert': False, 'chunk_index': 0, 'chunk_start_date': ts.isoformat(), 'chunk_end_date': ts.isoformat(), 'model_id': 'test_model_id', 'accurate': True, } ] model_metrics_activity.info.assert_called() model_metrics_activity.get_drift_metrics.assert_called_once() def test_calculate_drift_success_s(model_metrics_activity): ts = Timestamp('2023-05-26 11:12:27') drift_df = _sample_drift_metrics_df(ts) model_metrics_activity.get_drift_metrics = MagicMock(return_value=drift_df) reference_data = DataFrame( { 'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0], } ) input_data = { **metadata, 'model_name': 'test_model', 'model_id': 'test_model_id', 'reference_data': reference_data.to_dict('list'), 'target_data': { 'timestamp': ['2023-05-26 11:12:27'], 'variable': ['feature1'], 'value': [1.0], }, 'target_name': 'target', 'drift_metrics': ['ks_test'], 'chunk_period': 's', } result = model_metrics_activity.calculate_drift(input_data) expected_timestamp = ts.tz_localize('UTC').strftime(DATETIME_FORMAT_WITH_TZ) assert result == [ { 'timestamp': expected_timestamp, 'feature': 'feature1', 'method': 'ks_test', 'value': 0.5, 'alert': False, 'chunk_index': 0, 'chunk_start_date': ts.isoformat(), 'chunk_end_date': ts.isoformat(), 'model_id': 'test_model_id', 'accurate': True, } ] model_metrics_activity.info.assert_called() model_metrics_activity.get_drift_metrics.assert_called_once() @patch('laborious.activities.model_metrics.DataFrame') @patch('laborious.activities.model_metrics.to_datetime') def test_calculate_drift_get_drift_metrics_error( mock_to_datetime, mock_dataframe, model_metrics_activity ): # Arrange mock_to_datetime.return_value.dt.strftime.return_value = '2023-05-26 11:12:27' model_metrics_activity.get_drift_metrics = MagicMock( side_effect=Exception('Get drift metrics error') ) reference_data = DataFrame( { 'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0], } ) mock_target_df = MagicMock() mock_target_df.pivot.return_value = mock_target_df mock_target_df.index = ['2023-05-26 11:12:27'] mock_target_df.reset_index.return_value = mock_target_df mock_target_df.dropna.return_value = mock_target_df mock_target_df.__getitem__.return_value.apply.return_value = ['2023-05-26 11:12:27'] mock_target_df.drop.return_value.columns = ['feature1'] mock_dataframe.return_value = mock_target_df input_data = { **metadata, 'model_name': 'test_model', 'model_id': 'test_model_id', 'reference_data': reference_data.to_dict(), 'target_data': { 'timestamp': ['2023-05-26 11:12:27'], 'variable': ['feature1'], 'value': [1.0], }, 'target_name': 'target', 'drift_metrics': ['ks_test'], 'chunk_period': 'min', } # Act / Assert with raises(Exception, match='Get drift metrics error'): model_metrics_activity.calculate_drift(input_data) model_metrics_activity.error.assert_called_once_with( 'Error getting drift metrics: Get drift metrics error', metadata['metadata'] ) model_metrics_activity.send_notification.assert_called_once_with( metadata=metadata['metadata'], notification_id='MODEL_METRICS_GET_DRIFT_METRICS_ERROR', message='Error getting drift metrics: Get drift metrics error', block='model_metrics', level=NotificationLevel.ERROR, attachment_content=ANY, ) @patch('laborious.activities.model_metrics.to_datetime') @patch('laborious.activities.model_metrics.time.time') @patch('laborious.activities.model_metrics.DriftAnalysis') @patch('laborious.activities.model_metrics.metrics') def test_get_drift_metrics_success( mock_metrics, mock_model_analysis, mock_time, mock_to_datetime, model_metrics_activity ): # Arrange mock_time.return_value = 1000.0 mock_drift_df = DataFrame( { 'timestamp': ['2023-05-26 11:12:27'], 'method': ['ks_test'], 'value': [0.5], 'feature': ['feature1'], } ) mock_model_analysis.return_value.detect_univariate_drift.return_value = MagicMock() mock_model_analysis.return_value.detect_multivariate_drift.return_value = MagicMock() mock_model_analysis.return_value.get_drift_metrics_dataframe.return_value = mock_drift_df reference_data = DataFrame( { 'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0], } ) target_data = DataFrame( { 'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0], } ) reference_columns = reference_data.drop( columns=['target', 'timestamp'], errors='ignore' ).columns # Act result = model_metrics_activity.get_drift_metrics( reference_data=reference_data, target_data=target_data, target_name='target', reference_columns=reference_columns, drift_metrics=['ks_test'], chunk_period='min', metadata=metadata['metadata'], ) # Assert assert isinstance(result, DataFrame) model_metrics_activity.debug.assert_called() model_metrics_activity.observe_lag_sync.assert_called() model_metrics_activity.emit_metric_sync.assert_called() @patch('laborious.activities.model_metrics.to_datetime') @patch('laborious.activities.model_metrics.time.time') @patch('laborious.activities.model_metrics.DriftAnalysis') @patch('laborious.activities.model_metrics.metrics') def test_get_drift_metrics_univariate_error( mock_metrics, mock_model_analysis, mock_time, mock_to_datetime, model_metrics_activity ): # Arrange mock_time.return_value = 1000.0 mock_model_analysis.return_value.detect_univariate_drift.side_effect = Exception( 'Univariate drift error' ) reference_data = DataFrame( { 'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0], } ) target_data = DataFrame( { 'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0], } ) reference_columns = reference_data.drop( columns=['target', 'timestamp'], errors='ignore' ).columns # Act & Assert try: model_metrics_activity.get_drift_metrics( reference_data=reference_data, target_data=target_data, target_name='target', reference_columns=reference_columns, drift_metrics=['ks_test'], chunk_period='min', metadata=metadata['metadata'], ) except Exception as e: assert str(e) == 'Univariate drift error' model_metrics_activity.error.assert_called_once_with( 'Error detecting univariate drift: Univariate drift error', metadata['metadata'] ) model_metrics_activity.emit_metric_sync.assert_called_with( metric_object=mock_metrics.MODEL_ANALYZE_ERROR_COUNT, tags=ANY ) else: raise AssertionError('Expected Exception') @patch('laborious.activities.model_metrics.to_datetime') @patch('laborious.activities.model_metrics.time.time') @patch('laborious.activities.model_metrics.DriftAnalysis') @patch('laborious.activities.model_metrics.metrics') def test_get_drift_metrics_multivariate_error( mock_metrics, mock_model_analysis, mock_time, mock_to_datetime, model_metrics_activity ): mock_time.return_value = 1000.0 mock_model_analysis.return_value.detect_univariate_drift.return_value = MagicMock() mock_model_analysis.return_value.detect_multivariate_drift.side_effect = Exception( 'Multivariate drift error' ) reference_data = DataFrame( {'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0]} ) target_data = DataFrame( {'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0]} ) reference_columns = reference_data.drop( columns=['target', 'timestamp'], errors='ignore' ).columns try: model_metrics_activity.get_drift_metrics( reference_data=reference_data, target_data=target_data, target_name='target', reference_columns=reference_columns, drift_metrics=['ks_test'], chunk_period='min', metadata=metadata['metadata'], ) except Exception as e: assert str(e) == 'Multivariate drift error' model_metrics_activity.error.assert_called_once_with( 'Error detecting multivariate drift: Multivariate drift error', metadata['metadata'] ) model_metrics_activity.emit_metric_sync.assert_called_with( metric_object=mock_metrics.MODEL_ANALYZE_ERROR_COUNT, tags=ANY ) else: raise AssertionError('Expected Exception') @patch('laborious.activities.model_metrics.to_datetime') @patch('laborious.activities.model_metrics.time.time') @patch('laborious.activities.model_metrics.DriftAnalysis') @patch('laborious.activities.model_metrics.metrics') def test_get_drift_metrics_dataframe_error( mock_metrics, mock_model_analysis, mock_time, mock_to_datetime, model_metrics_activity ): mock_time.return_value = 1000.0 mock_model_analysis.return_value.detect_univariate_drift.return_value = MagicMock() mock_model_analysis.return_value.detect_multivariate_drift.return_value = MagicMock() mock_model_analysis.return_value.get_drift_metrics_dataframe.side_effect = Exception( 'Dataframe error' ) reference_data = DataFrame( {'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0]} ) target_data = DataFrame( {'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0]} ) reference_columns = reference_data.drop( columns=['target', 'timestamp'], errors='ignore' ).columns try: model_metrics_activity.get_drift_metrics( reference_data=reference_data, target_data=target_data, target_name='target', reference_columns=reference_columns, drift_metrics=['ks_test'], chunk_period='min', metadata=metadata['metadata'], ) except Exception as e: assert str(e) == 'Dataframe error' model_metrics_activity.error.assert_called_once_with( 'Error building drift metrics dataframe: Dataframe error', metadata['metadata'] ) model_metrics_activity.emit_metric_sync.assert_called_with( metric_object=mock_metrics.MODEL_ANALYZE_ERROR_COUNT, tags=ANY ) else: raise AssertionError('Expected Exception') def test_calculate_simple_metrics_success_all_metrics(model_metrics_activity): target_data = DataFrame( { 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'], 'target': [1.0, 2.0, 3.0], 'prediction': [1.1, 2.1, 2.9], } ) input_data = { **metadata, 'model_id': 'test_model_id', 'target_data': target_data.to_dict(), 'metrics': ['rmse', 'mse', 'mae', 'r2'], 'interval_minutes': 5, } patcher, mock_class, mock_instance = _mock_regression_metrics_class( [ {'metric': 'rmse', 'value': 0.1}, {'metric': 'mse', 'value': 0.01}, {'metric': 'mae', 'value': 0.1}, {'metric': 'r2', 'value': 0.99}, ] ) 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): 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, } 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 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 mock_instance.calculate.assert_called_once_with(['rmse']) def test_calculate_simple_metrics_success_mse_only(model_metrics_activity): 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': ['mse'], 'interval_minutes': 5, } patcher, mock_class, mock_instance = _mock_regression_metrics_class( [{'metric': 'mse', 'value': 0.01}] ) with patcher: 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 mock_instance.calculate.assert_called_once_with(['mse']) def test_calculate_simple_metrics_success_mae_only(model_metrics_activity): 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': ['mae'], 'interval_minutes': 5, } patcher, mock_class, mock_instance = _mock_regression_metrics_class( [{'metric': 'mae', 'value': 0.1}] ) with patcher: 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 mock_instance.calculate.assert_called_once_with(['mae']) def test_calculate_simple_metrics_success_r2_only(model_metrics_activity): 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, } 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 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 mock_instance.calculate.assert_called_once_with(['r2']) def test_calculate_simple_metrics_r2_zero_variance_delegates_to_lib(model_metrics_activity): """r2 zero-variance is now the lib's responsibility. Activity just passes through.""" target_data = DataFrame( { 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'], 'target': [1.0, 1.0], 'prediction': [1.1, 1.1], } ) input_data = { **metadata, 'model_id': 'test_model_id', 'target_data': target_data.to_dict(), 'metrics': ['r2'], 'interval_minutes': 5, } patcher, mock_class, mock_instance = _mock_regression_metrics_class( [{'metric': 'r2', 'value': 0.0}] ) 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): target_data = DataFrame( { 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'], 'target': [1.0, 2.0, 3.0], 'prediction': [1.1, 2.1, 2.9], } ) input_data = { **metadata, 'model_id': 'test_model_id', 'target_data': target_data.to_dict(), 'metrics': ['rmse', 'mae'], 'interval_minutes': 5, } patcher, mock_class, mock_instance = _mock_regression_metrics_class( [ {'metric': 'rmse', 'value': 0.1}, {'metric': 'mae', 'value': 0.1}, ] ) with patcher: result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data)) 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'], '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': ['unknown_metric', 'rmse'], 'interval_minutes': 5, } 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) 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'] assert '(window 5 min)' in call_kwargs['message'] # Model-level trigger so the orchestrator dedup merges windows (spec decision 9). assert call_kwargs['metadata']['schedule_name'] == 'test_model_id-simple_metrics' assert call_kwargs['metadata']['model_name'] == 'test_model' 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() @mark.parametrize( 'failing_stage', ['detect_univariate_drift', 'detect_multivariate_drift', 'get_drift_metrics_dataframe'], ) @patch('laborious.activities.model_metrics.DriftAnalysis') def test_get_drift_metrics_reraises_insufficient_data_untouched( mock_drift_analysis, failing_stage, model_metrics_activity ): """Insufficient data is not a stage failure: it propagates without the stage error report.""" error = DriftInsufficientDataError('not enough rows', analysis_rows=1) getattr(mock_drift_analysis.return_value, failing_stage).side_effect = error data = DataFrame({'timestamp': ['2023-05-26 11:12:27'], 'target': [1.0], 'feature1': [1.0]}) model_metrics_activity._drift_analyze_stage_error = MagicMock() with raises(DriftInsufficientDataError) as exc_info: model_metrics_activity.get_drift_metrics( reference_data=data, target_data=data, target_name='target', reference_columns=data.drop(columns=['target', 'timestamp']).columns, drift_metrics=['ks_test'], chunk_period='min', metadata=metadata['metadata'], ) assert exc_info.value is error model_metrics_activity._drift_analyze_stage_error.assert_not_called() def test_to_naive_utc_converts_aware_timestamps_to_naive_utc(): series = Series(['2024-01-01T03:00:00-03:00', '2024-01-01T09:00:00-03:00']) result = ModelMetrics._to_naive_utc(series) assert result.dt.tz is None assert result.tolist() == [Timestamp('2024-01-01 06:00:00'), Timestamp('2024-01-01 12:00:00')] def test_to_naive_utc_keeps_naive_timestamps(): series = Series(['2024-01-01 06:00:00']) result = ModelMetrics._to_naive_utc(series) assert result.dt.tz is None assert result.tolist() == [Timestamp('2024-01-01 06:00:00')] @mark.parametrize( 'rows, thresholds, expected_count', [ ([{'metric': 'rmse', 'value': 2.0}], {'rmse_max': 1.0}, 1), ([{'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': 'mae', 'value': 99.0}], {'rmse_max': 1.0}, 0), ([{'metric': 'rmse', 'value': 99.0}], None, 0), ([], {'rmse_max': 1.0}, 0), ( [{'metric': 'rmse', 'value': 2.0}, {'metric': 'r2', 'value': 0.1}], {'rmse_max': 1.0, 'r2_min': 0.9}, 2, ), ], ) def test_threshold_breaches(rows, thresholds, expected_count): assert len(threshold_breaches(rows, thresholds)) == expected_count