diff --git a/docs/E2E_TEST_REPORT.md b/docs/E2E_TEST_REPORT.md deleted file mode 100644 index 6c2ca24..0000000 --- a/docs/E2E_TEST_REPORT.md +++ /dev/null @@ -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`. diff --git a/docs/drift_insufficient_data_explicit_failure.md b/docs/drift_insufficient_data_explicit_failure.md deleted file mode 100644 index 0f6ce97..0000000 --- a/docs/drift_insufficient_data_explicit_failure.md +++ /dev/null @@ -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. diff --git a/e2e/test_minimal_retrain.py b/e2e/test_minimal_retrain.py index e108aef..579658a 100644 --- a/e2e/test_minimal_retrain.py +++ b/e2e/test_minimal_retrain.py @@ -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) diff --git a/input_sample.json b/input_sample.json index da3ef69..f3339f9 100644 --- a/input_sample.json +++ b/input_sample.json @@ -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" + } } ] } diff --git a/laborious/activities/mlflow.py b/laborious/activities/mlflow.py index 6cf311d..131193a 100644 --- a/laborious/activities/mlflow.py +++ b/laborious/activities/mlflow.py @@ -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: