1 Commits

Author SHA1 Message Date
Eduardo Rios
d5c1213293 SIENTIAPDE-2072: bump sientia_do pin to 1.12.2
Picks up the notification timestamp -> native datetime fix.
2026-08-17 16:43:02 -03:00
6 changed files with 126 additions and 381 deletions

View File

@@ -6,6 +6,7 @@ with workflow.unsafe.imports_passed_through():
import warnings import warnings
from typing import Any from typing import Any
import numpy as np
import pandas as pd import pandas as pd
from pandas import DataFrame, Index, Series, to_datetime from pandas import DataFrame, Index, Series, to_datetime
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
@@ -15,7 +16,6 @@ with workflow.unsafe.imports_passed_through():
from sientia_do.observability.sientia_monitoring import SientiaMonitoring from sientia_do.observability.sientia_monitoring import SientiaMonitoring
from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
from sientia_model.analytics.drift_analysis import DriftAnalysis, DriftInsufficientDataError from sientia_model.analytics.drift_analysis import DriftAnalysis, DriftInsufficientDataError
from sientia_model.metrics.regression import RegressionMetrics
from laborious import metrics from laborious import metrics
from laborious.utils.dataframe_debug import build_dataframe_debug_message from laborious.utils.dataframe_debug import build_dataframe_debug_message
@@ -365,62 +365,62 @@ class ModelMetrics(SientiaMonitoring):
@activity.defn(name='calculate_simple_metrics') @activity.defn(name='calculate_simple_metrics')
def calculate_simple_metrics(self, input_data: dict[str, Any]) -> list[dict]: def calculate_simple_metrics(self, input_data: dict[str, Any]) -> list[dict]:
""" """
Calculate simple regression metrics for a model using RegressionMetrics. Calculate simple metrics for a model. Metrics available are:
- rmse
- mse
- mae
- r2
- accuracy
- precision
- recall
- f1
Args: Args:
input_data: Input data containing: input_data (dict[str, Any]): Input data containing:
- metadata (dict): Workflow execution metadata - metadata (dict): Workflow execution metadata
- model_id (str): ID of the MLFlow model - model_id (str): ID of the MLFlow model
- target_data (list[dict]): Target data with target, prediction, timestamp columns - target_data (pd.DataFrame): Target data for calculating metrics, containing target and prediction columns
- metrics (list[str]): Metric names to calculate - metrics (list[str]): List of metrics to calculate
- interval_minutes (int): Window interval in minutes
- model_type (str | None): Model algorithm type (for r2 lock)
Returns: Returns:
list[dict]: Records with metric, value, model_id, timestamp, data_size, interval_minutes dict[Hashable, Any]: Dictionary containing the calculated metrics
""" """
metadata = input_data['metadata'] metadata = input_data['metadata']
model_id = input_data['model_id'] model_id = input_data['model_id']
target_data = DataFrame(input_data['target_data']) target_data = DataFrame(input_data['target_data'])
metric_names = list(input_data['metrics']) metric_names = input_data['metrics']
interval_minutes = input_data['interval_minutes'] interval_minutes = input_data['interval_minutes']
model_type = input_data.get('model_type')
data_size = target_data.shape[0] data_size = target_data.shape[0]
# Filter r2 when model_type is known and unsupported output_data = []
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,
)
if not metric_names: diff = target_data['target'] - target_data['prediction']
self.info(f'No metrics to calculate for model {model_id} after filtering', metadata) diff_squared = diff**2
return []
self.info(f'Calculating simple metrics for model {model_id}: {metric_names}', metadata) self.info(f'Calculating simple metrics for model {model_id}: {metric_names}', metadata)
# Build Series with DatetimeIndex for RegressionMetrics for metric in metric_names:
timestamps = pd.to_datetime(target_data['timestamp']) if metric == 'rmse':
real_data = Series(target_data['target'].values, index=timestamps, dtype=float) output_data.append({'metric': 'rmse', 'value': np.sqrt(np.mean(diff_squared))})
predictions = Series(target_data['prediction'].values, index=timestamps, dtype=float) elif metric == 'mse':
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)
regression = RegressionMetrics(real_data, predictions) ss_res = np.sum(diff_squared)
output_data = regression.calculate(metric_names) ss_tot = np.sum((y_true - y_mean) ** 2)
# 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 = DataFrame(output_data)
data['model_id'] = model_id data['model_id'] = model_id
data['timestamp'] = target_data['timestamp'].max() data['timestamp'] = target_data['timestamp'].max()
@@ -429,34 +429,4 @@ class ModelMetrics(SientiaMonitoring):
self._debug_dataframe(f'Simple metrics dataframe: Size {data.shape}', data, metadata) 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') return data.to_dict(orient='records')

View File

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

View File

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

View File

@@ -3,8 +3,8 @@ psycopg2-binary
sqlalchemy sqlalchemy
asyncua==1.0.6 asyncua==1.0.6
redis redis
sientia_do>=1.12.1 sientia_do>=1.12.2
sientia_model>=0.12.0 sientia_model>=0.8.2
prometheus-client prometheus-client
botocore botocore
boto3 boto3

View File

@@ -45,23 +45,6 @@ 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): def test_calculate_drift_invalid_chunk_period(model_metrics_activity):
# Arrange # Arrange
input_data = { input_data = {
@@ -710,6 +693,7 @@ def test_get_drift_metrics_dataframe_error(
def test_calculate_simple_metrics_success_all_metrics(model_metrics_activity): def test_calculate_simple_metrics_success_all_metrics(model_metrics_activity):
# Arrange
target_data = DataFrame( target_data = DataFrame(
{ {
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'], 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'],
@@ -726,28 +710,28 @@ def test_calculate_simple_metrics_success_all_metrics(model_metrics_activity):
'interval_minutes': 5, 'interval_minutes': 5,
} }
patcher, mock_class, mock_instance = _mock_regression_metrics_class( # Act
[
{'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)) result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
assert len(result) == 4 # Assert
assert set(result['metric'].values) == {'rmse', 'mse', 'mae', 'r2'} assert len(result['metric']) == 4
assert all(mid == 'test_model_id' for mid in result['model_id'].values) assert 'rmse' in result['metric'].values
assert all(ts == '2023-05-26 11:12:29' for ts in result['timestamp'].values) assert 'mse' in result['metric'].values
assert all(ds == 3 for ds in result['data_size'].values) assert 'mae' in result['metric'].values
assert all(im == 5 for im in result['interval_minutes'].values) assert 'r2' in result['metric'].values
mock_instance.calculate.assert_called_once_with(['rmse', 'mse', 'mae', 'r2']) 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()
def test_calculate_simple_metrics_success_rmse_only(model_metrics_activity): def test_calculate_simple_metrics_success_rmse_only(model_metrics_activity):
# Arrange
target_data = DataFrame( target_data = DataFrame(
{ {
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'], 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -764,23 +748,23 @@ def test_calculate_simple_metrics_success_rmse_only(model_metrics_activity):
'interval_minutes': 5, 'interval_minutes': 5,
} }
patcher, mock_class, mock_instance = _mock_regression_metrics_class( # Act
[{'metric': 'rmse', 'value': 0.1}]
)
with patcher:
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data)) result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
assert len(result) == 1 # Assert
assert len(result['metric']) == 1
assert result['metric'].values[0] == 'rmse' assert result['metric'].values[0] == 'rmse'
assert result['model_id'].values[0] == 'test_model_id' assert result['model_id'].values[0] == 'test_model_id'
assert result['timestamp'].values[0] == '2023-05-26 11:12:28' assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
assert result['data_size'].values[0] == 2 assert result['data_size'].values[0] == 2
assert result['interval_minutes'].values[0] == 5 assert result['interval_minutes'].values[0] == 5
mock_instance.calculate.assert_called_once_with(['rmse']) model_metrics_activity.info.assert_called_once_with(
"Calculating simple metrics for model test_model_id: ['rmse']", metadata['metadata']
)
def test_calculate_simple_metrics_success_mse_only(model_metrics_activity): def test_calculate_simple_metrics_success_mse_only(model_metrics_activity):
# Arrange
target_data = DataFrame( target_data = DataFrame(
{ {
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'], 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -797,23 +781,23 @@ def test_calculate_simple_metrics_success_mse_only(model_metrics_activity):
'interval_minutes': 5, 'interval_minutes': 5,
} }
patcher, mock_class, mock_instance = _mock_regression_metrics_class( # Act
[{'metric': 'mse', 'value': 0.01}]
)
with patcher:
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data)) result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
assert len(result) == 1 # Assert
assert len(result['metric']) == 1
assert result['metric'].values[0] == 'mse' assert result['metric'].values[0] == 'mse'
assert result['model_id'].values[0] == 'test_model_id' assert result['model_id'].values[0] == 'test_model_id'
assert result['timestamp'].values[0] == '2023-05-26 11:12:28' assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
assert result['data_size'].values[0] == 2 assert result['data_size'].values[0] == 2
assert result['interval_minutes'].values[0] == 5 assert result['interval_minutes'].values[0] == 5
mock_instance.calculate.assert_called_once_with(['mse']) model_metrics_activity.info.assert_called_once_with(
"Calculating simple metrics for model test_model_id: ['mse']", metadata['metadata']
)
def test_calculate_simple_metrics_success_mae_only(model_metrics_activity): def test_calculate_simple_metrics_success_mae_only(model_metrics_activity):
# Arrange
target_data = DataFrame( target_data = DataFrame(
{ {
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'], 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -830,23 +814,23 @@ def test_calculate_simple_metrics_success_mae_only(model_metrics_activity):
'interval_minutes': 5, 'interval_minutes': 5,
} }
patcher, mock_class, mock_instance = _mock_regression_metrics_class( # Act
[{'metric': 'mae', 'value': 0.1}]
)
with patcher:
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data)) result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
assert len(result) == 1 # Assert
assert len(result['metric']) == 1
assert result['metric'].values[0] == 'mae' assert result['metric'].values[0] == 'mae'
assert result['model_id'].values[0] == 'test_model_id' assert result['model_id'].values[0] == 'test_model_id'
assert result['timestamp'].values[0] == '2023-05-26 11:12:28' assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
assert result['data_size'].values[0] == 2 assert result['data_size'].values[0] == 2
assert result['interval_minutes'].values[0] == 5 assert result['interval_minutes'].values[0] == 5
mock_instance.calculate.assert_called_once_with(['mae']) model_metrics_activity.info.assert_called_once_with(
"Calculating simple metrics for model test_model_id: ['mae']", metadata['metadata']
)
def test_calculate_simple_metrics_success_r2_only(model_metrics_activity): def test_calculate_simple_metrics_success_r2_only(model_metrics_activity):
# Arrange
target_data = DataFrame( target_data = DataFrame(
{ {
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'], 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -863,24 +847,24 @@ def test_calculate_simple_metrics_success_r2_only(model_metrics_activity):
'interval_minutes': 5, 'interval_minutes': 5,
} }
patcher, mock_class, mock_instance = _mock_regression_metrics_class( # Act
[{'metric': 'r2', 'value': 0.95}]
)
with patcher:
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data)) result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
assert len(result) == 1 # Assert
assert len(result['metric']) == 1
assert result['metric'].values[0] == 'r2' assert result['metric'].values[0] == 'r2'
assert result['model_id'].values[0] == 'test_model_id' assert result['model_id'].values[0] == 'test_model_id'
assert result['timestamp'].values[0] == '2023-05-26 11:12:28' assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
assert result['data_size'].values[0] == 2 assert result['data_size'].values[0] == 2
assert result['interval_minutes'].values[0] == 5 assert result['interval_minutes'].values[0] == 5
mock_instance.calculate.assert_called_once_with(['r2']) model_metrics_activity.info.assert_called_once_with(
"Calculating simple metrics for model test_model_id: ['r2']", metadata['metadata']
)
def test_calculate_simple_metrics_r2_zero_variance_delegates_to_lib(model_metrics_activity): def test_calculate_simple_metrics_r2_zero_ss_tot(model_metrics_activity):
"""r2 zero-variance is now the lib's responsibility. Activity just passes through.""" # Arrange
# All target values are the same, so ss_tot will be 0
target_data = DataFrame( target_data = DataFrame(
{ {
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'], 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -897,18 +881,24 @@ def test_calculate_simple_metrics_r2_zero_variance_delegates_to_lib(model_metric
'interval_minutes': 5, 'interval_minutes': 5,
} }
patcher, mock_class, mock_instance = _mock_regression_metrics_class( # Act
[{'metric': 'r2', 'value': 0.0}]
)
with patcher:
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data)) result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
assert result['value'].values[0] == 0.0 # Assert
mock_instance.calculate.assert_called_once_with(['r2']) 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']
)
def test_calculate_simple_metrics_success_multiple_metrics_subset(model_metrics_activity): def test_calculate_simple_metrics_success_multiple_metrics_subset(model_metrics_activity):
# Arrange
target_data = DataFrame( target_data = DataFrame(
{ {
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'], 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'],
@@ -925,26 +915,23 @@ def test_calculate_simple_metrics_success_multiple_metrics_subset(model_metrics_
'interval_minutes': 5, 'interval_minutes': 5,
} }
patcher, mock_class, mock_instance = _mock_regression_metrics_class( # Act
[
{'metric': 'rmse', 'value': 0.1},
{'metric': 'mae', 'value': 0.1},
]
)
with patcher:
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data)) result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
assert len(result) == 2 # Assert
assert set(result['metric'].values) == {'rmse', 'mae'} assert len(result['metric']) == 2
assert all(mid == 'test_model_id' for mid in result['model_id'].values) assert 'rmse' in result['metric'].values
assert all(ts == '2023-05-26 11:12:29' for ts in result['timestamp'].values) assert 'mae' in result['metric'].values
assert all(ds == 3 for ds in result['data_size'].values) assert all(model_id == 'test_model_id' for model_id in result['model_id'].values)
assert all(im == 5 for im in result['interval_minutes'].values) assert all(timestamp == '2023-05-26 11:12:29' for timestamp in result['timestamp'].values)
mock_instance.calculate.assert_called_once_with(['rmse', 'mae']) 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']
)
def test_calculate_simple_metrics_unknown_metric_raises(model_metrics_activity): def test_calculate_simple_metrics_unknown_metric_ignored(model_metrics_activity):
target_data = DataFrame( target_data = DataFrame(
{ {
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'], 'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
@@ -961,211 +948,7 @@ def test_calculate_simple_metrics_unknown_metric_raises(model_metrics_activity):
'interval_minutes': 5, '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)) result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
assert len(result) == 1 assert len(result['metric']) == 1
assert result['metric'].values[0] == 'rmse' 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()

View File

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