Files
sientia-dataops-laborious-l…/tests/laborious/activities/test_model_metrics.py

1224 lines
40 KiB
Python

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
@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')]