5997210118
Compares each computed metric against optional per-model thresholds
from model_config.simple_metrics_thresholds. Convention:
- {metric}_max: breach when value > threshold (rmse, mse, mae)
- {metric}_min: breach when value < threshold (r2)
Fires WARNING notification on breach. Missing thresholds = no alerting.
Schema designed to be extensible for Card 2 (Drift) thresholds.
SIENTIAPDE-1986
463 lines
20 KiB
Python
463 lines
20 KiB
Python
from temporalio import activity, workflow
|
|
|
|
with workflow.unsafe.imports_passed_through():
|
|
import time
|
|
import traceback
|
|
import warnings
|
|
from typing import Any
|
|
|
|
import pandas as pd
|
|
from pandas import DataFrame, Index, Series, to_datetime
|
|
from sientia_do.notifications.handlers import CoreNotificationHandler as NotificationHandler
|
|
from sientia_do.notifications.models import NotificationLevel
|
|
from sientia_do.observability.logger import Logger
|
|
from sientia_do.observability.metrics_controller import MetricsController
|
|
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
|
|
|
|
warnings.filterwarnings('ignore', category=RuntimeWarning, message='Degrees of freedom <= 0')
|
|
warnings.filterwarnings(
|
|
'ignore', category=RuntimeWarning, message='invalid value encountered in scalar divide'
|
|
)
|
|
|
|
|
|
class ModelMetrics(SientiaMonitoring):
|
|
"""
|
|
Metrics and statistical analysis activities for the Laborious pipeline.
|
|
|
|
This class centralizes drift/statistical computations and model-quality
|
|
aggregates used by scheduled workflows. Besides producing tabular outputs
|
|
for persistence, it also emits operational metrics (count, lag, error)
|
|
through ``SientiaMonitoring`` so execution health is observable in runtime.
|
|
"""
|
|
|
|
_MAX_DEBUG_DATAFRAME_ROWS = 100
|
|
|
|
def __init__(
|
|
self,
|
|
logger: Logger,
|
|
notification_handler: NotificationHandler,
|
|
metrics_controller: MetricsController,
|
|
):
|
|
SientiaMonitoring.__init__(self, logger, notification_handler, metrics_controller)
|
|
|
|
def close(self) -> None:
|
|
"""
|
|
Shutdown monitoring resources associated with model metrics activities.
|
|
|
|
This is invoked during worker teardown to flush/close metric controller
|
|
internals and prevent dangling telemetry tasks.
|
|
"""
|
|
SientiaMonitoring.shutdown(self)
|
|
|
|
def __del__(self):
|
|
self.close()
|
|
|
|
def _debug_dataframe(self, message: str, data: Any, metadata: dict[str, Any]) -> None:
|
|
"""
|
|
Log dataframe content only when row count is below the configured threshold
|
|
|
|
Args:
|
|
- message (str): Base log message to identify the dataframe in logs
|
|
- data (Any): Dataframe-like payload to be logged
|
|
- metadata (dict[str, Any]): Workflow metadata for contextual logging
|
|
"""
|
|
self.debug(
|
|
build_dataframe_debug_message(
|
|
message=message,
|
|
data=data,
|
|
max_rows=self._MAX_DEBUG_DATAFRAME_ROWS,
|
|
),
|
|
metadata,
|
|
)
|
|
|
|
def _drift_analyze_stage_error(
|
|
self,
|
|
exc: Exception,
|
|
context: str,
|
|
metadata: dict[str, Any],
|
|
core_labels: dict[str, Any],
|
|
) -> None:
|
|
"""
|
|
Log analyzer failure for a drift stage and increment the analyze error metric.
|
|
|
|
Args:
|
|
- exc (Exception): Failure raised by ``sientia_model``.
|
|
- context (str): Short label for the log line (e.g. univariate detection).
|
|
- metadata (dict[str, Any]): Workflow metadata for logging.
|
|
- core_labels (dict[str, Any]): Tags from ``get_core_labels`` for metrics.
|
|
"""
|
|
self.error(f'{context}: {exc}', metadata)
|
|
self.emit_metric_sync(metric_object=metrics.MODEL_ANALYZE_ERROR_COUNT, tags=core_labels)
|
|
|
|
def get_drift_metrics(
|
|
self,
|
|
reference_data: DataFrame,
|
|
target_data: DataFrame,
|
|
target_name: str,
|
|
reference_columns: Index,
|
|
drift_metrics: list[str],
|
|
chunk_period: str,
|
|
metadata: dict[str, Any],
|
|
) -> DataFrame:
|
|
"""
|
|
Compute univariate and multivariate drift outputs and merge them into one dataframe.
|
|
|
|
The method orchestrates three analysis stages (univariate drift,
|
|
multivariate drift, and dataframe projection), emitting lag/count/error
|
|
metrics for each stage independently so failures are attributable.
|
|
|
|
Args:
|
|
- reference_data (DataFrame): Baseline dataset representing expected behavior.
|
|
- target_data (DataFrame): Current analysis dataset to compare against reference.
|
|
- target_name (str): Target column name used by ``DriftAnalysis`` config.
|
|
- reference_columns (Index): Feature columns evaluated for drift.
|
|
- drift_metrics (list[str]): Enabled univariate methods.
|
|
- chunk_period (str): Time bucket granularity used by analysis methods.
|
|
- metadata (dict[str, Any]): Workflow metadata for logs and notifications.
|
|
|
|
Return:
|
|
DataFrame: Consolidated drift dataframe from ``get_drift_metrics_dataframe`` using
|
|
``method`` / ``value`` (and optional ``threshold``, ``drift_type``), ready for
|
|
activity-level formatting before Postgres export.
|
|
"""
|
|
# ``DriftAnalysis`` uses truthiness checks on ``features`` (e.g. ``if not features``);
|
|
# a pandas ``Index`` is ambiguous in boolean context — normalize to a list.
|
|
feature_names: list[str] = list(reference_columns)
|
|
|
|
config = {
|
|
'target': target_name,
|
|
'prediction': 'prediction',
|
|
'timestamp': 'timestamp',
|
|
'features': feature_names,
|
|
}
|
|
|
|
drift_analysis = DriftAnalysis(config=config)
|
|
|
|
self._debug_dataframe(
|
|
f'Reference data: Size {reference_data.shape}', reference_data, metadata
|
|
)
|
|
|
|
self._debug_dataframe(f'Target data: Size {target_data.shape}', target_data, metadata)
|
|
|
|
core_labels = self.get_core_labels(metadata, operation_type='detect_univariate_drift')
|
|
start_time = time.time()
|
|
try:
|
|
univariate_drift = drift_analysis.detect_univariate_drift(
|
|
reference_df=reference_data,
|
|
analysis_df=target_data,
|
|
features=feature_names,
|
|
timestamp_col=config['timestamp'],
|
|
methods=drift_metrics,
|
|
chunk_period=chunk_period,
|
|
)
|
|
except Exception as e:
|
|
if isinstance(e, DriftInsufficientDataError):
|
|
raise
|
|
self._drift_analyze_stage_error(
|
|
e, 'Error detecting univariate drift', metadata, core_labels
|
|
)
|
|
raise
|
|
self.observe_lag_sync(start_time, metrics.MODEL_ANALYZE_LAG, core_labels)
|
|
self.emit_metric_sync(metric_object=metrics.MODEL_ANALYZE_COUNT, tags=core_labels)
|
|
|
|
core_labels = self.get_core_labels(metadata, operation_type='detect_multivariate_drift')
|
|
start_time = time.time()
|
|
try:
|
|
multivariate_drift = drift_analysis.detect_multivariate_drift(
|
|
reference_df=reference_data,
|
|
analysis_df=target_data,
|
|
features=feature_names,
|
|
timestamp_col=config['timestamp'],
|
|
chunk_period=chunk_period,
|
|
)
|
|
except Exception as e:
|
|
if isinstance(e, DriftInsufficientDataError):
|
|
raise
|
|
self._drift_analyze_stage_error(
|
|
e, 'Error detecting multivariate drift', metadata, core_labels
|
|
)
|
|
raise
|
|
self.observe_lag_sync(start_time, metrics.MODEL_ANALYZE_LAG, core_labels)
|
|
self.emit_metric_sync(metric_object=metrics.MODEL_ANALYZE_COUNT, tags=core_labels)
|
|
|
|
start_time = time.time()
|
|
core_labels = self.get_core_labels(metadata, operation_type='get_drift_metrics_dataframe')
|
|
try:
|
|
drift_df = drift_analysis.get_drift_metrics_dataframe(
|
|
univariate_drift=univariate_drift,
|
|
multivariate_drift=multivariate_drift,
|
|
)
|
|
except Exception as e:
|
|
if isinstance(e, DriftInsufficientDataError):
|
|
raise
|
|
self._drift_analyze_stage_error(
|
|
e, 'Error building drift metrics dataframe', metadata, core_labels
|
|
)
|
|
raise
|
|
self.observe_lag_sync(start_time, metrics.MODEL_ANALYZE_LAG, core_labels)
|
|
self.emit_metric_sync(metric_object=metrics.MODEL_ANALYZE_COUNT, tags=core_labels)
|
|
|
|
return drift_df
|
|
|
|
@staticmethod
|
|
def _to_naive_utc(series: Series) -> Series:
|
|
"""
|
|
Parse ``series`` as datetime and return a TZ-naive UTC copy.
|
|
|
|
``sientia_model.analytics.drift_analysis.DriftAnalysis`` preserves the
|
|
timezone of the input dataframe in its outputs, while target rows
|
|
loaded from PostgreSQL come in with ``+00:00``. Forcing both sides of
|
|
a comparison to TZ-naive UTC keeps ``isin`` / ``floor`` operations
|
|
deterministic regardless of how the analyzer constructs its
|
|
timestamps.
|
|
|
|
Args:
|
|
- series (Series): Input series containing datetime-parseable values.
|
|
|
|
Return:
|
|
Series: Datetime64 series with ``tz=None`` representing UTC instants.
|
|
"""
|
|
parsed = to_datetime(series)
|
|
if getattr(parsed.dt, 'tz', None) is not None:
|
|
parsed = parsed.dt.tz_convert('UTC').dt.tz_localize(None)
|
|
return parsed
|
|
|
|
@activity.defn(name='calculate_drift')
|
|
def calculate_drift(self, input_data: dict[str, Any]) -> list[dict]:
|
|
"""
|
|
Calculate drift metrics for a model.
|
|
|
|
Args:
|
|
input_data (dict[str, Any]): Input data containing:
|
|
- metadata (dict): Workflow execution metadata
|
|
- model_name (str): Name of the MLFlow model to calculate drift for
|
|
- reference_data (pd.DataFrame): Reference data for the model
|
|
- target_data (pd.DataFrame): Target data for calculating drift
|
|
- target_name (str): Name of the target column
|
|
- drift_metrics (list[str]): List of drift metrics to calculate
|
|
"""
|
|
metadata = input_data['metadata']
|
|
model_name = input_data['model_name']
|
|
model_id = input_data['model_id']
|
|
reference_raw_data = input_data['reference_data']
|
|
target_data = DataFrame(input_data['target_data'])
|
|
target_name = input_data['target_name']
|
|
drift_metrics = input_data['drift_metrics']
|
|
chunk_period = input_data['chunk_period']
|
|
|
|
if chunk_period not in ['min', 's']:
|
|
self.error(f'Invalid chunk period: {chunk_period}', metadata)
|
|
raise ValueError(f'Invalid chunk period: {chunk_period}, must be "min" or "s"')
|
|
|
|
self.info(f'Calculating drift for model {model_name}', metadata)
|
|
|
|
target_data = target_data.pivot(index='timestamp', columns='variable', values='value')
|
|
target_data['timestamp'] = target_data.index
|
|
# Keep timestamps as datetime: DriftAnalysis._chunk_dataframe relies on
|
|
# ``pd.Grouper(freq=...)`` which rejects string timestamp columns.
|
|
target_data['timestamp'] = to_datetime(target_data['timestamp'])
|
|
target_data = target_data.reset_index(drop=True)
|
|
target_data.dropna(inplace=True)
|
|
|
|
if reference_raw_data is not None:
|
|
self.info('Using reference data', metadata)
|
|
reference_data = DataFrame(reference_raw_data)
|
|
if 'timestamp' in reference_data.columns:
|
|
reference_data['timestamp'] = to_datetime(reference_data['timestamp'])
|
|
accurate = True
|
|
else:
|
|
# Get 30% first rows of target_data
|
|
self.warning('Using 30% first rows of target data as reference data', metadata)
|
|
target_data.sort_values(by='timestamp', ascending=True, inplace=True)
|
|
reference_data = target_data.head(int(len(target_data) * 0.3))
|
|
accurate = False
|
|
|
|
self.send_notification(
|
|
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=reference_data.to_csv(),
|
|
)
|
|
|
|
reference_columns = reference_data.drop(
|
|
columns=[target_name, 'timestamp', 'target', 'prediction'], errors='ignore'
|
|
).columns
|
|
|
|
try:
|
|
drift_df = self.get_drift_metrics(
|
|
reference_data=reference_data,
|
|
target_data=target_data,
|
|
target_name=target_name,
|
|
reference_columns=reference_columns,
|
|
drift_metrics=drift_metrics,
|
|
chunk_period=chunk_period,
|
|
metadata=metadata,
|
|
)
|
|
except Exception as e:
|
|
if isinstance(e, DriftInsufficientDataError):
|
|
self.error(str(e), metadata)
|
|
notification_id = e.notification_id
|
|
notification_message = str(e)
|
|
else:
|
|
self.error(f'Error getting drift metrics: {e}', metadata)
|
|
notification_id = 'MODEL_METRICS_GET_DRIFT_METRICS_ERROR'
|
|
notification_message = f'Error getting drift metrics: {e}'
|
|
self.send_notification(
|
|
metadata=metadata,
|
|
notification_id=notification_id,
|
|
message=notification_message,
|
|
block='model_metrics',
|
|
level=NotificationLevel.ERROR,
|
|
attachment_content=traceback.format_exc(),
|
|
)
|
|
raise
|
|
|
|
# Drop chunks whose floored timestamp does not appear in the analysis window.
|
|
# ``DriftAnalysis`` chunks over ``analysis_df``; this only excludes rows that
|
|
# do not belong to the current target window (e.g. stray merged reference rows).
|
|
target_floor = self._to_naive_utc(target_data['timestamp']).dt.floor(chunk_period)
|
|
drift_floor = self._to_naive_utc(drift_df['timestamp']).dt.floor(chunk_period)
|
|
drift_df = drift_df[drift_floor.isin(target_floor)]
|
|
|
|
if drift_df.empty:
|
|
self.warning(
|
|
'No drift metrics found after dropping rows where timestamp is not in target data',
|
|
metadata,
|
|
)
|
|
return []
|
|
|
|
# Analyzer emits diagnostic columns that are not stored in ``sientia_data.drift_metrics``.
|
|
drift_df = drift_df.drop(columns=['threshold', 'drift_type'], errors='ignore')
|
|
|
|
drift_df['model_id'] = str(model_id)
|
|
drift_df['accurate'] = accurate
|
|
|
|
# ``timestamp`` is overridden with the most recent target instant so
|
|
# every persisted row shares a single business timestamp (the run's
|
|
# logical "now"), matching what downstream consumers expect.
|
|
latest_target_timestamp = self._to_naive_utc(target_data['timestamp']).max()
|
|
drift_df['timestamp'] = (
|
|
pd.Timestamp(latest_target_timestamp)
|
|
.tz_localize('UTC')
|
|
.strftime(DATETIME_FORMAT_WITH_TZ)
|
|
)
|
|
|
|
# ``chunk_start_date`` / ``chunk_end_date`` may carry nanosecond
|
|
# precision (beyond ``timestamptz`` microseconds), so serialize as ISO
|
|
# text for the ``text`` Postgres columns.
|
|
for column in ('chunk_start_date', 'chunk_end_date'):
|
|
drift_df[column] = drift_df[column].apply(
|
|
lambda value: pd.Timestamp(value).isoformat() if pd.notna(value) else None
|
|
)
|
|
|
|
self._debug_dataframe(f'Drift dataframe: Size {drift_df.shape}', drift_df, metadata)
|
|
|
|
return drift_df.to_dict(orient='records')
|
|
|
|
@activity.defn(name='calculate_simple_metrics')
|
|
def calculate_simple_metrics(self, input_data: dict[str, Any]) -> list[dict]:
|
|
"""
|
|
Calculate simple regression metrics for a model using RegressionMetrics.
|
|
|
|
Args:
|
|
input_data: Input data containing:
|
|
- metadata (dict): Workflow execution metadata
|
|
- model_id (str): ID of the MLFlow model
|
|
- 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:
|
|
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 = list(input_data['metrics'])
|
|
interval_minutes = input_data['interval_minutes']
|
|
model_type = input_data.get('model_type')
|
|
|
|
data_size = target_data.shape[0]
|
|
|
|
# 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,
|
|
)
|
|
|
|
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)
|
|
|
|
# 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)
|
|
|
|
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()
|
|
data['data_size'] = data_size
|
|
data['interval_minutes'] = interval_minutes
|
|
|
|
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')
|