SIENTIAPDE-1646
SIENTIAPDE-1646 Add new scheduling configurations and remove outdated documentation - Introduced new scheduling configurations for minimal retrain, drift analysis, and simple metrics in `input_sample.json`. - Removed obsolete documentation files related to drift analysis and E2E test reports to streamline project resources. - Updated E2E tests for minimal retrain to enhance reporting and error handling during model retraining processes.
This commit is contained in:
vitor-aignosi
@@ -1,48 +0,0 @@
|
||||
# E2E test run report
|
||||
|
||||
**Date:** 2026-05-08
|
||||
**Command:** `source venv/bin/activate && rtk pytest e2e/ -v --tb=short`
|
||||
**Environment:** Linux, Python 3.11.15, pytest 9.0.3
|
||||
|
||||
## Summary
|
||||
|
||||
| Metric | Count |
|
||||
|--------|------:|
|
||||
| Collected | 47 |
|
||||
| **Passed** | **47** |
|
||||
| **Failed** | **0** |
|
||||
|
||||
Full pytest output (when using `rtk`) is stored under `~/.local/share/rtk/tee/` as timestamped `*_pytest.log` files.
|
||||
|
||||
---
|
||||
|
||||
## Regression fixed during this run (drift)
|
||||
|
||||
An initial e2e run failed **3** drift tests with:
|
||||
|
||||
`ValueError: The truth value of a Index is ambiguous`
|
||||
|
||||
**Cause:** `calculate_drift` passes `reference_data.columns` (a **pandas `Index`**) into `ModelMetrics.get_drift_metrics`, which forwards it to `sientia_model.analytics.drift_analysis.DriftAnalysis`. The analyzer uses patterns such as `if not features:` on the feature list. Boolean evaluation of an `Index` raises in pandas.
|
||||
|
||||
**Fix (in `laborious/activities/model_metrics.py`):** At the start of `get_drift_metrics`, normalize with `feature_names: list[str] = list(reference_columns)` and use `feature_names` in the `DriftAnalysis` config and in `detect_univariate_drift` / `detect_multivariate_drift`.
|
||||
|
||||
After this change, the full **`e2e/`** suite was re-run and **all 47 tests passed**.
|
||||
|
||||
---
|
||||
|
||||
## Suite coverage (high level)
|
||||
|
||||
| Area | File(s) | Notes |
|
||||
|------|---------|--------|
|
||||
| Drift workflow | `e2e/test_drift.py` | Happy path, 30% reference fallback, empty target, bad `chunk_period`, empty merge / no export, sub-minute chunking |
|
||||
| Predictions batch | `e2e/test_predictions_batch_*.py` | Main workflow, prediction process gates / repeat, format export |
|
||||
| Child workflows | `e2e/test_child_workflows_e2e.py` | Format + export path |
|
||||
| Minimal retrain | `e2e/test_minimal_retrain.py` | Success / failure / missing target / no data |
|
||||
| MinIO offload | `e2e/test_minio_offload.py` | Load query + batch path |
|
||||
| Simple metrics | `e2e/test_simple_metrics.py` | Persistence, subset, edge cases |
|
||||
|
||||
---
|
||||
|
||||
## Relation to earlier reports
|
||||
|
||||
Older failures described in previous versions of this document (e.g. Jensen–Shannon NULLs vs `drift_metrics.value` NOT NULL, sparse `chunk_period='s'` data) are **not** reproduced in this run. If those topics resurface after data or dependency changes, see the dedicated notes under `docs/` (e.g. drift / JS investigations) and `e2e/scenarios.md`.
|
||||
@@ -1,51 +0,0 @@
|
||||
# Drift: falha explícita quando dados são insuficientes (solução C)
|
||||
|
||||
## Objetivo
|
||||
|
||||
Quando existir **dado alvo no intervalo analisado** mas o pipeline **não produzir nenhuma métrica de drift** (DataFrame vazio após `DriftAnalysis` e/ou após filtros em `calculate_drift`), o sistema deve **responder de forma assertiva**: mensagem clara, identificador de notificação estável e **falha controlada** (exceção ou erro de atividade), **não** apenas `return []` ou workflow que termina sem export e sem erro.
|
||||
|
||||
Isso evita estados ambíguos (ex.: e2e ou operação assumindo que “sucesso + zero linhas” é válido quando na verdade o `chunk_period` ou o volume de pontos não permite calcular drift).
|
||||
|
||||
## Comportamento atual (resumo)
|
||||
|
||||
- `ModelMetrics.calculate_drift` (`laborious/activities/model_metrics.py`): se `get_drift_metrics` retornar vazio ou, após `drift_floor.isin(target_floor)`, ficar vazio, registra **warning** e devolve **`[]`**.
|
||||
- `Drift.run` (`laborious/workflows/drift.py`): se `drift_data` for falsy, **não exporta** e o workflow **termina sem erro**.
|
||||
|
||||
Nenhum dos dois distingue “não havia dados alvo” (já tratado com early return / short-circuit) de **“havia dados, mas nenhuma métrica foi computada”**.
|
||||
|
||||
## Comportamento desejado (solução C)
|
||||
|
||||
### Quando considerar “dados insuficientes / nenhuma métrica com dados presentes”
|
||||
|
||||
Disparar tratamento assertivo se **todas** forem verdade:
|
||||
|
||||
1. Após pivot/`dropna`, `target_data` tem **pelo menos uma linha** com janela temporal válida.
|
||||
2. `reference_data` está disponível (ou o fallback de 30% foi aplicado) de forma que a análise **deveria** poder rodar.
|
||||
3. `reference_columns` (features efetivas do drift univariado) **não está vazio** — caso contrário, falha de configuração, não “insuficiência de amostra”.
|
||||
4. O resultado de `get_drift_metrics` é **vazio**, **ou** fica vazio **somente** após o filtro por `target_floor` / timestamps.
|
||||
|
||||
Opcionalmente, reforçar no **`sientia_model.analytics.drift_analysis.DriftAnalysis`**: se `analysis_df` não for vazio mas **não houver chunks** ou **nenhuma linha** univariada/multivariada, levantar exceção específica ou retornar um código/estrutura que o Laborious traduza em falha explícita (evita duplicar heurística só no Laborious).
|
||||
|
||||
### Resposta assertiva mínima
|
||||
|
||||
1. **Log / notificação** com ID estável, por exemplo: `MODEL_METRICS_DRIFT_INSUFFICIENT_DATA` (ou nome alinhado ao catálogo interno).
|
||||
2. **Mensagem** incluindo contexto acionável: `model_id`, `chunk_period`, contagem de linhas alvo, contagem de features, intervalo de tempo dos dados.
|
||||
3. **Falha de atividade**: em vez de `return []`, **levantar** `ValueError` (ou exceção de domínio dedicada) após enviar a notificação, para o Temporal marcar a execução como falha e testes/e2e poderem distinguir “sem dados” de “falha de insumos para o granularidade pedida”.
|
||||
|
||||
### Ajuste no workflow `Drift` (opcional mas coerente)
|
||||
|
||||
Se `calculate_drift` passar a **lançar** nesse cenário, o workflow já falha na atividade; não é obrigatório alterar `Drift.run` além de garantir que erros não sejam engolidos.
|
||||
|
||||
Se por política **não** se quiser falhar o workflow, documentar explicitamente a alternativa (única): retorno estruturado `{'status': 'insufficient_data', 'detail': ...}` — **não** é a opção C pedida aqui, que prioriza **assertividade e visibilidade**.
|
||||
|
||||
## Impacto em testes
|
||||
|
||||
- Cenários e2e que hoje esperam **sucesso silencioso com zero linhas** (ex.: `test_drift_chunk_period_seconds_preserves_seconds_in_chunk_start_date` com poucos pontos e `chunk_period='s'`) devem ser atualizados para esperar **falha da atividade** com mensagem/notificação, **ou** o teste deve fornecer volume de dados suficiente para gerar ao menos uma linha de drift — conforme o requisito de produto escolhido após esta mudança.
|
||||
|
||||
## Appendix: problema REPEAT / `UniqueViolation` (contexto)
|
||||
|
||||
A “solução A” (novo timestamp na linha inserida pelo REPEAT) **já está parcialmente implementada** em `sientia_do`: `_build_repeat_last_prediction_query` faz `INSERT ... SELECT` com coluna `timestamp` = **parâmetro** `:last_timestamp`, não copia o timestamp da linha anterior.
|
||||
|
||||
O e2e ainda falha quando **o valor de `last_timestamp` passado pelo workflow** (`prediction_process.path_flag_handler`) é **igual** ao timestamp da única predição existente (ex.: seed fixo `2024-01-01 12:00:00+00:00`). Nesse caso o INSERT tenta duplicar `(model_id, timestamp)` e o Postgres aplica `unique_model_id_timestamp`.
|
||||
|
||||
Correção típica: garantir que, no caminho REPEAT, `last_timestamp` seja o **instante do batch / “agora” da execução**, distinto do timestamp da última linha persistida — ou ajustar a atividade para derivar um timestamp único quando `last_timestamp` colide com a última linha.
|
||||
@@ -22,6 +22,7 @@ from datetime import datetime, timedelta, timezone
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
import pandas as pd
|
||||
from sqlalchemy import text
|
||||
from temporalio.testing import WorkflowEnvironment
|
||||
from temporalio.worker import Worker
|
||||
@@ -34,6 +35,7 @@ from e2e.helpers import (
|
||||
)
|
||||
from laborious.activities.activities import Activities
|
||||
from laborious.workflows.minimal_retrain import MinimalRetrain
|
||||
from sientia_model.wrappers.sientia_model import SientiaModel
|
||||
|
||||
# Columns defined by the production DDL for ``sientia_data.log_retrain``.
|
||||
# The legacy ``retrain_reports`` table had ``id`` and ``created_at``; the new
|
||||
@@ -49,6 +51,86 @@ EXPECTED_RETRAIN_REPORT_COLUMNS = [
|
||||
'version',
|
||||
]
|
||||
|
||||
# Matches ``retrain_model`` return ``message`` when ``success`` is True (also written to ``log_retrain.status``).
|
||||
RETRAIN_ACTIVITY_SUCCESS_MESSAGE = 'Model retrained successfully.'
|
||||
|
||||
|
||||
class _FakeSientiaModelForMinimalRetrain(SientiaModel):
|
||||
"""
|
||||
Fake SientiaModel that uses the real SientiaModel lifecycle to surface
|
||||
index-alignment issues during ``retrain()``.
|
||||
|
||||
It intentionally performs strict alignment inside ``_retrain_model``:
|
||||
``y.loc[x.index]``.
|
||||
"""
|
||||
|
||||
def __init__(self, *, target: str = 'sensor_1'):
|
||||
super().__init__(
|
||||
model_type='FakeMinimalRetrain',
|
||||
model_version='0.0.0',
|
||||
model=object(),
|
||||
transformer=object(),
|
||||
)
|
||||
self.target = target
|
||||
self.model_is_fitted = True
|
||||
self.force_retrain_error = False
|
||||
|
||||
def store_model( # type: ignore[override]
|
||||
self,
|
||||
name: str,
|
||||
signature=None,
|
||||
pip_requirements=None,
|
||||
code_path=None,
|
||||
) -> None:
|
||||
# No-op: E2E tests validate workflow persistence, not real MLflow artifacts.
|
||||
return None
|
||||
|
||||
def _predict(self, data: pd.DataFrame):
|
||||
pred = pd.DataFrame({'prediction': [0.5] * len(data)}, index=data.index)
|
||||
return pred, {}
|
||||
|
||||
def _transform(self, data: pd.DataFrame):
|
||||
out = data.drop(columns=[self.target], errors='ignore').copy()
|
||||
out.index = data.index
|
||||
return out, {}
|
||||
|
||||
def _train_transformer(self, train_data: pd.DataFrame, val_data: pd.DataFrame) -> None:
|
||||
return None
|
||||
|
||||
def _train_model(
|
||||
self,
|
||||
x: pd.DataFrame,
|
||||
y: pd.DataFrame,
|
||||
x_val: pd.DataFrame | None = None,
|
||||
y_val: pd.DataFrame | None = None,
|
||||
) -> None:
|
||||
return None
|
||||
|
||||
def _retrain_transformer(self, data: pd.DataFrame) -> None:
|
||||
return None
|
||||
|
||||
def _retrain_model(self, x: pd.DataFrame, y: pd.DataFrame | None) -> None:
|
||||
if self.force_retrain_error:
|
||||
raise RuntimeError('training did not converge')
|
||||
if y is None:
|
||||
return
|
||||
# Strict alignment on purpose to reproduce the production failure mode.
|
||||
_ = y.loc[x.index]
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mlflow_repository_stub():
|
||||
"""
|
||||
Override the shared E2E fixture: return a real fake ``SientiaModel`` wrapper
|
||||
instead of a MagicMock wrapper.
|
||||
"""
|
||||
repo = MagicMock()
|
||||
repo._client = MagicMock()
|
||||
|
||||
wrapper = _FakeSientiaModelForMinimalRetrain(target='sensor_1')
|
||||
repo.get_cached_model = MagicMock(return_value=wrapper)
|
||||
return repo
|
||||
|
||||
|
||||
def _retrain_input(model_id: int, **overrides) -> dict:
|
||||
"""Load and override the minimal-retrain base scenario."""
|
||||
@@ -86,8 +168,6 @@ def _configure_retrain_happy_path(mlflow_repository_stub) -> None:
|
||||
|
||||
- ``_client.get_model_version_by_alias``: returns ``mv`` with a stable
|
||||
``run_id`` (used as ``source_run_id``).
|
||||
- ``get_cached_model``: returns a wrapper exposing inert ``retrain`` and
|
||||
``store_model`` methods.
|
||||
- ``start_run``: returns a context manager yielding a ``run_info`` with
|
||||
run/experiment ids.
|
||||
- ``log_params``: inert.
|
||||
@@ -104,11 +184,6 @@ def _configure_retrain_happy_path(mlflow_repository_stub) -> None:
|
||||
|
||||
mlflow_repository_stub._client.get_model_version_by_alias.return_value = mv_src
|
||||
|
||||
cached_wrapper = MagicMock()
|
||||
cached_wrapper.retrain = MagicMock(return_value=None)
|
||||
cached_wrapper.store_model = MagicMock(return_value=None)
|
||||
mlflow_repository_stub.get_cached_model.return_value = cached_wrapper
|
||||
|
||||
@contextmanager
|
||||
def fake_start_run(**kwargs):
|
||||
run_info = MagicMock()
|
||||
@@ -155,8 +230,6 @@ async def test_minimal_retrain_happy_path_writes_success_report(
|
||||
make_workflow_id('test-retrain-happy'),
|
||||
)
|
||||
|
||||
assert log_artifact_mock.called, 'retrain_model should log the input CSV artifact'
|
||||
|
||||
with postgres_engine.connect() as conn:
|
||||
rows = (
|
||||
conn.execute(
|
||||
@@ -171,14 +244,27 @@ async def test_minimal_retrain_happy_path_writes_success_report(
|
||||
)
|
||||
|
||||
assert len(rows) == 1
|
||||
for column in EXPECTED_RETRAIN_REPORT_COLUMNS:
|
||||
assert column in rows[0], f'Missing log_retrain column: {column}'
|
||||
|
||||
row = rows[0]
|
||||
for column in EXPECTED_RETRAIN_REPORT_COLUMNS:
|
||||
assert column in row, f'Missing log_retrain column: {column}'
|
||||
|
||||
assert row['status'] == RETRAIN_ACTIVITY_SUCCESS_MESSAGE, (
|
||||
"Expected retrain_model to return success (experiment_response['success'] is True). "
|
||||
'Persisted log_retrain.status is the activity message; when success is False the run '
|
||||
'never reaches mlflow.log_artifact — diagnose the retrain failure from status below, '
|
||||
'not from a skipped artifact upload. '
|
||||
f"Got status={row['status']!r}, version={row.get('version')!r}, "
|
||||
f"mlflow_run_id={row.get('mlflow_run_id')!r}."
|
||||
)
|
||||
|
||||
assert log_artifact_mock.called, (
|
||||
'After a successful retrain, retrain_model must call mlflow.log_artifact for the '
|
||||
'input CSV inside start_run.'
|
||||
)
|
||||
|
||||
# ``model_id`` is now ``text``; compare against the stringified id.
|
||||
assert row['model_id'] == str(model_id)
|
||||
assert row['model_name'] == 'test_model'
|
||||
assert row['status'] == 'Model retrained successfully.'
|
||||
assert row['version'] == '7'
|
||||
assert row['mlflow_run_id'] == 'retrain-run-id'
|
||||
# ``mlflow_experiment_id`` is now ``int8``; assert the integer value
|
||||
@@ -213,9 +299,7 @@ async def test_minimal_retrain_failure_writes_report_without_version_columns(
|
||||
|
||||
_seed_retrain_training_rows(postgres_engine, model_id)
|
||||
_configure_retrain_happy_path(mlflow_repository_stub)
|
||||
mlflow_repository_stub.get_cached_model.return_value.retrain.side_effect = RuntimeError(
|
||||
'training did not converge'
|
||||
)
|
||||
mlflow_repository_stub.get_cached_model.return_value.force_retrain_error = True
|
||||
|
||||
input_data = _retrain_input(model_id)
|
||||
|
||||
|
||||
@@ -67,6 +67,77 @@
|
||||
"timestamp",
|
||||
"created_at"
|
||||
]
|
||||
},
|
||||
{
|
||||
"schedule_name": "minimal-retrain-test-runtime",
|
||||
"model_id": "1001",
|
||||
"model_name": "test-runtime",
|
||||
"workflow_type": "minimal_retrain",
|
||||
"frequency": "1h",
|
||||
"max_retry_policy": 1,
|
||||
"query": "select * from sientia_data.laborious_data where model_id = 1 and \"timestamp\" > NOW() - INTERVAL '60 minutes' order by \"timestamp\" desc;",
|
||||
"schema": "sientia_data",
|
||||
"table_name": "log_retrain",
|
||||
"datetime_columns": ["timestamp", "created_at"],
|
||||
"model_config": {
|
||||
"target": "Square"
|
||||
},
|
||||
"active": true,
|
||||
"updated_at": {
|
||||
"$date": "2026-05-07T23:35:01.600Z"
|
||||
}
|
||||
},
|
||||
{
|
||||
"schedule_name": "drift-test-runtime",
|
||||
"model_id": "1001",
|
||||
"model_name": "test-runtime",
|
||||
"workflow_type": "drift",
|
||||
"frequency": "5m",
|
||||
"offset": "2m",
|
||||
"max_retry_policy": 1,
|
||||
"execution_timeout_seconds": 300,
|
||||
"task_timeout_seconds": 300,
|
||||
"interval": 5,
|
||||
"drift_metrics": [
|
||||
"kolmogorov_smirnov",
|
||||
"jensen_shannon",
|
||||
"wasserstein"
|
||||
],
|
||||
"chunk_period": "min",
|
||||
"schema": "sientia_data",
|
||||
"source_table_name": "laborious_data",
|
||||
"target_table_name": "drift_metrics",
|
||||
"model_config": {
|
||||
"target": "Square"
|
||||
},
|
||||
"active": true,
|
||||
"updated_at": {
|
||||
"$date": "2026-05-07T23:35:01.600Z"
|
||||
}
|
||||
},
|
||||
{
|
||||
"schedule_name": "simple-metrics-test-runtime",
|
||||
"model_id": "1001",
|
||||
"model_name": "test-runtime",
|
||||
"workflow_type": "simple_metrics",
|
||||
"frequency": "5m",
|
||||
"offset": "2m",
|
||||
"max_retry_policy": 1,
|
||||
"execution_timeout_seconds": 300,
|
||||
"task_timeout_seconds": 300,
|
||||
"interval_minutes": 5,
|
||||
"metrics": ["rmse", "mse", "mae", "r2"],
|
||||
"schema": "sientia_data",
|
||||
"predictions_table_name": "predictions",
|
||||
"data_table_name": "laborious_data",
|
||||
"target_table_name": "simple_metrics",
|
||||
"model_config": {
|
||||
"target": "Square"
|
||||
},
|
||||
"active": true,
|
||||
"updated_at": {
|
||||
"$date": "2026-05-07T23:35:01.600Z"
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
@@ -322,9 +322,9 @@ class MLFlow(SientiaMonitoring):
|
||||
"""
|
||||
Load the production wrapper and call ``wrapper.predict`` on the prepared feature frame.
|
||||
|
||||
The activity normalizes ``NaN`` to ``None`` for JSON-friendly columns, rebuilds a
|
||||
``timestamp`` column in the internal string format, preserves the original index for
|
||||
alignment, and records ``response_time`` seconds on the output frame. Non-DataFrame
|
||||
The activity normalizes ``NaN`` to ``None`` for JSON-friendly columns, sets the row index
|
||||
the same way as ``retrain_model`` (UTC ``DatetimeIndex`` from ``DATETIME_FORMAT_WITH_TZ``),
|
||||
restores that index on the prediction frame, and records ``response_time``. Non-DataFrame
|
||||
predictions are coerced to a single ``prediction`` column.
|
||||
|
||||
Args:
|
||||
@@ -346,14 +346,12 @@ class MLFlow(SientiaMonitoring):
|
||||
|
||||
self._debug_dataframe('Input data for prediction:', data, metadata)
|
||||
|
||||
input_index = data.index
|
||||
|
||||
data.replace(np.nan, None, inplace=True)
|
||||
|
||||
data['timestamp'] = data.index
|
||||
data['timestamp'] = to_datetime(
|
||||
data['timestamp'], format=DATETIME_FORMAT_WITH_TZ
|
||||
).dt.strftime(DATETIME_FORMAT)
|
||||
data.index = pd.DatetimeIndex(
|
||||
to_datetime(data.index, format=DATETIME_FORMAT_WITH_TZ, utc=True)
|
||||
)
|
||||
input_index = data.index
|
||||
|
||||
try:
|
||||
wrapper = self.mlflow_repository.get_cached_model(
|
||||
@@ -493,14 +491,11 @@ class MLFlow(SientiaMonitoring):
|
||||
data = data.pivot(index='timestamp', columns='variable', values='value')
|
||||
data.fillna(np.nan, inplace=True)
|
||||
data.columns.name = None
|
||||
data.index.name = None
|
||||
|
||||
data['timestamp'] = data.index
|
||||
data['timestamp'] = to_datetime(
|
||||
data['timestamp'], format=DATETIME_FORMAT_WITH_TZ
|
||||
).dt.strftime(DATETIME_FORMAT)
|
||||
data['timestamp'] = to_datetime(data['timestamp'], format=DATETIME_FORMAT)
|
||||
|
||||
data.columns.name = None
|
||||
data.index = pd.DatetimeIndex(
|
||||
to_datetime(data.index, format=DATETIME_FORMAT_WITH_TZ, utc=True)
|
||||
)
|
||||
|
||||
target = model_config.get('target')
|
||||
if target is None:
|
||||
|
||||
Reference in New Issue
Block a user