Compare commits
15 Commits
e53e792e79
...
feature/SI
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
5997210118 | ||
|
|
4e3b5756be | ||
|
|
666188a7b3 | ||
|
|
b7fdcebc15 | ||
|
|
33b4ae8406 | ||
|
|
3b55072ab4 | ||
|
|
272e02dadc | ||
|
|
19c8a028d2 | ||
|
|
11691398da | ||
|
|
054dcbfa50 | ||
|
|
e393263ddf | ||
|
|
11224537d3 | ||
|
|
26502b5274 | ||
|
|
ad4da333a0 | ||
|
|
a4594997e8 |
@@ -142,7 +142,7 @@ def _recent_minute_timestamps(count: int, offset_minutes: int = 6) -> list[str]:
|
||||
def _configure_reference_csv(mlflow_repository_stub, reference_rows: pd.DataFrame) -> None:
|
||||
"""
|
||||
Wire ``mlflow_repository_stub`` so ``get_reference_data`` returns
|
||||
``reference_rows`` by writing them to ``dst_path/evaluation_data.csv``.
|
||||
``reference_rows`` by writing them to ``dst_path/retrain_input.csv``.
|
||||
|
||||
Args:
|
||||
- mlflow_repository_stub: External MLflow repository fixture.
|
||||
@@ -150,12 +150,16 @@ def _configure_reference_csv(mlflow_repository_stub, reference_rows: pd.DataFram
|
||||
"""
|
||||
|
||||
def _download(run_id: str, artifact_path: str, dst_path: str, metadata=None):
|
||||
target = Path(dst_path) / 'evaluation_data.csv'
|
||||
target = Path(dst_path) / artifact_path
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
reference_rows.to_csv(target, index=False)
|
||||
|
||||
mlflow_repository_stub._client.get_model_version_by_alias.return_value = MagicMock(
|
||||
run_id='fake-reference-run'
|
||||
)
|
||||
file_info = MagicMock()
|
||||
file_info.path = 'retrain_input.csv'
|
||||
mlflow_repository_stub._client.list_artifacts.return_value = [file_info]
|
||||
mlflow_repository_stub.download_artifacts.side_effect = _download
|
||||
|
||||
|
||||
|
||||
@@ -53,6 +53,7 @@ class MLFlow(SientiaMonitoring):
|
||||
|
||||
_MAX_DEBUG_DATAFRAME_ROWS = 100
|
||||
_DEFAULT_MODEL_ALIAS = 'production'
|
||||
_REFERENCE_ARTIFACT_CANDIDATES = ('evaluation_data.csv', 'test_data.csv')
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -527,7 +528,21 @@ class MLFlow(SientiaMonitoring):
|
||||
)
|
||||
|
||||
# Keep heavy model fitting outside MLflow run timing.
|
||||
wrapper.retrain(data)
|
||||
prediction_data = wrapper.retrain(data)
|
||||
prediction_data.rename(columns={target: 'prediction'}, inplace=True)
|
||||
|
||||
# Merge prediction data with retrain data
|
||||
evaluation_data = pd.merge(
|
||||
data, prediction_data, left_index=True, right_index=True, how='left'
|
||||
)
|
||||
|
||||
# Rename target column to "target"
|
||||
evaluation_data.rename(columns={target: 'target'}, inplace=True)
|
||||
|
||||
# Reset index and put as column "timestamp"
|
||||
evaluation_data['timestamp'] = evaluation_data.index
|
||||
evaluation_data.reset_index(drop=True, inplace=True)
|
||||
evaluation_data.sort_values(by='timestamp', inplace=True, ascending=True)
|
||||
|
||||
run_name = f'{model_name}-retrain-{datetime.now().strftime("%Y%m%d%H%M%S")}'
|
||||
|
||||
@@ -541,8 +556,12 @@ class MLFlow(SientiaMonitoring):
|
||||
tmp_dir = tempfile.mkdtemp(prefix='laborious_retrain_')
|
||||
try:
|
||||
raw_csv = Path(tmp_dir) / 'retrain_input.csv'
|
||||
evaluation_csv = Path(tmp_dir) / 'evaluation_data.csv'
|
||||
data.to_csv(raw_csv, index=False)
|
||||
evaluation_data.to_csv(evaluation_csv, index=False)
|
||||
|
||||
mlflow.log_artifact(str(raw_csv))
|
||||
mlflow.log_artifact(str(evaluation_csv))
|
||||
finally:
|
||||
rmtree(tmp_dir, ignore_errors=True)
|
||||
|
||||
@@ -639,21 +658,59 @@ class MLFlow(SientiaMonitoring):
|
||||
self.error(trace, metadata=metadata)
|
||||
raise e
|
||||
|
||||
def _resolve_reference_artifact_name(self, run_id: str) -> str | None:
|
||||
"""
|
||||
Pick the first available reference CSV artifact path from the MLflow run.
|
||||
|
||||
Candidates are checked in priority order: ``retrain_input.csv``, then ``train_data.csv``.
|
||||
A path matches when it equals the candidate or ends with ``/<candidate>`` for nested layouts.
|
||||
|
||||
Args:
|
||||
- run_id: MLflow run UUID linked to the production model version.
|
||||
|
||||
Return:
|
||||
Artifact path string for ``download_artifacts``, or ``None`` if no candidate exists.
|
||||
"""
|
||||
listed = self.mlflow_repository._client.list_artifacts(run_id)
|
||||
paths = [file_info.path for file_info in listed]
|
||||
for candidate in self._REFERENCE_ARTIFACT_CANDIDATES:
|
||||
for path in paths:
|
||||
if path == candidate or path.endswith(f'/{candidate}'):
|
||||
return path
|
||||
return None
|
||||
|
||||
def _find_downloaded_csv(self, tmpdir: str, artifact_name: str) -> Path | None:
|
||||
"""
|
||||
Locate a downloaded reference CSV in the temp directory.
|
||||
|
||||
Args:
|
||||
- tmpdir: Directory where ``download_artifacts`` wrote files.
|
||||
- artifact_name: Basename of the resolved artifact (e.g. ``retrain_input.csv``).
|
||||
|
||||
Return:
|
||||
``Path`` to the CSV file if found, else ``None``.
|
||||
"""
|
||||
direct = Path(tmpdir) / artifact_name
|
||||
if direct.exists():
|
||||
return direct
|
||||
matches = list(Path(tmpdir).rglob(artifact_name))
|
||||
return matches[0] if matches else None
|
||||
|
||||
@activity.defn(name='get_reference_data')
|
||||
def get_reference_data(self, input_data: dict[str, Any]) -> list[dict] | None:
|
||||
"""
|
||||
Download ``evaluation_data.csv`` from the MLflow run linked to ``production`` and parse it.
|
||||
Download reference training CSV from the MLflow run linked to the production alias.
|
||||
|
||||
Used by drift workflows to compare live data against the reference distribution logged with
|
||||
the model. Artifacts are downloaded to a temp directory, discovered via ``rglob`` (nested
|
||||
layout-safe), then timestamps are normalized to ``DATETIME_FORMAT`` string columns before
|
||||
returning record-oriented dicts.
|
||||
Resolves ``retrain_input.csv`` or ``train_data.csv`` via artifact listing before download.
|
||||
``retrain_input.csv`` is preferred when both exist (most recent retrain snapshot). Used by
|
||||
drift workflows to compare live data against the reference distribution logged with the model.
|
||||
Timestamps are normalized to ``DATETIME_FORMAT`` string columns before returning records.
|
||||
|
||||
Args:
|
||||
- input_data: ``metadata`` and ``model_name`` for registry lookup.
|
||||
- input_data: ``metadata``, ``model_name``, and optional ``model_config`` with ``alias``.
|
||||
|
||||
Return:
|
||||
List of row dicts, or ``None`` if the artifact path is missing or any step fails.
|
||||
List of row dicts with normalized timestamps, or ``None`` if resolution or load fails.
|
||||
"""
|
||||
metadata = input_data['metadata']
|
||||
model_name = input_data['model_name']
|
||||
@@ -666,20 +723,25 @@ class MLFlow(SientiaMonitoring):
|
||||
)
|
||||
run_id = mv.run_id
|
||||
|
||||
artifact_path = self._resolve_reference_artifact_name(run_id)
|
||||
if artifact_path is None:
|
||||
self.warning(f'Reference data not found for model {model_name}', metadata)
|
||||
return None
|
||||
|
||||
artifact_name = Path(artifact_path).name
|
||||
tmpdir = tempfile.mkdtemp(prefix='laborious_eval_')
|
||||
try:
|
||||
self.mlflow_repository.download_artifacts(
|
||||
run_id=run_id,
|
||||
artifact_path='evaluation_data.csv',
|
||||
artifact_path=artifact_path,
|
||||
dst_path=tmpdir,
|
||||
metadata=metadata,
|
||||
)
|
||||
csv_candidates = list(Path(tmpdir).rglob('evaluation_data.csv'))
|
||||
if not csv_candidates:
|
||||
csv_path = self._find_downloaded_csv(tmpdir, artifact_name)
|
||||
if csv_path is None:
|
||||
self.warning(f'Reference data not found for model {model_name}', metadata)
|
||||
return None
|
||||
|
||||
reference_data = pd.read_csv(csv_candidates[0])
|
||||
reference_data = pd.read_csv(csv_path)
|
||||
finally:
|
||||
rmtree(tmpdir, ignore_errors=True)
|
||||
|
||||
|
||||
@@ -6,7 +6,6 @@ with workflow.unsafe.imports_passed_through():
|
||||
import warnings
|
||||
from typing import Any
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
from pandas import DataFrame, Index, Series, to_datetime
|
||||
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
|
||||
@@ -16,6 +15,7 @@ with workflow.unsafe.imports_passed_through():
|
||||
from sientia_do.observability.sientia_monitoring import SientiaMonitoring
|
||||
from sientia_do.temporal.constants import DATETIME_FORMAT_WITH_TZ
|
||||
from sientia_model.analytics.drift_analysis import DriftAnalysis, DriftInsufficientDataError
|
||||
from sientia_model.metrics.regression import RegressionMetrics
|
||||
|
||||
from laborious import metrics
|
||||
from laborious.utils.dataframe_debug import build_dataframe_debug_message
|
||||
@@ -365,62 +365,62 @@ class ModelMetrics(SientiaMonitoring):
|
||||
@activity.defn(name='calculate_simple_metrics')
|
||||
def calculate_simple_metrics(self, input_data: dict[str, Any]) -> list[dict]:
|
||||
"""
|
||||
Calculate simple metrics for a model. Metrics available are:
|
||||
- rmse
|
||||
- mse
|
||||
- mae
|
||||
- r2
|
||||
- accuracy
|
||||
- precision
|
||||
- recall
|
||||
- f1
|
||||
Calculate simple regression metrics for a model using RegressionMetrics.
|
||||
|
||||
Args:
|
||||
input_data (dict[str, Any]): Input data containing:
|
||||
input_data: Input data containing:
|
||||
- metadata (dict): Workflow execution metadata
|
||||
- model_id (str): ID of the MLFlow model
|
||||
- target_data (pd.DataFrame): Target data for calculating metrics, containing target and prediction columns
|
||||
- metrics (list[str]): List of metrics to calculate
|
||||
- target_data (list[dict]): Target data with target, prediction, timestamp columns
|
||||
- metrics (list[str]): Metric names to calculate
|
||||
- interval_minutes (int): Window interval in minutes
|
||||
- model_type (str | None): Model algorithm type (for r2 lock)
|
||||
Returns:
|
||||
dict[Hashable, Any]: Dictionary containing the calculated metrics
|
||||
list[dict]: Records with metric, value, model_id, timestamp, data_size, interval_minutes
|
||||
"""
|
||||
|
||||
metadata = input_data['metadata']
|
||||
model_id = input_data['model_id']
|
||||
target_data = DataFrame(input_data['target_data'])
|
||||
metric_names = input_data['metrics']
|
||||
metric_names = list(input_data['metrics'])
|
||||
interval_minutes = input_data['interval_minutes']
|
||||
model_type = input_data.get('model_type')
|
||||
|
||||
data_size = target_data.shape[0]
|
||||
|
||||
output_data = []
|
||||
# Filter r2 when model_type is known and unsupported
|
||||
if (
|
||||
model_type
|
||||
and 'r2' in metric_names
|
||||
and not RegressionMetrics.is_r2_supported(model_type)
|
||||
):
|
||||
metric_names = [m for m in metric_names if m != 'r2']
|
||||
self.warning(
|
||||
f'r2 excluded for model {model_id}: not supported for model_type={model_type}',
|
||||
metadata,
|
||||
)
|
||||
self.send_notification(
|
||||
metadata=metadata,
|
||||
notification_id='SIMPLE_METRICS_R2_UNSUPPORTED',
|
||||
message=f'r2 excluded: not a valid metric for model_type={model_type}',
|
||||
block='model_metrics',
|
||||
level=NotificationLevel.WARNING,
|
||||
)
|
||||
|
||||
diff = target_data['target'] - target_data['prediction']
|
||||
diff_squared = diff**2
|
||||
if not metric_names:
|
||||
self.info(f'No metrics to calculate for model {model_id} after filtering', metadata)
|
||||
return []
|
||||
|
||||
self.info(f'Calculating simple metrics for model {model_id}: {metric_names}', metadata)
|
||||
|
||||
for metric in metric_names:
|
||||
if metric == 'rmse':
|
||||
output_data.append({'metric': 'rmse', 'value': np.sqrt(np.mean(diff_squared))})
|
||||
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)
|
||||
# Build Series with DatetimeIndex for RegressionMetrics
|
||||
timestamps = pd.to_datetime(target_data['timestamp'])
|
||||
real_data = Series(target_data['target'].values, index=timestamps, dtype=float)
|
||||
predictions = Series(target_data['prediction'].values, index=timestamps, dtype=float)
|
||||
|
||||
ss_res = np.sum(diff_squared)
|
||||
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})
|
||||
regression = RegressionMetrics(real_data, predictions)
|
||||
output_data = regression.calculate(metric_names)
|
||||
|
||||
# Wrap with metadata columns matching the existing output schema
|
||||
data = DataFrame(output_data)
|
||||
data['model_id'] = model_id
|
||||
data['timestamp'] = target_data['timestamp'].max()
|
||||
@@ -429,4 +429,34 @@ class ModelMetrics(SientiaMonitoring):
|
||||
|
||||
self._debug_dataframe(f'Simple metrics dataframe: Size {data.shape}', data, metadata)
|
||||
|
||||
# Threshold alerting (optional — no crash when absent)
|
||||
thresholds = input_data.get('thresholds')
|
||||
if thresholds:
|
||||
# Convention: _max thresholds breach when value > threshold,
|
||||
# _min thresholds breach when value < threshold.
|
||||
for row in output_data:
|
||||
metric_name = row['metric']
|
||||
value = row['value']
|
||||
max_key = f'{metric_name}_max'
|
||||
min_key = f'{metric_name}_min'
|
||||
|
||||
breach_msg = None
|
||||
if max_key in thresholds and value > thresholds[max_key]:
|
||||
breach_msg = f'{metric_name}={value} exceeds {max_key}={thresholds[max_key]}'
|
||||
elif min_key in thresholds and value < thresholds[min_key]:
|
||||
breach_msg = f'{metric_name}={value} below {min_key}={thresholds[min_key]}'
|
||||
|
||||
if breach_msg:
|
||||
self.warning(
|
||||
f'Threshold breach for model {model_id}: {breach_msg}',
|
||||
metadata,
|
||||
)
|
||||
self.send_notification(
|
||||
metadata=metadata,
|
||||
notification_id='SIMPLE_METRICS_THRESHOLD_BREACH',
|
||||
message=f'Threshold breach for model {model_id}: {breach_msg}',
|
||||
block='model_metrics',
|
||||
level=NotificationLevel.WARNING,
|
||||
)
|
||||
|
||||
return data.to_dict(orient='records')
|
||||
|
||||
@@ -143,9 +143,7 @@ class OPC(SientiaMonitoring):
|
||||
attachment_content=error_data.get('attachment_content', None),
|
||||
)
|
||||
else:
|
||||
self.info(
|
||||
f'OPC server {opc_id}:{server["server_name"]} connected successfully.'
|
||||
)
|
||||
self.info(f'OPC server {opc_id}:{server["server_name"]} connected successfully.')
|
||||
|
||||
def write_data(
|
||||
self,
|
||||
|
||||
@@ -11,8 +11,7 @@ def build_mlflow_config() -> dict[str, Any]:
|
||||
same string workers and notebooks should use for ``MLFLOW_TRACKING_URI``-style clients.
|
||||
|
||||
Environment Variables:
|
||||
MLFLOW_HOST: Host with scheme; port optional if MLFLOW_PORT is set (default: http://localhost)
|
||||
MLFLOW_PORT: Appended when host has no explicit port (default: 5080)
|
||||
MLFLOW_URL: Host with scheme
|
||||
MLFLOW_USERNAME: Basic-auth or service user (default: aignosi)
|
||||
MLFLOW_PASSWORD: Password or token (default: aignosi)
|
||||
|
||||
|
||||
@@ -755,9 +755,7 @@ class OpcRepository(SientiaMonitoring):
|
||||
pass
|
||||
node_obj.set_value(data, variant_type)
|
||||
|
||||
def _write_reconnect_in_progress(
|
||||
self, metadata: dict[str, Any]
|
||||
) -> tuple[bool, dict[str, Any]]:
|
||||
def _write_reconnect_in_progress(self, metadata: dict[str, Any]) -> tuple[bool, dict[str, Any]]:
|
||||
"""
|
||||
Fail a write because a background reconnect thread is already running.
|
||||
|
||||
|
||||
@@ -147,9 +147,13 @@ async def main():
|
||||
to_install_runtime = runtime
|
||||
|
||||
try:
|
||||
await plugin_store.install_runtime(runtime_name=to_install_runtime, metadata=metadata_runtime)
|
||||
await plugin_store.install_runtime(
|
||||
runtime_name=to_install_runtime, metadata=metadata_runtime
|
||||
)
|
||||
except Exception as exc:
|
||||
logger.custom_critical(f'Failed to install runtime {to_install_runtime}: {exc}', metadata_runtime)
|
||||
logger.custom_critical(
|
||||
f'Failed to install runtime {to_install_runtime}: {exc}', metadata_runtime
|
||||
)
|
||||
metrics.APP_UP.labels(pod_id=POD_ID).set(0)
|
||||
sys.exit(1)
|
||||
|
||||
@@ -212,6 +216,7 @@ async def main():
|
||||
activities.export_data_to_postgres,
|
||||
],
|
||||
logger=logger,
|
||||
runtime='core',
|
||||
),
|
||||
prepare_worker(
|
||||
temporal_client=temporal_client,
|
||||
@@ -224,6 +229,7 @@ async def main():
|
||||
activities.export_data_to_postgres,
|
||||
],
|
||||
logger=logger,
|
||||
runtime='core',
|
||||
),
|
||||
prepare_worker(
|
||||
temporal_client=temporal_client,
|
||||
|
||||
@@ -31,6 +31,8 @@ class SimpleMetrics:
|
||||
|
||||
model_config = input_data['model_config']
|
||||
target_name = model_config['target']
|
||||
model_type = model_config.get('model_type')
|
||||
thresholds = model_config.get('simple_metrics_thresholds')
|
||||
|
||||
query = f"""
|
||||
select p."timestamp", p.prediction, ld.value as "target"
|
||||
@@ -70,6 +72,8 @@ class SimpleMetrics:
|
||||
'target_data': target_data,
|
||||
'metrics': input_data.get('metrics', ['rmse', 'mse', 'mae', 'r2']),
|
||||
'interval_minutes': interval_minutes,
|
||||
'model_type': model_type,
|
||||
'thresholds': thresholds,
|
||||
},
|
||||
retry_policy=retry_policy,
|
||||
start_to_close_timeout=timedelta(seconds=300),
|
||||
|
||||
@@ -4,7 +4,7 @@ sqlalchemy
|
||||
asyncua==1.0.6
|
||||
redis
|
||||
git+ssh://git@github.com/Aignosi/sientia-dataops-library.git@1.12.1
|
||||
git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.10.0
|
||||
git+ssh://git@github.com/Aignosi/sientia-model-library.git@0.13.1
|
||||
prometheus-client
|
||||
botocore
|
||||
boto3
|
||||
|
||||
@@ -4,7 +4,7 @@ sqlalchemy
|
||||
asyncua==1.0.6
|
||||
redis
|
||||
sientia_do>=1.12.1
|
||||
sientia_model>=0.8.2
|
||||
sientia_model>=0.12.0
|
||||
prometheus-client
|
||||
botocore
|
||||
boto3
|
||||
|
||||
@@ -399,6 +399,8 @@ def test_retrain_model_success_data_success_retrain(
|
||||
'value': [1.0, 2.0],
|
||||
}
|
||||
)
|
||||
pivoted_index = pd.DatetimeIndex([ts])
|
||||
wrapper.retrain.return_value = _retrain_prediction_frame(pivoted_index, 'target', [1.0])
|
||||
|
||||
payload = MagicMock()
|
||||
payload.retrieve = MagicMock(return_value=raw_data)
|
||||
@@ -428,6 +430,7 @@ def test_retrain_model_success_with_payload_data(
|
||||
mock_to_datetime, mock_rmtree, mock_mkdtemp, mock_log_artifact, mlflow
|
||||
):
|
||||
mock_mkdtemp.return_value = 'tmp'
|
||||
mock_to_datetime.side_effect = lambda idx, **kwargs: pd.DatetimeIndex(idx)
|
||||
mv_alias = MagicMock(run_id='src')
|
||||
mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv_alias
|
||||
wrapper = MagicMock()
|
||||
@@ -437,20 +440,21 @@ def test_retrain_model_success_with_payload_data(
|
||||
mock_cm.__exit__.return_value = False
|
||||
mlflow.mlflow_repository.start_run.return_value = mock_cm
|
||||
|
||||
raw_data = MagicMock(columns=['variable', 'timestamp', 'value', 'created_at'])
|
||||
raw_data.__getitem__.return_value.max.return_value = 'ts'
|
||||
ts = pd.Timestamp('2020-01-01', tz='UTC')
|
||||
raw_data = pd.DataFrame(
|
||||
{
|
||||
'variable': ['target', 'f1', 'target', 'f1'],
|
||||
'timestamp': [ts, ts, ts + pd.Timedelta(hours=1), ts + pd.Timedelta(hours=1)],
|
||||
'value': [1.0, 2.0, 3.0, 4.0],
|
||||
'created_at': [ts, ts, ts + pd.Timedelta(hours=1), ts + pd.Timedelta(hours=1)],
|
||||
}
|
||||
)
|
||||
pivoted_index = pd.DatetimeIndex([ts, ts + pd.Timedelta(hours=1)])
|
||||
wrapper.retrain.return_value = _retrain_prediction_frame(pivoted_index, 'target', [1.0, 3.0])
|
||||
|
||||
payload = MagicMock()
|
||||
payload.retrieve = MagicMock(return_value=raw_data)
|
||||
|
||||
pivoted = MagicMock()
|
||||
raw_data.sort_values.return_value = raw_data
|
||||
raw_data.drop_duplicates.return_value = raw_data
|
||||
raw_data.pivot.return_value = pivoted
|
||||
pivoted.fillna = MagicMock()
|
||||
pivoted.columns.name = None
|
||||
pivoted.index = MagicMock()
|
||||
pivoted.__setitem__ = MagicMock()
|
||||
|
||||
response = mlflow.retrain_model(
|
||||
{
|
||||
**metadata,
|
||||
@@ -482,13 +486,17 @@ def test_retrain_model_always_uses_retrain_even_with_full_retrain_flag(
|
||||
mlflow.mlflow_repository.start_run.return_value = mock_cm
|
||||
|
||||
ts = pd.Timestamp('2020-01-01', tz='UTC')
|
||||
ts_next = ts + pd.Timedelta(days=1)
|
||||
raw_data = pd.DataFrame(
|
||||
{
|
||||
'variable': ['target', 'f1', 'target', 'f1'],
|
||||
'timestamp': [ts, ts, ts + pd.Timedelta(days=1), ts + pd.Timedelta(days=1)],
|
||||
'timestamp': [ts, ts, ts_next, ts_next],
|
||||
'value': [1.0, 2.0, 3.0, 4.0],
|
||||
}
|
||||
)
|
||||
pivoted_index = pd.DatetimeIndex([ts, ts_next])
|
||||
wrapper.retrain.return_value = _retrain_prediction_frame(pivoted_index, 'target', [1.0, 3.0])
|
||||
|
||||
payload = MagicMock()
|
||||
payload.retrieve = MagicMock(return_value=raw_data)
|
||||
|
||||
@@ -652,6 +660,18 @@ def test_update_production_model_error(mlflow):
|
||||
raise AssertionError('Expected exception')
|
||||
|
||||
|
||||
def _artifact_file_info(path: str) -> MagicMock:
|
||||
file_info = MagicMock()
|
||||
file_info.path = path
|
||||
return file_info
|
||||
|
||||
|
||||
def _retrain_prediction_frame(
|
||||
index: pd.DatetimeIndex, target: str, values: list[float]
|
||||
) -> pd.DataFrame:
|
||||
return pd.DataFrame({target: values}, index=index)
|
||||
|
||||
|
||||
@patch('laborious.activities.mlflow.to_datetime')
|
||||
def test_get_reference_data_success(mock_to_datetime, mlflow):
|
||||
input_data = {
|
||||
@@ -661,6 +681,9 @@ def test_get_reference_data_success(mock_to_datetime, mlflow):
|
||||
|
||||
mv = MagicMock(run_id='run1')
|
||||
mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv
|
||||
mlflow.mlflow_repository._client.list_artifacts.return_value = [
|
||||
_artifact_file_info('evaluation_data.csv'),
|
||||
]
|
||||
|
||||
mock_reference_data = MagicMock()
|
||||
mock_reference_data.__getitem__.return_value = MagicMock()
|
||||
@@ -672,10 +695,19 @@ def test_get_reference_data_success(mock_to_datetime, mlflow):
|
||||
with patch('laborious.activities.mlflow.pd.read_csv', return_value=mock_reference_data):
|
||||
with patch('laborious.activities.mlflow.tempfile.mkdtemp', return_value='/t'):
|
||||
with patch('laborious.activities.mlflow.rmtree'):
|
||||
with patch('laborious.activities.mlflow.Path') as mp:
|
||||
mp.return_value.rglob.return_value = [MagicMock()]
|
||||
with patch.object(
|
||||
mlflow,
|
||||
'_find_downloaded_csv',
|
||||
return_value=MagicMock(),
|
||||
):
|
||||
result = mlflow.get_reference_data(input_data)
|
||||
|
||||
mlflow.mlflow_repository.download_artifacts.assert_called_once_with(
|
||||
run_id='run1',
|
||||
artifact_path='evaluation_data.csv',
|
||||
dst_path='/t',
|
||||
metadata=metadata['metadata'],
|
||||
)
|
||||
mock_reference_data.to_dict.assert_called_once_with(orient='records')
|
||||
assert result == mock_reference_data.to_dict.return_value
|
||||
|
||||
@@ -694,6 +726,97 @@ def test_get_reference_data_not_found(mlflow):
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_get_reference_data_only_test_data_csv(mlflow):
|
||||
input_data = {
|
||||
**metadata,
|
||||
'model_name': 'test_model',
|
||||
}
|
||||
mv = MagicMock(run_id='run1')
|
||||
mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv
|
||||
mlflow.mlflow_repository._client.list_artifacts.return_value = [
|
||||
_artifact_file_info('test_data.csv'),
|
||||
]
|
||||
|
||||
mock_reference_data = MagicMock()
|
||||
mock_reference_data.__getitem__.return_value = MagicMock()
|
||||
mock_reference_data.to_dict.return_value = [{'timestamp': '2023-05-26 11:12:27', 'value': 1.0}]
|
||||
|
||||
with patch('laborious.activities.mlflow.to_datetime') as mock_to_datetime:
|
||||
mock_to_datetime.return_value.dt.strftime.return_value = MagicMock()
|
||||
with patch('laborious.activities.mlflow.pd.read_csv', return_value=mock_reference_data):
|
||||
with patch('laborious.activities.mlflow.tempfile.mkdtemp', return_value='/t'):
|
||||
with patch('laborious.activities.mlflow.rmtree'):
|
||||
with patch.object(
|
||||
mlflow,
|
||||
'_find_downloaded_csv',
|
||||
return_value=MagicMock(),
|
||||
):
|
||||
result = mlflow.get_reference_data(input_data)
|
||||
|
||||
mlflow.mlflow_repository.download_artifacts.assert_called_once_with(
|
||||
run_id='run1',
|
||||
artifact_path='test_data.csv',
|
||||
dst_path='/t',
|
||||
metadata=metadata['metadata'],
|
||||
)
|
||||
assert result == mock_reference_data.to_dict.return_value
|
||||
|
||||
|
||||
def test_get_reference_data_prefers_evaluation_data_when_both_listed(mlflow):
|
||||
input_data = {
|
||||
**metadata,
|
||||
'model_name': 'test_model',
|
||||
}
|
||||
mv = MagicMock(run_id='run1')
|
||||
mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv
|
||||
mlflow.mlflow_repository._client.list_artifacts.return_value = [
|
||||
_artifact_file_info('test_data.csv'),
|
||||
_artifact_file_info('evaluation_data.csv'),
|
||||
]
|
||||
|
||||
mock_reference_data = MagicMock()
|
||||
mock_reference_data.__getitem__.return_value = MagicMock()
|
||||
mock_reference_data.to_dict.return_value = [{'timestamp': '2023-05-26 11:12:27', 'value': 1.0}]
|
||||
|
||||
with patch('laborious.activities.mlflow.to_datetime') as mock_to_datetime:
|
||||
mock_to_datetime.return_value.dt.strftime.return_value = MagicMock()
|
||||
with patch('laborious.activities.mlflow.pd.read_csv', return_value=mock_reference_data):
|
||||
with patch('laborious.activities.mlflow.tempfile.mkdtemp', return_value='/t'):
|
||||
with patch('laborious.activities.mlflow.rmtree'):
|
||||
with patch.object(
|
||||
mlflow,
|
||||
'_find_downloaded_csv',
|
||||
return_value=MagicMock(),
|
||||
):
|
||||
result = mlflow.get_reference_data(input_data)
|
||||
|
||||
mlflow.mlflow_repository.download_artifacts.assert_called_once_with(
|
||||
run_id='run1',
|
||||
artifact_path='evaluation_data.csv',
|
||||
dst_path='/t',
|
||||
metadata=metadata['metadata'],
|
||||
)
|
||||
assert result == mock_reference_data.to_dict.return_value
|
||||
|
||||
|
||||
def test_get_reference_data_no_candidate_artifacts_returns_none(mlflow):
|
||||
input_data = {
|
||||
**metadata,
|
||||
'model_name': 'test_model',
|
||||
}
|
||||
mv = MagicMock(run_id='run1')
|
||||
mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv
|
||||
mlflow.mlflow_repository._client.list_artifacts.return_value = [
|
||||
_artifact_file_info('other_artifact.csv'),
|
||||
]
|
||||
|
||||
result = mlflow.get_reference_data(input_data)
|
||||
|
||||
mlflow.mlflow_repository.download_artifacts.assert_not_called()
|
||||
mlflow.warning.assert_called()
|
||||
assert result is None
|
||||
|
||||
|
||||
def test_get_reference_data_missing_csv_file_returns_none(mlflow):
|
||||
input_data = {
|
||||
**metadata,
|
||||
@@ -701,13 +824,17 @@ def test_get_reference_data_missing_csv_file_returns_none(mlflow):
|
||||
}
|
||||
mv = MagicMock(run_id='run1')
|
||||
mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv
|
||||
mlflow.mlflow_repository._client.list_artifacts.return_value = [
|
||||
_artifact_file_info('evaluation_data.csv'),
|
||||
]
|
||||
|
||||
with patch('laborious.activities.mlflow.tempfile.mkdtemp', return_value='tmp'):
|
||||
with patch('laborious.activities.mlflow.rmtree'):
|
||||
with patch('laborious.activities.mlflow.Path') as mp:
|
||||
mp.return_value.rglob.return_value = []
|
||||
with patch.object(mlflow, '_find_downloaded_csv', return_value=None):
|
||||
result = mlflow.get_reference_data(input_data)
|
||||
|
||||
mlflow.mlflow_repository.download_artifacts.assert_called_once()
|
||||
mlflow.warning.assert_called()
|
||||
assert result is None
|
||||
|
||||
|
||||
@@ -719,6 +846,9 @@ def test_get_reference_data_exception(mlflow):
|
||||
|
||||
mv = MagicMock(run_id='run1')
|
||||
mlflow.mlflow_repository._client.get_model_version_by_alias.return_value = mv
|
||||
mlflow.mlflow_repository._client.list_artifacts.return_value = [
|
||||
_artifact_file_info('evaluation_data.csv'),
|
||||
]
|
||||
mlflow.mlflow_repository.download_artifacts.side_effect = Exception('dl fail')
|
||||
|
||||
result = mlflow.get_reference_data(input_data)
|
||||
|
||||
@@ -45,6 +45,23 @@ metadata = {
|
||||
}
|
||||
|
||||
|
||||
def _mock_regression_metrics_class(calculate_return):
|
||||
"""Returns a patch context manager that mocks RegressionMetrics."""
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.calculate.return_value = calculate_return
|
||||
mock_class = MagicMock(return_value=mock_instance)
|
||||
mock_class.is_r2_supported = MagicMock(return_value=True)
|
||||
mock_class.supported_metrics = MagicMock(return_value=['rmse', 'mse', 'mae', 'r2'])
|
||||
return (
|
||||
patch(
|
||||
'laborious.activities.model_metrics.RegressionMetrics',
|
||||
mock_class,
|
||||
),
|
||||
mock_class,
|
||||
mock_instance,
|
||||
)
|
||||
|
||||
|
||||
def test_calculate_drift_invalid_chunk_period(model_metrics_activity):
|
||||
# Arrange
|
||||
input_data = {
|
||||
@@ -693,7 +710,6 @@ def test_get_drift_metrics_dataframe_error(
|
||||
|
||||
|
||||
def test_calculate_simple_metrics_success_all_metrics(model_metrics_activity):
|
||||
# Arrange
|
||||
target_data = DataFrame(
|
||||
{
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'],
|
||||
@@ -710,28 +726,28 @@ def test_calculate_simple_metrics_success_all_metrics(model_metrics_activity):
|
||||
'interval_minutes': 5,
|
||||
}
|
||||
|
||||
# Act
|
||||
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
|
||||
|
||||
# Assert
|
||||
assert len(result['metric']) == 4
|
||||
assert 'rmse' in result['metric'].values
|
||||
assert 'mse' in result['metric'].values
|
||||
assert 'mae' in result['metric'].values
|
||||
assert 'r2' in result['metric'].values
|
||||
assert all(model_id == 'test_model_id' for model_id in result['model_id'].values)
|
||||
assert all(timestamp == '2023-05-26 11:12:29' for timestamp in result['timestamp'].values)
|
||||
assert all(data_size == 3 for data_size in result['data_size'].values)
|
||||
assert all(interval_minutes == 5 for interval_minutes in result['interval_minutes'].values)
|
||||
model_metrics_activity.info.assert_called_once_with(
|
||||
"Calculating simple metrics for model test_model_id: ['rmse', 'mse', 'mae', 'r2']",
|
||||
metadata['metadata'],
|
||||
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},
|
||||
]
|
||||
)
|
||||
model_metrics_activity.debug.assert_called_once()
|
||||
|
||||
with patcher:
|
||||
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
|
||||
|
||||
assert len(result) == 4
|
||||
assert set(result['metric'].values) == {'rmse', 'mse', 'mae', 'r2'}
|
||||
assert all(mid == 'test_model_id' for mid in result['model_id'].values)
|
||||
assert all(ts == '2023-05-26 11:12:29' for ts in result['timestamp'].values)
|
||||
assert all(ds == 3 for ds in result['data_size'].values)
|
||||
assert all(im == 5 for im in result['interval_minutes'].values)
|
||||
mock_instance.calculate.assert_called_once_with(['rmse', 'mse', 'mae', 'r2'])
|
||||
|
||||
|
||||
def test_calculate_simple_metrics_success_rmse_only(model_metrics_activity):
|
||||
# Arrange
|
||||
target_data = DataFrame(
|
||||
{
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
|
||||
@@ -748,23 +764,23 @@ def test_calculate_simple_metrics_success_rmse_only(model_metrics_activity):
|
||||
'interval_minutes': 5,
|
||||
}
|
||||
|
||||
# Act
|
||||
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
|
||||
patcher, mock_class, mock_instance = _mock_regression_metrics_class(
|
||||
[{'metric': 'rmse', 'value': 0.1}]
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert len(result['metric']) == 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
|
||||
model_metrics_activity.info.assert_called_once_with(
|
||||
"Calculating simple metrics for model test_model_id: ['rmse']", metadata['metadata']
|
||||
)
|
||||
mock_instance.calculate.assert_called_once_with(['rmse'])
|
||||
|
||||
|
||||
def test_calculate_simple_metrics_success_mse_only(model_metrics_activity):
|
||||
# Arrange
|
||||
target_data = DataFrame(
|
||||
{
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
|
||||
@@ -781,23 +797,23 @@ def test_calculate_simple_metrics_success_mse_only(model_metrics_activity):
|
||||
'interval_minutes': 5,
|
||||
}
|
||||
|
||||
# Act
|
||||
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
|
||||
patcher, mock_class, mock_instance = _mock_regression_metrics_class(
|
||||
[{'metric': 'mse', 'value': 0.01}]
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert len(result['metric']) == 1
|
||||
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
|
||||
model_metrics_activity.info.assert_called_once_with(
|
||||
"Calculating simple metrics for model test_model_id: ['mse']", metadata['metadata']
|
||||
)
|
||||
mock_instance.calculate.assert_called_once_with(['mse'])
|
||||
|
||||
|
||||
def test_calculate_simple_metrics_success_mae_only(model_metrics_activity):
|
||||
# Arrange
|
||||
target_data = DataFrame(
|
||||
{
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
|
||||
@@ -814,23 +830,23 @@ def test_calculate_simple_metrics_success_mae_only(model_metrics_activity):
|
||||
'interval_minutes': 5,
|
||||
}
|
||||
|
||||
# Act
|
||||
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
|
||||
patcher, mock_class, mock_instance = _mock_regression_metrics_class(
|
||||
[{'metric': 'mae', 'value': 0.1}]
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert len(result['metric']) == 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
|
||||
model_metrics_activity.info.assert_called_once_with(
|
||||
"Calculating simple metrics for model test_model_id: ['mae']", metadata['metadata']
|
||||
)
|
||||
mock_instance.calculate.assert_called_once_with(['mae'])
|
||||
|
||||
|
||||
def test_calculate_simple_metrics_success_r2_only(model_metrics_activity):
|
||||
# Arrange
|
||||
target_data = DataFrame(
|
||||
{
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
|
||||
@@ -847,24 +863,24 @@ def test_calculate_simple_metrics_success_r2_only(model_metrics_activity):
|
||||
'interval_minutes': 5,
|
||||
}
|
||||
|
||||
# Act
|
||||
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
|
||||
patcher, mock_class, mock_instance = _mock_regression_metrics_class(
|
||||
[{'metric': 'r2', 'value': 0.95}]
|
||||
)
|
||||
|
||||
# Assert
|
||||
assert len(result['metric']) == 1
|
||||
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
|
||||
model_metrics_activity.info.assert_called_once_with(
|
||||
"Calculating simple metrics for model test_model_id: ['r2']", metadata['metadata']
|
||||
)
|
||||
mock_instance.calculate.assert_called_once_with(['r2'])
|
||||
|
||||
|
||||
def test_calculate_simple_metrics_r2_zero_ss_tot(model_metrics_activity):
|
||||
# Arrange
|
||||
# All target values are the same, so ss_tot will be 0
|
||||
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'],
|
||||
@@ -881,24 +897,18 @@ def test_calculate_simple_metrics_r2_zero_ss_tot(model_metrics_activity):
|
||||
'interval_minutes': 5,
|
||||
}
|
||||
|
||||
# Act
|
||||
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
|
||||
|
||||
# Assert
|
||||
assert len(result['metric']) == 1
|
||||
assert result['metric'].values[0] == 'r2'
|
||||
assert result['value'].values[0] == 0.0 # Should return 0.0 when ss_tot == 0
|
||||
assert result['model_id'].values[0] == 'test_model_id'
|
||||
assert result['timestamp'].values[0] == '2023-05-26 11:12:28'
|
||||
assert result['data_size'].values[0] == 2
|
||||
assert result['interval_minutes'].values[0] == 5
|
||||
model_metrics_activity.info.assert_called_once_with(
|
||||
"Calculating simple metrics for model test_model_id: ['r2']", metadata['metadata']
|
||||
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):
|
||||
# Arrange
|
||||
target_data = DataFrame(
|
||||
{
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28', '2023-05-26 11:12:29'],
|
||||
@@ -915,23 +925,26 @@ def test_calculate_simple_metrics_success_multiple_metrics_subset(model_metrics_
|
||||
'interval_minutes': 5,
|
||||
}
|
||||
|
||||
# Act
|
||||
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
|
||||
|
||||
# Assert
|
||||
assert len(result['metric']) == 2
|
||||
assert 'rmse' in result['metric'].values
|
||||
assert 'mae' in result['metric'].values
|
||||
assert all(model_id == 'test_model_id' for model_id in result['model_id'].values)
|
||||
assert all(timestamp == '2023-05-26 11:12:29' for timestamp in result['timestamp'].values)
|
||||
assert all(data_size == 3 for data_size in result['data_size'].values)
|
||||
assert all(interval_minutes == 5 for interval_minutes in result['interval_minutes'].values)
|
||||
model_metrics_activity.info.assert_called_once_with(
|
||||
"Calculating simple metrics for model test_model_id: ['rmse', 'mae']", metadata['metadata']
|
||||
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))
|
||||
|
||||
def test_calculate_simple_metrics_unknown_metric_ignored(model_metrics_activity):
|
||||
assert len(result) == 2
|
||||
assert set(result['metric'].values) == {'rmse', 'mae'}
|
||||
assert all(mid == 'test_model_id' for mid in result['model_id'].values)
|
||||
assert all(ts == '2023-05-26 11:12:29' for ts in result['timestamp'].values)
|
||||
assert all(ds == 3 for ds in result['data_size'].values)
|
||||
assert all(im == 5 for im in result['interval_minutes'].values)
|
||||
mock_instance.calculate.assert_called_once_with(['rmse', 'mae'])
|
||||
|
||||
|
||||
def test_calculate_simple_metrics_unknown_metric_raises(model_metrics_activity):
|
||||
target_data = DataFrame(
|
||||
{
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
|
||||
@@ -948,7 +961,211 @@ def test_calculate_simple_metrics_unknown_metric_ignored(model_metrics_activity)
|
||||
'interval_minutes': 5,
|
||||
}
|
||||
|
||||
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.calculate.side_effect = ValueError('Unknown metric: unknown_metric')
|
||||
mock_class = MagicMock(return_value=mock_instance)
|
||||
mock_class.is_r2_supported = MagicMock(return_value=True)
|
||||
|
||||
assert len(result['metric']) == 1
|
||||
with patch('laborious.activities.model_metrics.RegressionMetrics', mock_class):
|
||||
with raises(ValueError, match='Unknown metric'):
|
||||
model_metrics_activity.calculate_simple_metrics(input_data)
|
||||
|
||||
|
||||
def test_calculate_simple_metrics_r2_skipped_for_nonlinear_model(model_metrics_activity):
|
||||
"""When model_type is non-linear, r2 is excluded and a warning notification fires."""
|
||||
target_data = DataFrame(
|
||||
{
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
|
||||
'target': [1.0, 2.0],
|
||||
'prediction': [1.1, 2.1],
|
||||
}
|
||||
)
|
||||
|
||||
input_data = {
|
||||
**metadata,
|
||||
'model_id': 'test_model_id',
|
||||
'target_data': target_data.to_dict(),
|
||||
'metrics': ['rmse', 'r2'],
|
||||
'interval_minutes': 5,
|
||||
'model_type': 'XGBoost',
|
||||
}
|
||||
|
||||
mock_instance = MagicMock()
|
||||
mock_instance.calculate.return_value = [{'metric': 'rmse', 'value': 0.1}]
|
||||
mock_class = MagicMock(return_value=mock_instance)
|
||||
mock_class.is_r2_supported = MagicMock(return_value=False)
|
||||
|
||||
with patch('laborious.activities.model_metrics.RegressionMetrics', mock_class):
|
||||
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
|
||||
|
||||
assert len(result) == 1
|
||||
assert result['metric'].values[0] == 'rmse'
|
||||
mock_class.is_r2_supported.assert_called_once_with('XGBoost')
|
||||
mock_instance.calculate.assert_called_once_with(['rmse'])
|
||||
model_metrics_activity.warning.assert_called_once()
|
||||
model_metrics_activity.send_notification.assert_called_once()
|
||||
call_kwargs = model_metrics_activity.send_notification.call_args.kwargs
|
||||
assert call_kwargs['notification_id'] == 'SIMPLE_METRICS_R2_UNSUPPORTED'
|
||||
assert call_kwargs['level'] == NotificationLevel.WARNING
|
||||
|
||||
|
||||
def test_calculate_simple_metrics_no_model_type_includes_r2(model_metrics_activity):
|
||||
"""When model_type is None (legacy input), r2 is included without check."""
|
||||
target_data = DataFrame(
|
||||
{
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
|
||||
'target': [1.0, 2.0],
|
||||
'prediction': [1.1, 2.1],
|
||||
}
|
||||
)
|
||||
|
||||
input_data = {
|
||||
**metadata,
|
||||
'model_id': 'test_model_id',
|
||||
'target_data': target_data.to_dict(),
|
||||
'metrics': ['r2'],
|
||||
'interval_minutes': 5,
|
||||
# no model_type key
|
||||
}
|
||||
|
||||
patcher, mock_class, mock_instance = _mock_regression_metrics_class(
|
||||
[{'metric': 'r2', 'value': 0.95}]
|
||||
)
|
||||
|
||||
with patcher:
|
||||
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
|
||||
|
||||
assert result['metric'].values[0] == 'r2'
|
||||
mock_class.is_r2_supported.assert_not_called()
|
||||
|
||||
|
||||
def test_calculate_simple_metrics_threshold_breach_rmse(model_metrics_activity):
|
||||
"""When rmse exceeds rmse_max, a WARNING notification fires."""
|
||||
target_data = DataFrame(
|
||||
{
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
|
||||
'target': [1.0, 2.0],
|
||||
'prediction': [1.1, 2.1],
|
||||
}
|
||||
)
|
||||
|
||||
input_data = {
|
||||
**metadata,
|
||||
'model_id': 'test_model_id',
|
||||
'target_data': target_data.to_dict(),
|
||||
'metrics': ['rmse'],
|
||||
'interval_minutes': 5,
|
||||
'thresholds': {'rmse_max': 0.05},
|
||||
}
|
||||
|
||||
patcher, mock_class, mock_instance = _mock_regression_metrics_class(
|
||||
[
|
||||
{'metric': 'rmse', 'value': 0.1},
|
||||
]
|
||||
)
|
||||
|
||||
with patcher:
|
||||
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
|
||||
|
||||
assert len(result) == 1
|
||||
model_metrics_activity.send_notification.assert_called_once()
|
||||
call_kwargs = model_metrics_activity.send_notification.call_args.kwargs
|
||||
assert call_kwargs['notification_id'] == 'SIMPLE_METRICS_THRESHOLD_BREACH'
|
||||
assert call_kwargs['level'] == NotificationLevel.WARNING
|
||||
assert 'rmse' in call_kwargs['message']
|
||||
|
||||
|
||||
def test_calculate_simple_metrics_threshold_no_breach(model_metrics_activity):
|
||||
"""When rmse is below rmse_max, no notification fires."""
|
||||
target_data = DataFrame(
|
||||
{
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
|
||||
'target': [1.0, 2.0],
|
||||
'prediction': [1.1, 2.1],
|
||||
}
|
||||
)
|
||||
|
||||
input_data = {
|
||||
**metadata,
|
||||
'model_id': 'test_model_id',
|
||||
'target_data': target_data.to_dict(),
|
||||
'metrics': ['rmse'],
|
||||
'interval_minutes': 5,
|
||||
'thresholds': {'rmse_max': 1.0},
|
||||
}
|
||||
|
||||
patcher, mock_class, mock_instance = _mock_regression_metrics_class(
|
||||
[
|
||||
{'metric': 'rmse', 'value': 0.1},
|
||||
]
|
||||
)
|
||||
|
||||
with patcher:
|
||||
model_metrics_activity.calculate_simple_metrics(input_data)
|
||||
|
||||
model_metrics_activity.send_notification.assert_not_called()
|
||||
|
||||
|
||||
def test_calculate_simple_metrics_threshold_r2_below_min(model_metrics_activity):
|
||||
"""When r2 drops below r2_min, a WARNING notification fires."""
|
||||
target_data = DataFrame(
|
||||
{
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
|
||||
'target': [1.0, 2.0],
|
||||
'prediction': [1.1, 2.1],
|
||||
}
|
||||
)
|
||||
|
||||
input_data = {
|
||||
**metadata,
|
||||
'model_id': 'test_model_id',
|
||||
'target_data': target_data.to_dict(),
|
||||
'metrics': ['r2'],
|
||||
'interval_minutes': 5,
|
||||
'thresholds': {'r2_min': 0.95},
|
||||
}
|
||||
|
||||
patcher, mock_class, mock_instance = _mock_regression_metrics_class(
|
||||
[
|
||||
{'metric': 'r2', 'value': 0.8},
|
||||
]
|
||||
)
|
||||
|
||||
with patcher:
|
||||
model_metrics_activity.calculate_simple_metrics(input_data)
|
||||
|
||||
model_metrics_activity.send_notification.assert_called_once()
|
||||
call_kwargs = model_metrics_activity.send_notification.call_args.kwargs
|
||||
assert 'r2' in call_kwargs['message']
|
||||
|
||||
|
||||
def test_calculate_simple_metrics_no_thresholds_no_alert(model_metrics_activity):
|
||||
"""When thresholds is None (not configured), no alerting, no crash."""
|
||||
target_data = DataFrame(
|
||||
{
|
||||
'timestamp': ['2023-05-26 11:12:27', '2023-05-26 11:12:28'],
|
||||
'target': [1.0, 2.0],
|
||||
'prediction': [1.1, 2.1],
|
||||
}
|
||||
)
|
||||
|
||||
input_data = {
|
||||
**metadata,
|
||||
'model_id': 'test_model_id',
|
||||
'target_data': target_data.to_dict(),
|
||||
'metrics': ['rmse'],
|
||||
'interval_minutes': 5,
|
||||
# no thresholds key
|
||||
}
|
||||
|
||||
patcher, mock_class, mock_instance = _mock_regression_metrics_class(
|
||||
[
|
||||
{'metric': 'rmse', 'value': 999.0},
|
||||
]
|
||||
)
|
||||
|
||||
with patcher:
|
||||
result = DataFrame(model_metrics_activity.calculate_simple_metrics(input_data))
|
||||
|
||||
assert len(result) == 1
|
||||
model_metrics_activity.send_notification.assert_not_called()
|
||||
|
||||
@@ -103,6 +103,8 @@ async def test_run(workflow_mock: AsyncMock, simple_metrics: SimpleMetrics):
|
||||
'target_data': target_data,
|
||||
'metrics': input_data['metrics'],
|
||||
'interval_minutes': input_data['interval_minutes'],
|
||||
'model_type': None,
|
||||
'thresholds': None,
|
||||
},
|
||||
retry_policy=ANY,
|
||||
start_to_close_timeout=ANY,
|
||||
@@ -209,6 +211,8 @@ async def test_run_default_metrics(workflow_mock: AsyncMock, simple_metrics: Sim
|
||||
'target_data': target_data,
|
||||
'metrics': ['rmse', 'mse', 'mae', 'r2'], # Default value
|
||||
'interval_minutes': input_data['interval_minutes'],
|
||||
'model_type': None,
|
||||
'thresholds': None,
|
||||
},
|
||||
retry_policy=ANY,
|
||||
start_to_close_timeout=ANY,
|
||||
|
||||
10
values.yaml
10
values.yaml
@@ -233,11 +233,15 @@ global:
|
||||
runtimes:
|
||||
basic:
|
||||
replicas: 1
|
||||
single:
|
||||
legacy:
|
||||
replicas: 1
|
||||
env:
|
||||
- name: "GITHUB_BRANCH"
|
||||
value: "main"
|
||||
- name: "MLFLOW_HOST"
|
||||
value: "http://sientia-tracker-mlflow-tracking.sientia-tracker.svc.cluster.local"
|
||||
- name: "MLFLOW_PORT"
|
||||
value: "80"
|
||||
|
||||
# -----------------------------------------------------------------------------
|
||||
# Chart-level configuration (applies to all runtimes)
|
||||
@@ -305,8 +309,8 @@ ssh:
|
||||
|
||||
# kubectl create secret docker-registry docker-hub-secret --namespace sientia --docker-server=http://aignosi.azurecr.io --docker-username=aignosi --docker-password=<pwd>
|
||||
#
|
||||
# helm upgrade --install sientia-laborious-worker sientia/sientia-module -n sientia --create-namespace -f ./values.yaml --version 0.6.5
|
||||
#
|
||||
# helm upgrade --install sientia-laborious-worker /home/grezewave/Documents/projects/sientia/sientia-core-applications/sientia-module -n sientia --create-namespace -f ./values.yaml
|
||||
|
||||
# kubectl create secret generic git-ssh-key-sientia-laborious-worker \
|
||||
# --namespace sientia \
|
||||
# --from-file=ssh-privatekey=git_key \
|
||||
|
||||
Reference in New Issue
Block a user