feat(simple_metrics): rewrite calculate_simple_metrics to use RegressionMetrics
Replaces manual numpy if/elif chain with RegressionMetrics from sientia_model. Fixes 3 known bugs in one pass: - NaN now handled via _align_dropna (was silently propagated) - r2 zero-variance uses sklearn r2_score (was divergent, pinned at 0.0) - Unknown metric names raise ValueError (were silently dropped) Also adds r2 lock: when model_type is known and non-linear, r2 is excluded from calculation with a WARNING notification. Drops the now-unused `numpy` import (the manual math it backed is gone, and no other method in this file references it). SIENTIAPDE-1986
This commit is contained in:
@@ -6,7 +6,6 @@ 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
|
||||||
@@ -16,6 +15,7 @@ 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 metrics for a model. Metrics available are:
|
Calculate simple regression metrics for a model using RegressionMetrics.
|
||||||
- rmse
|
|
||||||
- mse
|
|
||||||
- mae
|
|
||||||
- r2
|
|
||||||
- accuracy
|
|
||||||
- precision
|
|
||||||
- recall
|
|
||||||
- f1
|
|
||||||
Args:
|
Args:
|
||||||
input_data (dict[str, Any]): Input data containing:
|
input_data: 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 (pd.DataFrame): Target data for calculating metrics, containing target and prediction columns
|
- target_data (list[dict]): Target data with target, prediction, timestamp columns
|
||||||
- metrics (list[str]): List of metrics to calculate
|
- metrics (list[str]): Metric names to calculate
|
||||||
|
- interval_minutes (int): Window interval in minutes
|
||||||
|
- model_type (str | None): Model algorithm type (for r2 lock)
|
||||||
Returns:
|
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']
|
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 = input_data['metrics']
|
metric_names = list(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]
|
||||||
|
|
||||||
output_data = []
|
# Filter r2 when model_type is known and unsupported
|
||||||
|
if (
|
||||||
|
model_type
|
||||||
|
and 'r2' in metric_names
|
||||||
|
and not RegressionMetrics.is_r2_supported(model_type)
|
||||||
|
):
|
||||||
|
metric_names = [m for m in metric_names if m != 'r2']
|
||||||
|
self.warning(
|
||||||
|
f'r2 excluded for model {model_id}: not supported for model_type={model_type}',
|
||||||
|
metadata,
|
||||||
|
)
|
||||||
|
self.send_notification(
|
||||||
|
metadata=metadata,
|
||||||
|
notification_id='SIMPLE_METRICS_R2_UNSUPPORTED',
|
||||||
|
message=f'r2 excluded: not a valid metric for model_type={model_type}',
|
||||||
|
block='model_metrics',
|
||||||
|
level=NotificationLevel.WARNING,
|
||||||
|
)
|
||||||
|
|
||||||
diff = target_data['target'] - target_data['prediction']
|
if not metric_names:
|
||||||
diff_squared = diff**2
|
self.info(f'No metrics to calculate for model {model_id} after filtering', metadata)
|
||||||
|
return []
|
||||||
|
|
||||||
self.info(f'Calculating simple metrics for model {model_id}: {metric_names}', metadata)
|
self.info(f'Calculating simple metrics for model {model_id}: {metric_names}', metadata)
|
||||||
|
|
||||||
for metric in metric_names:
|
# Build Series with DatetimeIndex for RegressionMetrics
|
||||||
if metric == 'rmse':
|
timestamps = pd.to_datetime(target_data['timestamp'])
|
||||||
output_data.append({'metric': 'rmse', 'value': np.sqrt(np.mean(diff_squared))})
|
real_data = Series(target_data['target'].values, index=timestamps, dtype=float)
|
||||||
elif metric == 'mse':
|
predictions = Series(target_data['prediction'].values, index=timestamps, dtype=float)
|
||||||
output_data.append({'metric': 'mse', 'value': np.mean(diff_squared)})
|
|
||||||
elif metric == 'mae':
|
|
||||||
output_data.append({'metric': 'mae', 'value': np.mean(np.abs(diff))})
|
|
||||||
elif metric == 'r2':
|
|
||||||
y_true = target_data['target']
|
|
||||||
y_mean = np.mean(y_true)
|
|
||||||
|
|
||||||
ss_res = np.sum(diff_squared)
|
regression = RegressionMetrics(real_data, predictions)
|
||||||
ss_tot = np.sum((y_true - y_mean) ** 2)
|
output_data = regression.calculate(metric_names)
|
||||||
|
|
||||||
# Evita divisão por zero
|
|
||||||
if ss_tot == 0:
|
|
||||||
r2_score = 0.0
|
|
||||||
else:
|
|
||||||
r2_score = 1 - (ss_res / ss_tot)
|
|
||||||
|
|
||||||
output_data.append({'metric': 'r2', 'value': r2_score})
|
|
||||||
|
|
||||||
|
# Wrap with metadata columns matching the existing output schema
|
||||||
data = DataFrame(output_data)
|
data = 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()
|
||||||
|
|||||||
Reference in New Issue
Block a user