Merge pull request #22 from Aignosi/feature/SIENTIAPDE-1430

Implement Advanced Training Features, Enhance Data Preprocessing, and Bolster Testing & Documentation
This commit is contained in:
Bruno Domingues
2025-12-19 16:42:51 -03:00
committed by GitHub
29 changed files with 2974 additions and 173 deletions

View File

@@ -0,0 +1,227 @@
# Changelog de Parâmetros do Pipeline de Treinamento
Este documento descreve as alterações nos parâmetros de entrada do pipeline Temporal para treinamento de modelos.
## Resumo das Alterações
### Parâmetros ALTERADOS (Breaking Changes)
| Parâmetro | Tipo Anterior | Tipo Novo | Descrição |
|-----------|---------------|-----------|-----------|
| `lag_train` | `int` | `dict[str, int]` | Agora é um dicionário com lag por variável. Ex: `{"var1": 2, "var2": 3}` |
| `lag_val` | `int` | `dict[str, int]` | Agora é um dicionário com lag por variável. Ex: `{"var1": 1, "var2": 1}` |
### Parâmetros NOVOS (Obrigatórios)
| Parâmetro | Tipo | Descrição | Valores Válidos |
|-----------|------|-----------|-----------------|
| `model_name` | `str` | Nome do tipo de modelo | `"Linear Regression"`, `"Polynomial Regression"` |
| `degree` | `int` | Grau do polinômio (1 = linear) | `>= 1` |
| `interaction_only` | `bool` | Apenas termos de interação para polinomial | `true`, `false` |
| `nan_treatment` | `str` | Tratamento de valores NaN | `"drop"`, `"linear interpolation"`, `"fill linear"` |
| `scaler_name` | `str` | Nome do scaler a usar | `"Standard Scaler"`, `"None"` |
### Parâmetros NOVOS (Opcionais)
| Parâmetro | Tipo | Descrição | Default |
|-----------|------|-----------|---------|
| `start_date` | `str \| null` | Data inicial para filtrar dados (formato: `"YYYY-MM-DD HH:MM:SS"`) | `null` |
| `end_date` | `str \| null` | Data final para filtrar dados (formato: `"YYYY-MM-DD HH:MM:SS"`) | `null` |
| `support_filters` | `dict \| null` | Filtros customizados por variável | `{}` |
| `static_threshold` | `int \| null` | Threshold para remoção de janelas estáticas (1-1000). Só usado quando `rem_static_win` é `true`. | `1` |
---
## Exemplo de Input Completo
### Formato ANTERIOR (não funciona mais):
```json
{
"experiment_run_id": 123,
"target_variable": "temperatura",
"variable_columns": ["pressao", "umidade", "velocidade"],
"lag_train": 2,
"lag_val": 1,
"rem_static_win": true,
"low_lim": {"pressao": 0, "umidade": 0, "velocidade": 0},
"upp_lim": {"pressao": 100, "umidade": 100, "velocidade": 50},
"window": 0,
"use_scaler": true,
"include_ar": false,
"bucket_name": "training-data",
"file_name": "dataset.csv",
"line_separator": ";",
"decimal_separator": ",",
"train_size": 80,
"shuffle": false,
"experiment_name": "modelo-temperatura",
"removed_intervals": []
}
```
### Formato NOVO (obrigatório):
```json
{
"experiment_run_id": 123,
"target_variable": "temperatura",
"variable_columns": ["pressao", "umidade", "velocidade"],
"lag_train": {
"pressao": 2,
"umidade": 2,
"velocidade": 2
},
"lag_val": {
"pressao": 1,
"umidade": 1,
"velocidade": 1
},
"rem_static_win": true,
"static_threshold": null,
"low_lim": {"pressao": 0, "umidade": 0, "velocidade": 0},
"upp_lim": {"pressao": 100, "umidade": 100, "velocidade": 50},
"window": 0,
"use_scaler": true,
"include_ar": false,
"bucket_name": "training-data",
"file_name": "dataset.csv",
"line_separator": ";",
"decimal_separator": ",",
"train_size": 80,
"shuffle": false,
"experiment_name": "modelo-temperatura",
"removed_intervals": [],
"model_name": "Linear Regression",
"degree": 1,
"interaction_only": false,
"nan_treatment": "drop",
"scaler_name": "Standard Scaler",
"start_date": null,
"end_date": null,
"support_filters": {}
}
```
---
## Exemplo para Polynomial Regression
```json
{
"experiment_run_id": 124,
"target_variable": "temperatura",
"variable_columns": ["pressao", "umidade"],
"lag_train": {
"pressao": 0,
"umidade": 0
},
"lag_val": {
"pressao": 0,
"umidade": 0
},
"rem_static_win": false,
"low_lim": {"pressao": 0, "umidade": 0},
"upp_lim": {"pressao": 100, "umidade": 100},
"window": 0,
"use_scaler": true,
"include_ar": false,
"bucket_name": "training-data",
"file_name": "dataset.csv",
"line_separator": ";",
"decimal_separator": ",",
"train_size": 80,
"shuffle": false,
"experiment_name": "modelo-polinomial",
"removed_intervals": [],
"model_name": "Polynomial Regression",
"degree": 2,
"interaction_only": false,
"nan_treatment": "linear interpolation",
"scaler_name": "Standard Scaler",
"start_date": "2024-01-01 00:00:00",
"end_date": "2024-12-31 23:59:59",
"support_filters": {}
}
```
---
## Exemplo com Intervalos Removidos
```json
{
"removed_intervals": [
["2024-03-01 00:00:00", "2024-03-15 23:59:59"],
["2024-06-01 00:00:00", "2024-06-30 23:59:59"]
]
}
```
---
## Parâmetros Logados no MLflow
Os seguintes parâmetros são agora logados no MLflow:
| Parâmetro MLflow | Descrição |
|------------------|-----------|
| `model_name` | Nome do modelo (`Linear Regression` ou `Polynomial Regression`) |
| `models_params` | `{"degree": int, "interaction_only": bool}` |
| `target_variable` | Variável alvo |
| `input_variables` | Lista de variáveis de entrada |
| `nan_treatment` | Tratamento de NaN |
| `lag_train` | Dicionário de lags para treino |
| `lag_transform` | Dicionário de lags para transformação |
| `static_threshold` | Threshold para janelas estáticas (1 ou null) |
| `lower_limits` | Limites inferiores por variável |
| `upper_limits` | Limites superiores por variável |
| `scaler_name` | Nome do scaler |
| `scaler_params` | Parâmetros do scaler (mean, variance) |
| `include_ar` | Se inclui variável autoregressiva |
| `train_size` | Proporção de treino (0.0 - 1.0) |
| `test_size` | Proporção de teste (0.0 - 1.0) |
| `start_date` | Data inicial (ou null) |
| `end_date` | Data final (ou null) |
| `removed_intervals` | Lista de intervalos removidos |
| `retrain` | Sempre `false` para novos modelos |
| `support_filters` | Filtros customizados |
---
## Validações de Negócio
O sistema valida automaticamente:
1. **`train_size`**: Deve estar entre 10 e 100
2. **`variable_columns`**: Não pode estar vazio
3. **`lag_train` / `lag_val`**: Todos os valores devem ser >= 0
4. **`window`**: Deve ser >= 0
5. **`degree`**: Deve ser >= 1
6. **`nan_treatment`**: Deve ser `"drop"`, `"linear interpolation"` ou `"fill linear"`
7. **`scaler_name`**: Deve ser `"Standard Scaler"` ou `"None"`
8. **`model_name`**: Deve ser `"Linear Regression"` ou `"Polynomial Regression"`
9. **`low_lim` / `upp_lim`**: Devem ter as mesmas chaves, e `low_lim[var] < upp_lim[var]`
10. **`bucket_name` / `file_name` / `experiment_name`**: Não podem estar vazios
11. **`degree` vs `model_name`**: Se `model_name` = "Polynomial Regression", `degree` deve ser >= 2; se "Linear Regression", `degree` deve ser = 1
12. **`removed_intervals`**: Cada elemento deve ser lista/tupla com pelo menos 2 elementos (start, end)
13. **`target_variable`**: Não pode estar vazio
14. **`static_threshold`**: Se `rem_static_win` = `true` e `static_threshold` tiver valor, deve estar entre 1 e 1000 (inclusive). Se `null`, assume valor `1`.
---
## Arquivos Modificados
- `model_manager/sientia/models.py` - `DataPreprocessor` e `LinearRegressionModel`
- `model_manager/sientia/metrics.py` - Funções RCE adicionadas
- `model_manager/utils/models/train_model_params.py` - Novos parâmetros
- `model_manager/utils/repository/training_repository.py` - Uso dos novos parâmetros
- `model_manager/utils/repository/model_repository.py` - Logging no MLflow

206
README.md
View File

@@ -39,6 +39,12 @@ An enterprise-grade ML model training orchestration platform built on Temporal.
- [Testing](#testing)
- [Test Structure](#test-structure)
- [Test Execution](#test-execution)
- [Integration Tests](#integration-tests)
- [Running Integration Tests](#running-integration-tests)
- [Test Scenarios](#test-scenarios)
- [Scenario File Structure](#scenario-file-structure)
- [Creating New Scenarios](#creating-new-scenarios)
- [Important Validations](#important-validations)
- [Monitoring and Metrics](#monitoring-and-metrics)
- [Application Health Metrics](#application-health-metrics)
- [Training Metrics](#training-metrics)
@@ -73,6 +79,7 @@ An enterprise-grade ML model training orchestration platform built on Temporal.
### Core Functionality
- **ML Model Training Pipeline**: Complete training workflow from validation to deployment using MLFlow
- **Polynomial Regression Support**: Configurable polynomial degree with interaction terms and mandatory scaler validation
- **Automated File Cleanup**: Scheduled cleanup of stale files from MinIO and local filesystem
- **Temporal Workflow Orchestration**: Robust workflow management with granular retry policies and fault tolerance
- **Parameter Validation**: Defense-in-depth validation with business rules and type checking
@@ -87,14 +94,19 @@ An enterprise-grade ML model training orchestration platform built on Temporal.
- **Notification System**: Integrated alerting and notification management via MongoDB
- **Scalable Architecture**: Kubernetes-ready deployment with horizontal scaling support
- **MLFlow Integration**: Seamless model and artifact persistence to MLFlow tracking server
- **Per-Variable Lag Configuration**: Flexible lag settings for each variable independently
- **Date Range Filtering**: Filter training data by start/end dates and removed intervals
- **NaN Treatment Options**: Configurable handling of missing values (drop, linear interpolation)
- **RCE Drift Metrics**: Reduced Coulomb Energy metrics for drift detection
### Development & Quality Assurance
- **Code Quality Tools**: Ruff (linting/formatting), mypy (type checking), Bandit (security analysis)
- **Automated Validation**: Pre-commit validation script (`validate.sh`) and CI/CD integration
- **Comprehensive Testing**: pytest with async support and **99%+ code coverage** 🎯
- **Comprehensive Testing**: pytest with async support and **100% code coverage** 🎯
- **Type Safety**: Static type checking with mypy for improved code reliability
- **Coverage Visualization**: Integration with Coverage Gutters for real-time coverage feedback
- **Automated Versioning**: Semantic versioning based on branch patterns (release/*, feature/*, fix/*, rc/*)
- **Integration Test Scenarios**: JSON-based test scenarios with batch execution support
## Architecture
@@ -173,6 +185,8 @@ The Model Manager system uses a Temporal-based workflow architecture with clear
- Integration with TrainingRepository for business logic separation
- MLFlow model saving and artifact management
- MinIO object storage operations
- **Polynomial Regression**: Support for configurable degree and interaction terms
- **Training Predictions**: Calculates y_train_pred before denormalization for accurate metrics
- **Cleanup**: File and directory cleanup operations
- `cleanup_minio_files()`: Removes stale files from MinIO based on timestamp prefixes
- `cleanup_temp_directories()`: Cleans local temporary directories
@@ -192,8 +206,8 @@ The Model Manager system uses a Temporal-based workflow architecture with clear
- `training_repository.py`: Training business logic and operations
- `model_repository.py`: MLFlow artifact generation and model persistence
- **Models**: Data models and schemas
- `train_model_params.py`: Training parameters model
- `train_model_result.py`: Training result model
- `train_model_params.py`: Training parameters model with comprehensive validation
- `train_model_result.py`: Training result model (includes y_train_pred)
- `experiment_status.py`: Experiment status enum
- **Key Features**:
- Environment variable-based configuration with sensible defaults
@@ -270,7 +284,7 @@ The **TrainModel** workflow orchestrates the complete ML model training pipeline
{
"experiment_run_id": 123,
"target_variable": "price",
"variable_columns": ["feature1", "feature2", "price"],
"variable_columns": ["feature1", "feature2"],
"train_size": 80,
"shuffle": true,
"use_scaler": true,
@@ -279,14 +293,23 @@ The **TrainModel** workflow orchestrates the complete ML model training pipeline
"file_name": "training_data.csv",
"line_separator": "\n",
"decimal_separator": ".",
"lag_train": 5,
"lag_val": 3,
"lag_train": {"feature1": 0, "feature2": 2},
"lag_val": {"feature1": 0, "feature2": 1},
"rem_static_win": false,
"low_lim": {"feature1": 0.0, "feature2": 0.0, "price": 0.0},
"upp_lim": {"feature1": 100.0, "feature2": 100.0, "price": 1000.0},
"static_threshold": null,
"low_lim": {"feature1": 0.0, "feature2": 0.0},
"upp_lim": {"feature1": 100.0, "feature2": 100.0},
"window": 10,
"experiment_name": "production_model_v1",
"removed_intervals": []
"removed_intervals": [],
"model_name": "Linear Regression",
"degree": 1,
"interaction_only": false,
"nan_treatment": "drop",
"start_date": null,
"end_date": null,
"scaler_name": "Standard Scaler",
"support_filters": {}
}
```
@@ -322,14 +345,22 @@ The workflow implements 5 different retry policies optimized for each operation
#### Business Validation Rules
The workflow validates 10 business rules beyond type checking:
The workflow validates comprehensive business rules beyond type checking:
1. **train_size**: Must be between 1-99%
1. **train_size**: Must be between 10-100%
2. **variable_columns**: Cannot be empty
3. **lag_train, lag_val, window**: Must be positive integers
4. **low_lim/upp_lim**: Must have same keys and low < upp for each variable
5. **target_variable**: Must be in variable_columns
6. **bucket_name, file_name, experiment_name**: Cannot be empty or whitespace
3. **lag_train, lag_val**: Per-variable dictionaries with non-negative values
4. **window**: Must be non-negative integer
5. **low_lim/upp_lim**: Must have same keys and low < upp for each variable
6. **target_variable**: Cannot be empty
7. **bucket_name, file_name, experiment_name**: Cannot be empty or whitespace
8. **degree**: Must be at least 1; must be >= 2 for Polynomial Regression
9. **nan_treatment**: Must be one of 'drop', 'linear interpolation', 'fill linear'
10. **scaler_name**: Must be 'Standard Scaler' or 'None'
11. **model_name**: Must be 'Linear Regression' or 'Polynomial Regression'
12. **Polynomial Regression requires Scaler**: Models with degree > 1 must have a scaler to prevent numerical overflow
13. **Linear Regression requires degree 1**: Linear models must have degree = 1
14. **static_threshold**: When `rem_static_win` is true and `static_threshold` has a value, it must be between 1 and 1000 (inclusive). If null, defaults to 1.
### Cleanup Files Workflow (`cleanup_files.py`)
@@ -871,6 +902,134 @@ pytest tests/activities/test_training.py
pytest tests/workflows/test_train_model.py
```
### Integration Tests
The project includes integration tests that validate the complete training workflow against a running Temporal cluster. These tests use JSON-based scenario files for easy configuration and maintenance.
#### Running Integration Tests
```bash
# List available scenarios
python scripts/run_training_test.py --list
# Run a specific test scenario
python scripts/run_training_test.py --scenario 01-linear-regression-basic
# Run with custom CSV data file
python scripts/run_training_test.py --scenario 03-polynomial-regression-degree2 --csv /path/to/data.csv
# Run ALL scenarios sequentially with summary report
python scripts/run_training_test.py --all
# Run all scenarios with custom CSV
python scripts/run_training_test.py --all --csv docs/custom-data.csv
```
#### Batch Execution Output
When running all scenarios with `--all`, the script provides:
- Progress indicators for each scenario (`[1/10] Running scenario: ...`)
- Status symbols (✓ for passed, ✗ for failed)
- Final summary with total/passed/failed counts
- Detailed error messages for failed scenarios
- Exit code 0 if all pass, 1 if any fail
Example output:
```
Running 10 scenarios...
[1/10] Running scenario: 01-linear-regression-basic
Loaded scenario: 01-linear-regression-basic
Uploaded CSV to MinIO: test-model-data-20231219-120000.csv
Created experiment_run with ID: 42
Workflow started: train-model-test-abc123
[1/10] ✓ 01-linear-regression-basic
...
============================================================
SUMMARY
============================================================
Total: 10 | Passed: 9 | Failed: 1
============================================================
✓ PASSED:
- 01-linear-regression-basic
- 02-linear-regression-with-scaler
...
✗ FAILED:
- 05-linear-regression-with-lags
Error: Failed to start Temporal workflow: connection refused
```
#### Test Scenarios
Test scenarios are defined as JSON files in `docs/test-scenarios/`. Each scenario configures a complete training workflow with specific parameters:
| Scenario | Description | Key Features |
|----------|-------------|--------------|
| `01-linear-regression-basic` | Basic linear regression | No scaler, no lags |
| `02-linear-regression-with-scaler` | Linear regression with normalization | Standard Scaler enabled |
| `03-polynomial-regression-degree2` | Polynomial regression (degree 2) | Requires scaler (mandatory) |
| `04-polynomial-regression-degree3` | Polynomial regression (degree 3) | Requires scaler (mandatory) |
| `05-linear-regression-with-lags` | Linear regression with lag features | Lag train/val configuration |
| `06-linear-regression-nan-interpolation` | Linear regression with NaN handling | `nanTreatment: "linear interpolation"` |
| `07-linear-regression-static-window-removal` | Linear regression with static window removal | `remStaticWin: true` |
| `08-linear-regression-with-limits` | Linear regression with variable limits | `lowLim`/`uppLim` configuration |
| `09-polynomial-degree2-with-scaler-and-lags` | Complete polynomial scenario | Scaler + lags + degree 2 |
| `10-linear-regression-with-ar` | Linear regression with autoregressive variable | `includeAr: true` |
| `11-linear-regression-static-threshold-custom` | Linear regression with custom static threshold | `staticThreshold: 100` |
#### Scenario File Structure
```json
{
"_description": "Human-readable description of the scenario",
"experimentName": "test-experiment-name",
"username": "user@example.com",
"modelName": "Linear Regression",
"targetVariable": "target_column_name",
"variableColumns": ["feature1", "feature2"],
"lagTrain": {"feature1": 0, "feature2": 0},
"lagVal": {"feature1": 0, "feature2": 0},
"remStaticWin": false,
"staticThreshold": null,
"lowLim": {},
"uppLim": {},
"window": 0,
"useScaler": false,
"includeAr": false,
"trainSize": 80,
"shuffle": true,
"lineSeparator": ",",
"decimalSeparator": ".",
"removedIntervals": [],
"degree": 1,
"interactionOnly": false,
"nanTreatment": "drop",
"startDate": null,
"endDate": null,
"scalerName": "None",
"supportFilters": {}
}
```
#### Creating New Scenarios
1. Copy an existing scenario file as a template
2. Modify parameters according to your test case
3. Save with a descriptive name: `XX-description.json`
4. Run with: `python scripts/run_training_test.py --scenario XX-description`
#### Important Validations
The training workflow enforces several business rules:
- **Polynomial Regression requires Scaler**: Models with `degree > 1` must have `useScaler: true` and a valid `scalerName` to prevent numerical overflow
- **Static Window Removal requires DatetimeIndex**: Scenarios with `remStaticWin: true` require data with a timestamp column for the `TimeSeriesDiscontinuityAnalyzer`
- **Variable Limits Consistency**: `lowLim` and `uppLim` must have matching keys, and `lowLim[key] < uppLim[key]` for all variables
## Monitoring and Metrics
The Model Manager system exposes comprehensive Prometheus metrics for operational visibility and performance monitoring:
@@ -889,6 +1048,11 @@ The Model Manager system exposes comprehensive Prometheus metrics for operationa
- Training success/failure rates through notification system
- Model save performance metrics
- Experiment status tracking
- **RCE Drift Metrics**: Reduced Coulomb Energy (RCE) for drift detection
- `silverman_radius`: Optimal bandwidth for kernel density estimation
- `rce_reference`: RCE value for reference data
- `rce_current`: RCE value for current data
- `rce_drift`: Drift score between reference and current distributions
### Cleanup Metrics
- Cleanup execution success/failure rates
@@ -972,7 +1136,7 @@ These timeouts control how long each activity in workflows can run before timing
### Code Quality & Testing
The project maintains **99%+ code coverage** with comprehensive unit and integration tests.
The project maintains **100% code coverage** with comprehensive unit and integration tests.
#### Running Tests
@@ -1073,8 +1237,8 @@ sientia-dataops-model-manager/
│ ├── sientia/ # Sientia-specific implementations
│ │ ├── __init__.py
│ │ ├── exceptions.py # Custom exceptions
│ │ ├── metrics.py # Business metrics
│ │ ├── models.py # ML model implementations
│ │ ├── metrics.py # Business metrics (includes RCE drift detection)
│ │ ├── models.py # ML model implementations (Linear & Polynomial Regression)
│ │ ├── model_serving.py # Model serving utilities
│ │ ├── reports.py # Report generation
│ │ └── utils.py # Utility functions
@@ -1085,7 +1249,7 @@ sientia-dataops-model-manager/
│ └── __init__.py
├── scripts/ # Test and utility scripts
│ ├── run_cleanup_test.py # Manual cleanup workflow test
│ └── run_training_test.py # Manual training workflow test
│ └── run_training_test.py # Training test with scenario support (--all for batch)
├── tests/ # Test suite
│ ├── activities/ # Activity tests
│ ├── workflows/ # Workflow tests
@@ -1094,6 +1258,8 @@ sientia-dataops-model-manager/
│ ├── schedules/ # Schedule tests
│ └── sientia/ # Sientia module tests
├── docs/ # Documentation and test data
│ ├── test-scenarios/ # JSON test scenario files for integration tests
│ └── test-model-data.csv # Sample CSV data for testing
├── .github/workflows/ # CI/CD workflows
│ ├── quality-gate.yml # PR quality checks
│ └── deploy.yml # Deployment workflow
@@ -1119,7 +1285,7 @@ sientia-dataops-model-manager/
### Test Coverage Guidelines
- **Minimum coverage**: 80% (enforced by CI/CD)
- **Current coverage**: 99%+ 🎯
- **Current coverage**: 100% 🎯
- **Test all branches**: Use Coverage Gutters to identify uncovered lines
- **Mock external dependencies**: Use `unittest.mock` for external services
- **Async testing**: Use `pytest-asyncio` for async activities and workflows

View File

@@ -0,0 +1,28 @@
{
"_description": "Cenário básico de regressão linear sem scaler",
"experimentName": "test-linear-regression-basic",
"username": "bruno.domingues@aignosi.com.br",
"modelName": "Linear Regression",
"targetVariable": "03CV020/CORRENTE_N_M1_PV(Value)",
"variableColumns": ["303-WIT-200(Value)"],
"lagTrain": {"303-WIT-200(Value)": 0},
"lagVal": {"303-WIT-200(Value)": 0},
"remStaticWin": false,
"lowLim": {},
"uppLim": {},
"window": 0,
"useScaler": false,
"includeAr": false,
"trainSize": 80,
"shuffle": true,
"lineSeparator": ",",
"decimalSeparator": ".",
"removedIntervals": [],
"degree": 1,
"interactionOnly": false,
"nanTreatment": "drop",
"startDate": null,
"endDate": null,
"scalerName": "None",
"supportFilters": {}
}

View File

@@ -0,0 +1,28 @@
{
"_description": "Regressão linear com Standard Scaler habilitado",
"experimentName": "test-linear-regression-scaler",
"username": "bruno.domingues@aignosi.com.br",
"modelName": "Linear Regression",
"targetVariable": "03CV020/CORRENTE_N_M1_PV(Value)",
"variableColumns": ["303-WIT-200(Value)"],
"lagTrain": {"303-WIT-200(Value)": 0},
"lagVal": {"303-WIT-200(Value)": 0},
"remStaticWin": false,
"lowLim": {},
"uppLim": {},
"window": 0,
"useScaler": true,
"includeAr": false,
"trainSize": 80,
"shuffle": true,
"lineSeparator": ",",
"decimalSeparator": ".",
"removedIntervals": [],
"degree": 1,
"interactionOnly": false,
"nanTreatment": "drop",
"startDate": null,
"endDate": null,
"scalerName": "Standard Scaler",
"supportFilters": {}
}

View File

@@ -0,0 +1,28 @@
{
"_description": "Regressão polinomial de grau 2 com scaler (obrigatório para evitar overflow)",
"experimentName": "test-polynomial-degree2",
"username": "bruno.domingues@aignosi.com.br",
"modelName": "Polynomial Regression",
"targetVariable": "03CV020/CORRENTE_N_M1_PV(Value)",
"variableColumns": ["303-WIT-200(Value)"],
"lagTrain": {"303-WIT-200(Value)": 0},
"lagVal": {"303-WIT-200(Value)": 0},
"remStaticWin": false,
"lowLim": {},
"uppLim": {},
"window": 0,
"useScaler": true,
"includeAr": false,
"trainSize": 80,
"shuffle": true,
"lineSeparator": ",",
"decimalSeparator": ".",
"removedIntervals": [],
"degree": 2,
"interactionOnly": false,
"nanTreatment": "drop",
"startDate": null,
"endDate": null,
"scalerName": "Standard Scaler",
"supportFilters": {}
}

View File

@@ -0,0 +1,28 @@
{
"_description": "Regressão polinomial de grau 3 com scaler",
"experimentName": "test-polynomial-degree3",
"username": "bruno.domingues@aignosi.com.br",
"modelName": "Polynomial Regression",
"targetVariable": "03CV020/CORRENTE_N_M1_PV(Value)",
"variableColumns": ["303-WIT-200(Value)"],
"lagTrain": {"303-WIT-200(Value)": 0},
"lagVal": {"303-WIT-200(Value)": 0},
"remStaticWin": false,
"lowLim": {},
"uppLim": {},
"window": 0,
"useScaler": true,
"includeAr": false,
"trainSize": 80,
"shuffle": true,
"lineSeparator": ",",
"decimalSeparator": ".",
"removedIntervals": [],
"degree": 3,
"interactionOnly": false,
"nanTreatment": "drop",
"startDate": null,
"endDate": null,
"scalerName": "Standard Scaler",
"supportFilters": {}
}

View File

@@ -0,0 +1,28 @@
{
"_description": "Regressão linear com lags de treino e validação",
"experimentName": "test-linear-with-lags",
"username": "bruno.domingues@aignosi.com.br",
"modelName": "Linear Regression",
"targetVariable": "03CV020/CORRENTE_N_M1_PV(Value)",
"variableColumns": ["303-WIT-200(Value)"],
"lagTrain": {"303-WIT-200(Value)": 5},
"lagVal": {"303-WIT-200(Value)": 3},
"remStaticWin": false,
"lowLim": {},
"uppLim": {},
"window": 0,
"useScaler": false,
"includeAr": false,
"trainSize": 80,
"shuffle": true,
"lineSeparator": ",",
"decimalSeparator": ".",
"removedIntervals": [],
"degree": 1,
"interactionOnly": false,
"nanTreatment": "drop",
"startDate": null,
"endDate": null,
"scalerName": "None",
"supportFilters": {}
}

View File

@@ -0,0 +1,28 @@
{
"_description": "Regressão linear com tratamento de NaN por interpolação linear",
"experimentName": "test-linear-nan-interpolation",
"username": "bruno.domingues@aignosi.com.br",
"modelName": "Linear Regression",
"targetVariable": "03CV020/CORRENTE_N_M1_PV(Value)",
"variableColumns": ["303-WIT-200(Value)"],
"lagTrain": {"303-WIT-200(Value)": 0},
"lagVal": {"303-WIT-200(Value)": 0},
"remStaticWin": false,
"lowLim": {},
"uppLim": {},
"window": 0,
"useScaler": false,
"includeAr": false,
"trainSize": 80,
"shuffle": true,
"lineSeparator": ",",
"decimalSeparator": ".",
"removedIntervals": [],
"degree": 1,
"interactionOnly": false,
"nanTreatment": "linear interpolation",
"startDate": null,
"endDate": null,
"scalerName": "None",
"supportFilters": {}
}

View File

@@ -0,0 +1,28 @@
{
"_description": "Regressão linear com remoção de janelas estáticas",
"experimentName": "test-linear-static-removal",
"username": "bruno.domingues@aignosi.com.br",
"modelName": "Linear Regression",
"targetVariable": "03CV020/CORRENTE_N_M1_PV(Value)",
"variableColumns": ["303-WIT-200(Value)"],
"lagTrain": {"303-WIT-200(Value)": 0},
"lagVal": {"303-WIT-200(Value)": 0},
"remStaticWin": true,
"lowLim": {},
"uppLim": {},
"window": 10,
"useScaler": false,
"includeAr": false,
"trainSize": 80,
"shuffle": true,
"lineSeparator": ",",
"decimalSeparator": ".",
"removedIntervals": [],
"degree": 1,
"interactionOnly": false,
"nanTreatment": "drop",
"startDate": null,
"endDate": null,
"scalerName": "None",
"supportFilters": {}
}

View File

@@ -0,0 +1,28 @@
{
"_description": "Regressão linear com limites inferior e superior para variáveis",
"experimentName": "test-linear-with-limits",
"username": "bruno.domingues@aignosi.com.br",
"modelName": "Linear Regression",
"targetVariable": "03CV020/CORRENTE_N_M1_PV(Value)",
"variableColumns": ["303-WIT-200(Value)"],
"lagTrain": {"303-WIT-200(Value)": 0},
"lagVal": {"303-WIT-200(Value)": 0},
"remStaticWin": false,
"lowLim": {"303-WIT-200(Value)": 0.0},
"uppLim": {"303-WIT-200(Value)": 1000.0},
"window": 0,
"useScaler": false,
"includeAr": false,
"trainSize": 80,
"shuffle": true,
"lineSeparator": ",",
"decimalSeparator": ".",
"removedIntervals": [],
"degree": 1,
"interactionOnly": false,
"nanTreatment": "drop",
"startDate": null,
"endDate": null,
"scalerName": "None",
"supportFilters": {}
}

View File

@@ -0,0 +1,28 @@
{
"_description": "Cenário completo: regressão polinomial grau 2 com scaler e lags",
"experimentName": "test-polynomial-complete",
"username": "bruno.domingues@aignosi.com.br",
"modelName": "Polynomial Regression",
"targetVariable": "03CV020/CORRENTE_N_M1_PV(Value)",
"variableColumns": ["303-WIT-200(Value)"],
"lagTrain": {"303-WIT-200(Value)": 3},
"lagVal": {"303-WIT-200(Value)": 2},
"remStaticWin": false,
"lowLim": {},
"uppLim": {},
"window": 0,
"useScaler": true,
"includeAr": false,
"trainSize": 80,
"shuffle": true,
"lineSeparator": ",",
"decimalSeparator": ".",
"removedIntervals": [],
"degree": 2,
"interactionOnly": false,
"nanTreatment": "drop",
"startDate": null,
"endDate": null,
"scalerName": "Standard Scaler",
"supportFilters": {}
}

View File

@@ -0,0 +1,28 @@
{
"_description": "Regressão linear com variável autoregressiva (AR)",
"experimentName": "test-linear-with-ar",
"username": "bruno.domingues@aignosi.com.br",
"modelName": "Linear Regression",
"targetVariable": "03CV020/CORRENTE_N_M1_PV(Value)",
"variableColumns": ["303-WIT-200(Value)"],
"lagTrain": {"303-WIT-200(Value)": 0},
"lagVal": {"303-WIT-200(Value)": 0},
"remStaticWin": false,
"lowLim": {},
"uppLim": {},
"window": 0,
"useScaler": false,
"includeAr": true,
"trainSize": 80,
"shuffle": true,
"lineSeparator": ",",
"decimalSeparator": ".",
"removedIntervals": [],
"degree": 1,
"interactionOnly": false,
"nanTreatment": "drop",
"startDate": null,
"endDate": null,
"scalerName": "None",
"supportFilters": {}
}

View File

@@ -0,0 +1,29 @@
{
"_description": "Regressão linear com remoção de janelas estáticas e static_threshold customizado",
"experimentName": "test-linear-static-threshold",
"username": "bruno.domingues@aignosi.com.br",
"modelName": "Linear Regression",
"targetVariable": "03CV020/CORRENTE_N_M1_PV(Value)",
"variableColumns": ["303-WIT-200(Value)"],
"lagTrain": {"303-WIT-200(Value)": 0},
"lagVal": {"303-WIT-200(Value)": 0},
"remStaticWin": true,
"staticThreshold": 100,
"lowLim": {},
"uppLim": {},
"window": 10,
"useScaler": false,
"includeAr": false,
"trainSize": 80,
"shuffle": true,
"lineSeparator": ",",
"decimalSeparator": ".",
"removedIntervals": [],
"degree": 1,
"interactionOnly": false,
"nanTreatment": "drop",
"startDate": null,
"endDate": null,
"scalerName": "None",
"supportFilters": {}
}

View File

@@ -26,3 +26,123 @@ def r2(real_data: pd.Series, predictions: pd.Series) -> float:
Calculates the R2 score between the real data and the predictions.
"""
return round(r2_score(real_data.astype(np.float64), predictions.astype(np.float64)), 2)
def silverman_radius(data: np.ndarray) -> float:
"""
Calculate the Silverman bandwidth (radius) for a given dataset.
Args:
data (np.ndarray): Input data (1D array)
Returns:
float: Silverman bandwidth (radius)
"""
n = len(data)
sigma = np.std(data)
iqr = np.percentile(data, 75) - np.percentile(data, 25)
radius = 0.9 * min(sigma, iqr / 1.34) * n ** (-1 / 5)
return radius
def rce_train(training_set: pd.DataFrame, radius: float | None = None) -> pd.DataFrame:
"""
Get the Reduced Coulomb Energy (RCE) prototypes.
Args:
training_set (pd.DataFrame): The training set
radius (float | None): The radius of the RCE prototypes. If None, computed using Silverman's rule.
Returns:
pd.DataFrame: The RCE prototypes
"""
train_vectors = training_set.values
# Vectorized distance computation for the radius calculation
diff_vectors = train_vectors[:, np.newaxis] - train_vectors[np.newaxis, :]
distances = np.linalg.norm(diff_vectors, axis=-1)
# Non-parametric radius: Silverman Radius (compute if not provided)
effective_radius = radius if radius is not None else silverman_radius(distances.flatten())
# Initialize prototypes with the first vector
prototypes = [train_vectors[0]]
for vector in train_vectors[1:]:
# Vectorized distance check between current vector and all prototypes
distances_to_prototypes = np.linalg.norm(prototypes - vector, axis=1)
# If no prototype is close, add the current vector as a new prototype
if np.all(distances_to_prototypes > effective_radius):
prototypes.append(vector)
return pd.DataFrame(prototypes)
def rce_test(test_set: pd.DataFrame, prototypes: pd.DataFrame) -> pd.Series:
"""
Get the signed Reduced Coulomb Energy (RCE) predictions.
Args:
test_set (pd.DataFrame): The test set
prototypes (pd.DataFrame): The RCE prototypes
Returns:
pd.Series: The signed distances to the closest prototype for each test vector
"""
test_vectors = test_set.values
prototype_vectors = prototypes.values
# Vectorized computation of distances between test vectors and all prototypes
diff_vectors = test_vectors[:, np.newaxis] - prototype_vectors[np.newaxis, :]
distances = np.linalg.norm(diff_vectors, axis=-1)
# Find the closest prototype for each test vector
min_distances = np.min(distances, axis=1)
closest_prototypes = prototype_vectors[np.argmin(distances, axis=1)]
# Compute the signed distance for each test vector
signed_distances = np.sqrt(min_distances**2) * np.sign(
np.mean(test_vectors - closest_prototypes, axis=1)
)
return pd.Series(signed_distances)
def rce_drift(reference_data: pd.DataFrame, real_data: pd.DataFrame, column: str) -> pd.Series:
"""
Detect drift using the Reduced Coulomb Energy (RCE) method.
Args:
reference_data (pd.DataFrame): The reference data
real_data (pd.DataFrame): The real data
column (str): The target column to be analyzed. 'target' or 'prediction'
Returns:
pd.Series: Normalized drift distances
"""
common_columns = list(set(reference_data.columns).intersection(real_data.columns))
reference_data = reference_data[common_columns]
real_data = real_data[common_columns]
# Get prototypes
if column == 'target':
prototypes = rce_train(reference_data.drop(columns=['prediction']), 0.1)
else:
prototypes = rce_train(reference_data.drop(columns=['target']), 0.1)
# Distances to prototypes
if column == 'target':
distances_train = rce_test(reference_data.drop(columns=['prediction']), prototypes)
distances_test = rce_test(real_data.drop(columns=['prediction']), prototypes)
else:
distances_train = rce_test(reference_data.drop(columns=['target']), prototypes)
distances_test = rce_test(real_data.drop(columns=['target']), prototypes)
# Find the maximum absolute distance in the training set
max_abs_distance = max(abs(distances_train.max()), abs(distances_train.min()))
# Normalize while preserving sign
distances = distances_test / max_abs_distance
return distances

View File

@@ -6,19 +6,24 @@ from sientia_do.operations.df_preprocessor import create_features, limit_dataset
from sientia_do.timeseries.analyzer import TimeSeriesDiscontinuityAnalyzer
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.linear_model import LinearRegression
from sklearn.preprocessing import StandardScaler
from sklearn.preprocessing import PolynomialFeatures, StandardScaler
DISCONTINUITY_TREATMENT = 'Discontinuity Treatment'
LAG_SELECTION = 'Lag Selection'
RANGE_SELECTION = 'Range Selection & Data Removal'
STATIC_WINDOW_REMOVAL = 'Static Window Removal'
DEFINE_VARIABLES_LIMITS = 'Define Variables Limits'
NORMALIZATION = 'Normalization'
FEATURE_CREATION = 'Feature Creation'
LAG_CREATION = 'Lag Creation'
class LinearRegressionModel(BaseEstimator, TransformerMixin):
"""
Linear Regression Model for Time Series Analysis.
Supports both simple linear regression and polynomial regression.
Thread-safety: This class is NOT thread-safe during fit() operations.
Do not call fit() on the same instance from multiple threads simultaneously.
After fitting, predict() is thread-safe for read-only operations.
@@ -36,6 +41,8 @@ class LinearRegressionModel(BaseEstimator, TransformerMixin):
model_params: dict[str, Any] | None = None,
clipping: dict[str, float] | None = None,
weights: dict[str, float] | None = None,
degree: int = 1,
interaction_only: bool = False,
):
"""
Linear Regression Model for Time Series Analysis
@@ -48,6 +55,8 @@ class LinearRegressionModel(BaseEstimator, TransformerMixin):
*Format: {'min': min_value, 'max': max_value}*
weights (dict): The weights for the Linear Regression model \\
*Format: {'variable_name': weight}*
degree (int): The degree of the polynomial features (1 = linear, >1 = polynomial)
interaction_only (bool): If True, only interaction features are produced
Returns:
LinearRegressionModel: The prediction model object
@@ -60,6 +69,43 @@ class LinearRegressionModel(BaseEstimator, TransformerMixin):
self.q1_target: float | None = None
self.q3_target: float | None = None
self.weights: dict[str, float] | None = weights
self.degree: int = degree
self.interaction_only: bool = interaction_only
self.poly: PolynomialFeatures | None = None
self.poly_feature_names: list[str] | None = None
def create_poly_features(self, input_data: pd.DataFrame, fit: bool = False) -> pd.DataFrame:
"""
Create polynomial features from input data.
Args:
input_data (pd.DataFrame): Input data with feature columns
fit (bool): If True, fit the PolynomialFeatures transformer
Returns:
pd.DataFrame: DataFrame with polynomial features
"""
if self.degree <= 1:
return input_data
if fit:
self.poly = PolynomialFeatures(
degree=self.degree,
interaction_only=self.interaction_only,
include_bias=False,
)
poly_features = self.poly.fit_transform(input_data)
self.poly_feature_names = list(self.poly.get_feature_names_out(input_data.columns))
else:
if self.poly is None:
raise ValueError('PolynomialFeatures not fitted. Call fit() first.')
poly_features = self.poly.transform(input_data)
return pd.DataFrame(
poly_features,
columns=self.poly_feature_names,
index=input_data.index,
)
def fit(self, input_data: pd.DataFrame) -> 'LinearRegressionModel':
"""
@@ -71,13 +117,48 @@ class LinearRegressionModel(BaseEstimator, TransformerMixin):
Returns:
LinearRegressionModel: The prediction model object
"""
assert self.variable_columns is not None, 'variable_columns must be set before fitting'
X_train = input_data[self.variable_columns]
y_train = input_data[self.target_variable]
if not self.target_variable:
raise ValueError('target_variable must be set before fitting')
# Infer variable_columns if not provided
if self.variable_columns is None:
self.variable_columns = [
col for col in input_data.columns if col != self.target_variable
]
# Validate columns exist
missing_cols = [col for col in self.variable_columns if col not in input_data.columns]
if missing_cols:
raise ValueError(f'Columns not found in input data: {missing_cols}')
if self.target_variable not in input_data.columns:
raise ValueError(f'Target variable {self.target_variable} not found in input data')
X_train = input_data[self.variable_columns].copy()
y_train = input_data[self.target_variable].copy()
# Handle infinite values
X_train = X_train.replace([np.inf, -np.inf], np.nan)
y_train = y_train.replace([np.inf, -np.inf], np.nan)
# Remove rows with NaN
valid_mask = ~(X_train.isna().any(axis=1) | y_train.isna())
X_train = X_train[valid_mask]
y_train = y_train[valid_mask]
# Remove columns with all NaN values
cols_to_drop = X_train.columns[X_train.isna().all()].tolist()
if cols_to_drop: # pragma: no cover
X_train = X_train.drop(columns=cols_to_drop)
self.variable_columns = [c for c in self.variable_columns if c not in cols_to_drop]
self.q1_target = y_train.quantile(0.25)
self.q3_target = y_train.quantile(0.75)
# Apply polynomial features if degree > 1
if self.degree > 1:
X_train = self.create_poly_features(X_train, fit=True)
# Fit the model
self.regr.fit(X_train, y_train)
@@ -86,7 +167,9 @@ class LinearRegressionModel(BaseEstimator, TransformerMixin):
round_intercept = np.round(self.regr.intercept_, 3)
# Save the weights
weights = dict(zip(self.variable_columns, [float(c) for c in round_coef], strict=True))
feature_names = self.poly_feature_names if self.degree > 1 else self.variable_columns
assert feature_names is not None, 'feature_names should be set at this point'
weights = dict(zip(feature_names, [float(c) for c in round_coef], strict=True))
weights = dict(sorted(weights.items(), key=lambda item: abs(item[1]), reverse=True))
weights = {'Bias': float(round_intercept), **weights}
self.weights = weights
@@ -104,7 +187,16 @@ class LinearRegressionModel(BaseEstimator, TransformerMixin):
Returns:
numpy.ndarray: The predicted target variable
"""
X_test = input_data[self.variable_columns]
assert self.variable_columns is not None, 'variable_columns must be set before predict'
X_test: pd.DataFrame = input_data[self.variable_columns].copy()
# Handle infinite values
X_test = X_test.replace([np.inf, -np.inf], np.nan)
# Apply polynomial features if degree > 1
if self.degree > 1:
X_test = self.create_poly_features(X_test, fit=False)
y_pred = self.regr.predict(X_test)
if self.clipping:
@@ -116,6 +208,15 @@ class LinearRegressionModel(BaseEstimator, TransformerMixin):
return y_pred
def get_regressor(self) -> LinearRegression:
"""
Get the underlying LinearRegression model.
Returns:
LinearRegression: The sklearn LinearRegression model
"""
return self.regr
class DataPreprocessor(BaseEstimator, TransformerMixin):
"""
@@ -141,6 +242,9 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
nan_treatment: str | None = None,
lag_train: dict[str, int] | None = None,
lag_transform: dict[str, int] | None = None,
start_date: str | None = None,
end_date: str | None = None,
removed_intervals: list[tuple[str, str]] | None = None,
static_threshold: int | None = None,
low_lim: dict[str, float] | None = None,
upp_lim: dict[str, float] | None = None,
@@ -161,11 +265,15 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
target_variable (str): The target variable name
input_columns (list): The input columns names in a list
nan_treatment (str): The treatment for missing values \\
*Options: 'drop', 'fill linear'*
*Options: 'drop', 'fill linear', 'linear interpolation'*
lag_train (dict): The lags for each variable to be applyed during training \\
*Format: {'variable_name': lag}*
lag_transform (dict): The lags for each variable to be applyed during transformation \\
*Format: {'variable_name': lag}*
start_date (str): The start date for filtering data (format: 'YYYY-MM-DD HH:MM:SS')
end_date (str): The end date for filtering data (format: 'YYYY-MM-DD HH:MM:SS')
removed_intervals (list): List of tuples with intervals to remove from data \\
*Format: [('start_date', 'end_date'), ...]*
static_threshold (int): The number of repeated values to be considered as static
low_lim (dict): The lower limits for each variable \\
*Format: {'variable_name': limit}*
@@ -188,6 +296,7 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
steps_order (list): The order of the steps to be executed in the pipeline \\
*Options for list: 'Discontinuity Treatment',
'Lag Selection',
'Range Selection & Data Removal',
'Static Window Removal',
'Define Variables Limits',
'Normalization',
@@ -203,6 +312,9 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
self.nan_treatment = nan_treatment
self.lag_train = lag_train if lag_train else {}
self.lag_transform = lag_transform if lag_transform else {}
self.start_date = start_date
self.end_date = end_date
self.removed_intervals = removed_intervals if removed_intervals else []
self.ar_var = ar_var
self.self_operations = self_operations
self.cross_operations = cross_operations
@@ -214,6 +326,7 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
self.scaler_name = scaler_name
self.scaler_params = scaler_params
self.feature_names_order: list[str] = [] # Initialize to avoid AttributeError
self._fitted_feature_order: list[str] | None = None # Track feature order after fit
if self.scaler_name == 'Standard Scaler':
self.scaler = StandardScaler()
@@ -226,11 +339,12 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
possible_steps = [
DISCONTINUITY_TREATMENT,
LAG_SELECTION,
RANGE_SELECTION,
STATIC_WINDOW_REMOVAL,
DEFINE_VARIABLES_LIMITS,
NORMALIZATION,
'Feature Creation',
'Lag Creation',
FEATURE_CREATION,
LAG_CREATION,
]
self.steps_order = steps_order or possible_steps
for step in possible_steps:
@@ -321,7 +435,61 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
pandas.DataFrame: The treated data
"""
if self.nan_treatment:
input_data = treat_nan(input_data, self.nan_treatment)
# Map 'linear interpolation' to 'fill linear' for compatibility
treatment = self.nan_treatment
if treatment == 'linear interpolation':
treatment = 'fill linear'
input_data = treat_nan(input_data, treatment)
return input_data
def _parse_datetime(self, date_str: str | None) -> pd.Timestamp | None:
"""Parse a date string to Timestamp, returning None on failure."""
if not date_str:
return None
try:
return pd.to_datetime(date_str)
except (ValueError, TypeError):
return None
def _filter_by_date_range(
self, input_data: pd.DataFrame, start: pd.Timestamp | None, end: pd.Timestamp | None
) -> pd.DataFrame:
"""Filter DataFrame by start and end dates."""
if start is not None:
input_data = input_data[input_data.index >= start]
if end is not None:
input_data = input_data[input_data.index <= end]
return input_data
def _remove_interval(self, input_data: pd.DataFrame, interval: tuple | list) -> pd.DataFrame:
"""Remove a single interval from the DataFrame."""
if len(interval) < 2:
return input_data
interval_start = self._parse_datetime(interval[0])
interval_end = self._parse_datetime(interval[1])
if interval_start is None or interval_end is None:
return input_data
mask = ~((input_data.index >= interval_start) & (input_data.index <= interval_end))
return input_data[mask]
def range_selection(self, input_data: pd.DataFrame) -> pd.DataFrame:
"""
Filter data by date range and remove specified intervals.
Args:
input_data (pandas.DataFrame): The input data with datetime index
Returns:
pandas.DataFrame: The filtered data
"""
start = self._parse_datetime(self.start_date)
end = self._parse_datetime(self.end_date)
input_data = self._filter_by_date_range(input_data, start, end)
if self.removed_intervals:
for interval in self.removed_intervals:
input_data = self._remove_interval(input_data, interval)
return input_data
def lag_selection(self, input_data: pd.DataFrame, lag_dict: dict) -> pd.DataFrame:
@@ -469,6 +637,10 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
if step == LAG_SELECTION:
data_treat = self.lag_selection(data_treat, self.lag_train)
# Range Selection & Data Removal
if step == RANGE_SELECTION:
data_treat = self.range_selection(data_treat)
# Static Window Treatment
if step == STATIC_WINDOW_REMOVAL:
data_treat = self.treat_static_windows(data_treat)
@@ -493,6 +665,9 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
'variance': round(variance, 3),
}
# Store fitted feature order for predict method
self._fitted_feature_order = list(existing_columns)
return self
def transform(self, x: pd.DataFrame) -> pd.DataFrame:
@@ -525,6 +700,10 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
if step == LAG_SELECTION:
data_treat = self.lag_selection(data_treat, self.lag_transform)
# Range Selection & Data Removal (typically skipped in transform)
if step == RANGE_SELECTION:
data_treat = self.range_selection(data_treat)
# Static Window Treatment
if step == STATIC_WINDOW_REMOVAL:
data_treat = self.treat_static_windows(data_treat)
@@ -540,11 +719,11 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
data_treat[feature_cols] = self.scaler.transform(data_treat[feature_cols])
# Feature Creation
if step == 'Feature Creation':
if step == FEATURE_CREATION:
data_treat = self.create_features(data_treat)
# Lag Creation
if step == 'Lag Creation':
if step == LAG_CREATION:
# Autoregressive Variable
if self.input_columns is not None and self.ar_var in self.input_columns:
data_treat = self.create_ar(data_treat)
@@ -553,3 +732,33 @@ class DataPreprocessor(BaseEstimator, TransformerMixin):
data_treat = self.create_lags(data_treat)
return data_treat
def predict(self, x: pd.DataFrame) -> pd.DataFrame:
"""
Transform data for prediction (removes target variable).
This method is a wrapper around transform() that:
1. Transforms the input data
2. Removes the target variable column
3. Ensures features are in the same order as during fit
Args:
x (pandas.DataFrame): The input data
Returns:
pandas.DataFrame: The transformed data without target variable,
with features in the same order as during fit
"""
data_treat = self.transform(x)
# Remove target variable if present
if self.target_variable in data_treat.columns: # pragma: no branch
data_treat = data_treat.drop(columns=self.target_variable)
# Ensure features are in the same order as during fit
if self._fitted_feature_order is not None:
# Filter to only include columns that exist in both
available_cols = [c for c in self._fitted_feature_order if c in data_treat.columns]
data_treat = data_treat[available_cols]
return data_treat

View File

@@ -36,10 +36,8 @@ def load_html_from_file(file_path):
with open(file_path, encoding='utf-8') as file:
return file.read()
except FileNotFoundError:
print(f'File not found: {file_path}')
return None
except OSError as e: # noqa: BLE001
print(f'Error reading file: {e}')
except OSError: # noqa: BLE001
return None
@@ -49,8 +47,6 @@ def inject_content(main_html, section_id, content):
if section:
section.clear()
section.append(BeautifulSoup(content, 'html.parser'))
else:
print(f"Section with id '{section_id}' not found in the main HTML template.")
return str(soup)

View File

@@ -1,6 +1,10 @@
from dataclasses import dataclass
from typing import Any
# Model name constants
MODEL_LINEAR_REGRESSION = 'Linear Regression'
MODEL_POLYNOMIAL_REGRESSION = 'Polynomial Regression'
@dataclass
class TrainModelParams:
@@ -17,8 +21,8 @@ class TrainModelParams:
Attributes:
variable_columns (list[str]): List of variable column names to use as features.
lag_train (int): Number of lags to apply during training phase.
lag_val (int): Number of lags to apply during validation phase.
lag_train (dict[str, int]): Dictionary of lags per variable for training phase.
lag_val (dict[str, int]): Dictionary of lags per variable for validation phase.
target_variable (str): Name of the target variable to predict.
rem_static_win (bool): Whether to remove static windows from data.
low_lim (dict[str, float]): Dictionary of lower limits for each variable.
@@ -35,11 +39,20 @@ class TrainModelParams:
experiment_run_id (int): Unique identifier for the experiment run.
experiment_name (str): Name of the experiment for tracking.
removed_intervals (list): List of time intervals to remove from the data.
model_name (str): Name of the model type ('Linear Regression' or 'Polynomial Regression').
degree (int): Degree of polynomial features (1 for linear, >1 for polynomial).
interaction_only (bool): If True, only interaction features are produced for polynomial.
nan_treatment (str): Treatment for NaN values ('drop' or 'linear interpolation').
start_date (str | None): Start date for filtering data.
end_date (str | None): End date for filtering data.
scaler_name (str): Name of the scaler to use ('Standard Scaler' or 'None').
support_filters (dict): Custom support filters per variable.
static_threshold (int | None): Threshold for static window removal (1-1000). Only used when rem_static_win is True.
"""
variable_columns: list[str]
lag_train: int
lag_val: int
lag_train: dict[str, int]
lag_val: dict[str, int]
target_variable: str
rem_static_win: bool
low_lim: dict[str, float]
@@ -56,6 +69,15 @@ class TrainModelParams:
experiment_run_id: int
experiment_name: str
removed_intervals: list
model_name: str
degree: int
interaction_only: bool
nan_treatment: str
start_date: str | None
end_date: str | None
scaler_name: str
support_filters: dict
static_threshold: int | None
@classmethod
def from_dict(cls, data: dict[str, Any]) -> 'TrainModelParams':
@@ -83,8 +105,8 @@ class TrainModelParams:
variable_columns=cls._check_none(
data.get('variable_columns'), list, 'variable_columns'
),
lag_train=cls._check_none(data.get('lag_train'), int, 'lag_train'),
lag_val=cls._check_none(data.get('lag_val'), int, 'lag_val'),
lag_train=cls._check_none(data.get('lag_train'), dict, 'lag_train'),
lag_val=cls._check_none(data.get('lag_val'), dict, 'lag_val'),
target_variable=cls._check_none(data.get('target_variable'), str, 'target_variable'),
rem_static_win=cls._check_none(data.get('rem_static_win'), bool, 'rem_static_win'),
low_lim=cls._check_none(data.get('low_lim'), dict, 'low_lim'),
@@ -107,6 +129,18 @@ class TrainModelParams:
removed_intervals=cls._check_type(
data.get('removed_intervals'), list, 'removed_intervals'
),
model_name=cls._check_none(data.get('model_name'), str, 'model_name'),
degree=cls._check_none(data.get('degree'), int, 'degree'),
interaction_only=cls._check_none(
data.get('interaction_only'), bool, 'interaction_only'
),
nan_treatment=cls._check_none(data.get('nan_treatment'), str, 'nan_treatment'),
start_date=cls._check_type(data.get('start_date'), str, 'start_date'),
end_date=cls._check_type(data.get('end_date'), str, 'end_date'),
scaler_name=cls._check_none(data.get('scaler_name'), str, 'scaler_name'),
support_filters=cls._check_type(data.get('support_filters'), dict, 'support_filters')
or {},
static_threshold=cls._check_type(data.get('static_threshold'), int, 'static_threshold'),
)
@staticmethod
@@ -172,25 +206,94 @@ class TrainModelParams:
Raises:
ValueError: If any business rule is violated
"""
# Validate train_size range (10-100%)
self._validate_numeric_ranges()
self._validate_model_params()
self._validate_intervals_and_dates()
self._validate_limits()
self._validate_required_strings()
def _validate_numeric_ranges(self) -> None:
"""Validate numeric parameters are within acceptable ranges."""
if not 10 <= self.train_size <= 100:
raise ValueError(f'train_size must be between 10 and 100, got {self.train_size}')
# Validate variable_columns is not empty
if not self.variable_columns:
raise ValueError('variable_columns cannot be empty')
# Validate positive integers
if self.lag_train < 0:
raise ValueError(f'lag_train must be positive, got {self.lag_train}')
for var, lag in self.lag_train.items():
if lag < 0:
raise ValueError(f'lag_train for {var} must be non-negative, got {lag}')
if self.lag_val < 0:
raise ValueError(f'lag_val must be positive, got {self.lag_val}')
for var, lag in self.lag_val.items():
if lag < 0:
raise ValueError(f'lag_val for {var} must be non-negative, got {lag}')
if self.window < 0:
raise ValueError(f'window must be positive, got {self.window}')
raise ValueError(f'window must be non-negative, got {self.window}')
# Validate low_lim and upp_lim consistency
# Validate static_threshold only when rem_static_win is True and value is provided
if self.rem_static_win and self.static_threshold is not None:
if not 1 <= self.static_threshold <= 1000:
raise ValueError(
f'static_threshold must be between 1 and 1000, got {self.static_threshold}'
)
def _validate_model_params(self) -> None:
"""Validate model-related parameters."""
if self.degree < 1:
raise ValueError(f'degree must be at least 1, got {self.degree}')
valid_nan_treatments = ['drop', 'linear interpolation', 'fill linear']
if self.nan_treatment not in valid_nan_treatments:
raise ValueError(
f'nan_treatment must be one of {valid_nan_treatments}, got {self.nan_treatment}'
)
valid_scalers = ['Standard Scaler', 'None']
if self.scaler_name not in valid_scalers:
raise ValueError(f'scaler_name must be one of {valid_scalers}, got {self.scaler_name}')
valid_models = [MODEL_LINEAR_REGRESSION, MODEL_POLYNOMIAL_REGRESSION]
if self.model_name not in valid_models:
raise ValueError(f'model_name must be one of {valid_models}, got {self.model_name}')
if self.model_name == MODEL_POLYNOMIAL_REGRESSION and self.degree < 2:
raise ValueError(
f'degree must be at least 2 for {MODEL_POLYNOMIAL_REGRESSION}, got {self.degree}'
)
if self.model_name == MODEL_POLYNOMIAL_REGRESSION and self.scaler_name == 'None':
raise ValueError(
f'scaler_name must be set (e.g., "Standard Scaler") for {MODEL_POLYNOMIAL_REGRESSION} '
'to avoid numerical overflow with large feature values'
)
if self.model_name == MODEL_LINEAR_REGRESSION and self.degree != 1:
raise ValueError(f'degree must be 1 for {MODEL_LINEAR_REGRESSION}, got {self.degree}')
def _validate_intervals_and_dates(self) -> None:
"""Validate removed_intervals format and date parameters."""
if self.removed_intervals:
for i, interval in enumerate(self.removed_intervals):
if not isinstance(interval, (list, tuple)):
raise ValueError(
f'removed_intervals[{i}] must be a list or tuple, '
f'got {type(interval).__name__}'
)
if len(interval) < 2:
raise ValueError(
f'removed_intervals[{i}] must have at least 2 elements (start, end), '
f'got {len(interval)}'
)
if self.start_date is not None and not isinstance(self.start_date, str):
raise TypeError(f'start_date must be a string, got {type(self.start_date).__name__}')
if self.end_date is not None and not isinstance(self.end_date, str):
raise TypeError(f'end_date must be a string, got {type(self.end_date).__name__}')
def _validate_limits(self) -> None:
"""Validate low_lim and upp_lim consistency."""
if set(self.low_lim.keys()) != set(self.upp_lim.keys()):
raise ValueError(
f'low_lim and upp_lim must have the same keys. '
@@ -198,7 +301,6 @@ class TrainModelParams:
f'upp_lim keys: {set(self.upp_lim.keys())}'
)
# Validate that low_lim < upp_lim for each variable
for var in self.low_lim:
if self.low_lim[var] >= self.upp_lim[var]:
raise ValueError(
@@ -206,13 +308,16 @@ class TrainModelParams:
f'Got low_lim={self.low_lim[var]}, upp_lim={self.upp_lim[var]}'
)
# Validate bucket_name and file_name are not empty
def _validate_required_strings(self) -> None:
"""Validate required string fields are not empty."""
if not self.target_variable.strip():
raise ValueError('target_variable cannot be empty or whitespace')
if not self.bucket_name.strip():
raise ValueError('bucket_name cannot be empty or whitespace')
if not self.file_name.strip():
raise ValueError('file_name cannot be empty or whitespace')
# Validate experiment_name is not empty
if not self.experiment_name.strip():
raise ValueError('experiment_name cannot be empty or whitespace')

View File

@@ -25,6 +25,7 @@ class TrainModelResult:
regr (LinearRegressionModel): The trained linear regression model.
scaler_dict (dict): A dictionary containing the scalers used to scale the features and target values.
y_pred (pd.Series | None): The predicted target values for the testing dataset. Default is None.
y_train_pred (pd.Series | None): The predicted target values for the training dataset. Default is None.
mse_val (float | None): The Mean Squared Error (MSE) of the predictions. Default is None.
mae_val (float | None): The Mean Absolute Error (MAE) of the predictions. Default is None.
r2_val (float | None): The R-squared (R²) value of the predictions. Default is None.
@@ -46,6 +47,7 @@ class TrainModelResult:
regr: LinearRegressionModel
scaler_dict: dict
y_pred: pd.Series | None = None
y_train_pred: pd.Series | None = None
mse_val: float | None = None
mae_val: float | None = None
r2_val: float | None = None

View File

@@ -183,8 +183,6 @@ class ModelRepository:
raise ValueError(error_msg)
# Prepare parameters
train_test_split = f'{data.params.train_size}-{100 - data.params.train_size}'
interval_strs = [
(str(interval[0]), str(interval[1]))
for interval in (data.params.removed_intervals or [])
@@ -197,19 +195,34 @@ class ModelRepository:
run_name=data.run_name, description=data.params.experiment_name
):
# Log model parameters
self.model_serving.log_param('model_type', 'Linear Regression')
self.model_serving.log_param('model_name', data.params.model_name)
self.model_serving.log_param(
'models_params',
{'degree': data.params.degree, 'interaction_only': data.params.interaction_only},
)
self.model_serving.log_param('target_variable', data.params.target_variable)
self.model_serving.log_param('input_variables', data.params.variable_columns)
self.model_serving.log_param('nan_treatment', data.params.nan_treatment)
self.model_serving.log_param('lag_train', data.params.lag_train)
self.model_serving.log_param('lag_val', data.params.lag_val)
self.model_serving.log_param('ma', data.params.window)
self.model_serving.log_param('low_lim', data.params.low_lim)
self.model_serving.log_param('upp_lim', data.params.upp_lim)
self.model_serving.log_param('normalized', data.scaler_dict)
self.model_serving.log_param('ar', data.params.include_ar)
self.model_serving.log_param('Train_test_split', train_test_split)
self.model_serving.log_param('Removed_intervals', interval_strs)
self.model_serving.log_param('Retrain', False)
self.model_serving.log_param('lag_transform', data.params.lag_val)
static_threshold_value = None
if data.params.rem_static_win:
static_threshold_value = (
data.params.static_threshold if data.params.static_threshold is not None else 1
)
self.model_serving.log_param('static_threshold', static_threshold_value)
self.model_serving.log_param('lower_limits', data.params.low_lim)
self.model_serving.log_param('upper_limits', data.params.upp_lim)
self.model_serving.log_param('scaler_name', data.params.scaler_name)
self.model_serving.log_param('scaler_params', data.scaler_dict)
self.model_serving.log_param('include_ar', data.params.include_ar)
self.model_serving.log_param('train_size', round(data.params.train_size / 100, 2))
self.model_serving.log_param('test_size', round(1 - (data.params.train_size / 100), 2))
self.model_serving.log_param('start_date', data.params.start_date)
self.model_serving.log_param('end_date', data.params.end_date)
self.model_serving.log_param('removed_intervals', interval_strs)
self.model_serving.log_param('retrain', False)
self.model_serving.log_param('support_filters', data.params.support_filters)
# Log evaluation metrics
self.model_serving.log_metric('MSE', data.mse_val)
@@ -274,10 +287,16 @@ class ModelRepository:
self.logger.error(error_msg)
raise ValueError(error_msg)
if data.y_train_pred is None:
error_msg = 'Training predictions (y_train_pred) are None'
self.logger.error(error_msg)
raise ValueError(error_msg)
# Prepare reference data (training set)
reference_data = pd.concat([data.x_train, data.y_train], axis=1)
reference_data = reference_data.rename(columns={data.params.target_variable: 'target'})
reference_data['prediction'] = data.regr.predict(data.x_train)
# Use pre-calculated predictions (calculated before denormalization to avoid overflow)
reference_data['prediction'] = data.y_train_pred
# Prepare current data (test set)
current_data = pd.concat([data.x_test, data.y_test], axis=1)

View File

@@ -67,6 +67,11 @@ class TrainingRepository:
Exception: If data loading, preprocessing, or training fails
"""
data = load_data(uploaded_file, params.line_separator, params.decimal_separator)
# Configure datetime index if timestamp column exists
# Required for TimeSeriesDiscontinuityAnalyzer (static window removal)
data = self._configure_datetime_index(data)
process_data = self._init_data_preprocessor(params)
process_data.fit(data)
data_view = process_data.transform(data)
@@ -88,6 +93,8 @@ class TrainingRepository:
regr = LinearRegressionModel(
target_variable=params.target_variable,
variable_columns=params.variable_columns,
degree=params.degree,
interaction_only=params.interaction_only,
)
regr.fit(data_train)
@@ -126,7 +133,9 @@ class TrainingRepository:
TrainModelResult: Updated result with predictions, denormalized data,
and metrics (mse_val, mae_val, r2_val)
"""
# Calculate predictions BEFORE denormalization (important for polynomial models)
y_pred_array = tmr.regr.predict(tmr.x_test)
y_train_pred_array = tmr.regr.predict(tmr.x_train)
if params.use_scaler:
scaler = tmr.process_data.get_scaler()
@@ -140,6 +149,9 @@ class TrainingRepository:
tmr.y_train = scaler.denormalize_single_input(tmr.y_train, params.target_variable)
tmr.y_test = scaler.denormalize_single_input(tmr.y_test, params.target_variable)
y_pred_array = scaler.denormalize_predictions(y_pred_array, params.target_variable)
y_train_pred_array = scaler.denormalize_predictions(
y_train_pred_array, params.target_variable
)
else:
# Fallback for sklearn StandardScaler: only inverse-transform features
feature_cols = getattr(
@@ -156,11 +168,15 @@ class TrainingRepository:
tmr.y_pred = pd.Series(y_pred_array, index=tmr.y_test.index)
tmr.y_pred.name = f'{params.target_variable}_pred'
tmr.y_train_pred = pd.Series(y_train_pred_array, index=tmr.y_train.index)
tmr.y_train_pred.name = f'{params.target_variable}_pred'
tmr.x_train = tmr.x_train.sort_index()
tmr.x_test = tmr.x_test.sort_index()
tmr.y_train = tmr.y_train.sort_index()
tmr.y_test = tmr.y_test.sort_index()
tmr.y_pred = tmr.y_pred.sort_index()
tmr.y_train_pred = tmr.y_train_pred.sort_index()
assert tmr.y_pred is not None, 'y_pred should be set at this point'
@@ -226,6 +242,20 @@ class TrainingRepository:
return scaler_dict
def _get_static_threshold(self, params: TrainModelParams) -> int | None:
"""
Get the static threshold value based on parameters.
Args:
params: Training parameters containing static window configuration
Returns:
int | None: Static threshold value (1-1000) if rem_static_win is True, None otherwise
"""
if not params.rem_static_win:
return None
return params.static_threshold if params.static_threshold is not None else 1
def _init_data_preprocessor(self, params: TrainModelParams) -> DataPreprocessor:
"""
Initialize DataPreprocessor with training parameters.
@@ -236,20 +266,27 @@ class TrainingRepository:
Returns:
DataPreprocessor: Configured preprocessor ready for fitting
"""
# Create lag dictionaries for each variable
lag_train_dict = dict.fromkeys(params.variable_columns, params.lag_train)
lag_val_dict = dict.fromkeys(params.variable_columns, params.lag_val)
# Convert removed_intervals to list of tuples if needed
removed_intervals = None
if params.removed_intervals:
removed_intervals = [
(interval[0], interval[1]) if isinstance(interval, (list, tuple)) else interval
for interval in params.removed_intervals
]
return DataPreprocessor(
target_variable=params.target_variable,
input_columns=params.variable_columns,
lag_train=lag_train_dict,
lag_transform=lag_val_dict,
static_threshold=1 if params.rem_static_win else None,
nan_treatment=params.nan_treatment,
lag_train=params.lag_train,
lag_transform=params.lag_val,
start_date=params.start_date,
end_date=params.end_date,
removed_intervals=removed_intervals,
static_threshold=self._get_static_threshold(params),
low_lim=params.low_lim,
upp_lim=params.upp_lim,
window=params.window,
scaler_name='Standard Scaler' if params.use_scaler else 'None',
scaler_name=params.scaler_name,
scaler_params={} if params.use_scaler else None,
ar_var=params.target_variable if params.include_ar else None,
)
@@ -279,10 +316,17 @@ class TrainingRepository:
coefficients = regr.regr.coef_
intercept = regr.regr.intercept_
# Get feature names - for polynomial models, use poly_feature_names
if params.degree > 1 and regr.poly_feature_names:
feature_names = regr.poly_feature_names
else:
feature_names = params.variable_columns
# Create coefficients dictionary
coefficients_dict = {}
for i, var in enumerate(params.variable_columns):
coefficients_dict[var] = float(coefficients[i])
for i, var in enumerate(feature_names):
if i < len(coefficients):
coefficients_dict[var] = float(coefficients[i])
# Create equation string
equation_parts = [f'{coef:.6f} * {var}' for var, coef in coefficients_dict.items()]
@@ -300,5 +344,73 @@ class TrainingRepository:
'intercept': float(intercept),
'equation_string': equation_string,
'latex_equation': latex_equation,
'model_type': 'Linear Regression',
'model_type': params.model_name,
'degree': params.degree,
'interaction_only': params.interaction_only,
'original_features': params.variable_columns,
}
def _configure_datetime_index(self, data: pd.DataFrame) -> pd.DataFrame:
"""
Configure datetime index for the DataFrame.
This method attempts to identify a timestamp column and set it as the
DataFrame index with DatetimeIndex type. This is required for
TimeSeriesDiscontinuityAnalyzer (used in static window removal).
The method looks for common timestamp column names and converts the
first matching column to datetime, then sets it as the index.
Args:
data: Input DataFrame
Returns:
pd.DataFrame: DataFrame with DatetimeIndex if timestamp column found,
otherwise returns original DataFrame unchanged
"""
# If index is already DatetimeIndex, just ensure it's sorted
if isinstance(data.index, pd.DatetimeIndex):
self.logger.info('DataFrame already has DatetimeIndex')
return data.sort_index()
# Common timestamp column names
timestamp_columns = [
'timestamp',
'Timestamp',
'TIMESTAMP',
'date',
'Date',
'DATE',
'datetime',
'DateTime',
]
for col in timestamp_columns:
if col in data.columns:
try:
data[col] = pd.to_datetime(data[col])
data = data.set_index(col)
data = data.sort_index()
self.logger.info(f'Configured datetime index from column: {col}')
return data
except (ValueError, TypeError) as e:
self.logger.warning(f'Failed to convert column {col} to datetime: {e}')
continue
# If no timestamp column found, check if first column looks like a timestamp
first_col = data.columns[0]
try:
# Try to parse first column as datetime
test_values = data[first_col].head(10).dropna()
if len(test_values) > 0:
pd.to_datetime(test_values)
data[first_col] = pd.to_datetime(data[first_col])
data = data.set_index(first_col)
data = data.sort_index()
self.logger.info(f'Configured datetime index from first column: {first_col}')
return data
except (ValueError, TypeError):
pass
self.logger.warning('No timestamp column found - some features may not work correctly')
return data

View File

@@ -15,6 +15,7 @@ Prerequisites:
from __future__ import annotations
import argparse
import asyncio
import json
import os
@@ -24,8 +25,8 @@ import uuid
from datetime import datetime, timedelta
from pathlib import Path
from dotenv import load_dotenv
import psycopg2
from dotenv import load_dotenv
from psycopg2.extras import Json
from temporalio import client
@@ -37,7 +38,8 @@ if ENV_PATH.exists():
load_dotenv(dotenv_path=ENV_PATH)
DOCS_PATH = Path('docs/test-model-data.csv')
DEFAULT_CSV_PATH = Path('docs/test-model-data.csv')
TEST_SCENARIOS_DIR = PROJECT_ROOT / 'docs' / 'test-scenarios'
MINIO_ALIAS = 'suse'
MINIO_BUCKET = 'model-training'
@@ -54,26 +56,45 @@ TEMPORAL_NAMESPACE = os.getenv('TEMPORAL_NAMESPACE')
TRAIN_TASK_QUEUE = os.getenv('TRAIN_TASK_QUEUE')
TEMPORAL_WORKFLOW = 'train_model'
BASE_REQUEST_DATA = {
'experimentName': 'model-manager-test-01',
'username': 'bruno.domingues@aignosi.com.br',
'modelType': 'Linear Regression',
'targetVariable': '03CV020/CORRENTE_N_M1_PV(Value)',
'variableColumns': ['303-WIT-200(Value)'],
'lagTrain': 0,
'lagVal': 0,
'remStaticWin': False,
'lowLim': {},
'uppLim': {},
'window': 0,
'useScaler': False,
'includeAr': False,
'trainSize': 80,
'shuffle': True,
'lineSeparator': ',',
'decimalSeparator': '.',
'removedIntervals': [],
}
def list_available_scenarios() -> list[str]:
"""List all available test scenario files."""
if not TEST_SCENARIOS_DIR.exists():
return []
return sorted([f.stem for f in TEST_SCENARIOS_DIR.glob('*.json')])
def load_scenario(scenario_name: str) -> dict:
"""Load a test scenario from JSON file.
Args:
scenario_name: Name of the scenario (without .json extension)
or full path to a JSON file.
Returns:
Dictionary with scenario data.
Raises:
FileNotFoundError: If scenario file doesn't exist.
"""
# Check if it's a full path
scenario_path = Path(scenario_name)
if scenario_path.suffix == '.json' and scenario_path.exists():
with open(scenario_path) as f:
return json.load(f)
# Otherwise, look in the test-scenarios directory
scenario_file = TEST_SCENARIOS_DIR / f'{scenario_name}.json'
if not scenario_file.exists():
available = list_available_scenarios()
available_str = ', '.join(available) if available else 'none'
raise FileNotFoundError(
f"Scenario '{scenario_name}' not found at {scenario_file}.\n"
f'Available scenarios: {available_str}'
)
with open(scenario_file) as f:
return json.load(f)
def _ensure_source_file(path: Path) -> None:
@@ -126,7 +147,7 @@ def insert_experiment_run(file_name: str, request_data: dict) -> int:
(
request_data['experimentName'],
request_data['username'],
'ORCHESTRATOR_REQUEST_SENT',
'ORCHESTRATOR_WAITING_PROC',
now,
now,
MINIO_BUCKET,
@@ -149,7 +170,6 @@ def build_workflow_payload(
'experiment_run_id': experiment_run_id,
'experiment_name': request_data['experimentName'],
'username': request_data['username'],
'model_type': request_data['modelType'],
'target_variable': request_data['targetVariable'],
'variable_columns': request_data['variableColumns'],
'lag_train': request_data['lagTrain'],
@@ -167,6 +187,15 @@ def build_workflow_payload(
'line_separator': request_data['lineSeparator'],
'decimal_separator': request_data['decimalSeparator'],
'removed_intervals': request_data['removedIntervals'],
# New parameters
'model_name': request_data.get('modelName', 'Linear Regression'),
'degree': request_data.get('degree', 1),
'interaction_only': request_data.get('interactionOnly', False),
'nan_treatment': request_data.get('nanTreatment', 'drop'),
'start_date': request_data.get('startDate'),
'end_date': request_data.get('endDate'),
'scaler_name': request_data.get('scalerName', 'None'),
'support_filters': request_data.get('supportFilters', {}),
}
@@ -191,24 +220,106 @@ async def trigger_temporal_workflow(workflow_input: dict) -> str:
return workflow_id
def main() -> None:
def parse_args() -> argparse.Namespace:
"""Parse command line arguments."""
parser = argparse.ArgumentParser(
description='Run training workflow tests with different scenarios.',
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
# List available scenarios
python scripts/run_training_test.py --list
# Run a specific scenario
python scripts/run_training_test.py --scenario linear-regression-basic
# Run with a custom JSON file
python scripts/run_training_test.py --scenario /path/to/custom-scenario.json
# Run with a custom CSV data file
python scripts/run_training_test.py --scenario linear-regression-basic --csv docs/other-data.csv
# Run all scenarios
python scripts/run_training_test.py --all
""",
)
parser.add_argument(
'--scenario',
'-s',
type=str,
help='Name of the test scenario (without .json) or path to a JSON file.',
)
parser.add_argument(
'--csv',
'-c',
type=Path,
default=DEFAULT_CSV_PATH,
help=f'Path to the CSV data file (default: {DEFAULT_CSV_PATH}).',
)
parser.add_argument(
'--list',
'-l',
action='store_true',
help='List all available test scenarios and exit.',
)
parser.add_argument(
'--all',
'-a',
action='store_true',
help='Run all available test scenarios sequentially.',
)
return parser.parse_args()
def run_single_scenario(scenario_name: str, csv_path: Path) -> dict:
"""Run a single test scenario and return the result.
Args:
scenario_name: Name of the scenario to run.
csv_path: Path to the CSV data file.
Returns:
Dictionary with scenario result including success status and details.
"""
result = {
'scenario': scenario_name,
'success': False,
'error': None,
'experiment_run_id': None,
's3_object_name': None,
'workflow_id': None,
}
# Load scenario
try:
uploaded_file_name = upload_to_minio(DOCS_PATH)
except subprocess.CalledProcessError as exc:
print(f'Failed to upload file to MinIO: {exc}', file=sys.stderr)
sys.exit(1)
experiment_request = load_scenario(scenario_name)
print(f" Loaded scenario: {scenario_name}")
except FileNotFoundError as exc:
print(str(exc), file=sys.stderr)
sys.exit(1)
result['error'] = str(exc)
return result
experiment_request = BASE_REQUEST_DATA.copy()
# Upload CSV to MinIO
try:
uploaded_file_name = upload_to_minio(csv_path)
result['s3_object_name'] = uploaded_file_name
print(f" Uploaded CSV to MinIO: {uploaded_file_name}")
except subprocess.CalledProcessError as exc:
result['error'] = f'Failed to upload file to MinIO: {exc}'
return result
except FileNotFoundError as exc:
result['error'] = str(exc)
return result
# Insert experiment run
try:
experiment_run_id = insert_experiment_run(uploaded_file_name, experiment_request)
result['experiment_run_id'] = experiment_run_id
print(f" Created experiment_run with ID: {experiment_run_id}")
except psycopg2.Error as exc:
print(f'Database error while inserting experiment_run: {exc}', file=sys.stderr)
sys.exit(1)
result['error'] = f'Database error while inserting experiment_run: {exc}'
return result
# Build and trigger workflow
workflow_payload = build_workflow_payload(
experiment_run_id=experiment_run_id,
file_name=uploaded_file_name,
@@ -217,16 +328,103 @@ def main() -> None:
try:
workflow_id = asyncio.run(trigger_temporal_workflow(workflow_payload))
result['workflow_id'] = workflow_id
result['success'] = True
print(f" Workflow started: {workflow_id}")
except Exception as exc: # noqa: BLE001
print(f'Failed to start Temporal workflow: {exc}', file=sys.stderr)
result['error'] = f'Failed to start Temporal workflow: {exc}'
return result
return result
def print_summary(results: list[dict]) -> None:
"""Print a summary of all scenario results.
Args:
results: List of result dictionaries from run_single_scenario.
"""
passed = [r for r in results if r['success']]
failed = [r for r in results if not r['success']]
print('\n' + '=' * 60)
print('SUMMARY')
print('=' * 60)
print(f"Total: {len(results)} | Passed: {len(passed)} | Failed: {len(failed)}")
print('=' * 60)
if passed:
print('\n✓ PASSED:')
for r in passed:
print(f" - {r['scenario']}")
if failed:
print('\n✗ FAILED:')
for r in failed:
print(f" - {r['scenario']}")
if r['error']:
print(f" Error: {r['error']}")
print()
def main() -> None:
args = parse_args()
# List scenarios and exit if requested
if args.list:
scenarios = list_available_scenarios()
if scenarios:
print('Available test scenarios:')
for scenario in scenarios:
print(f' - {scenario}')
else:
print(f'No scenarios found in {TEST_SCENARIOS_DIR}')
sys.exit(0)
# Run all scenarios if requested
if args.all:
scenarios = list_available_scenarios()
if not scenarios:
print(f'No scenarios found in {TEST_SCENARIOS_DIR}', file=sys.stderr)
sys.exit(1)
print(f'Running {len(scenarios)} scenarios...\n')
results = []
for i, scenario in enumerate(scenarios, 1):
print(f'[{i}/{len(scenarios)}] Running scenario: {scenario}')
result = run_single_scenario(scenario, args.csv)
results.append(result)
status = '' if result['success'] else ''
print(f'[{i}/{len(scenarios)}] {status} {scenario}\n')
print_summary(results)
# Exit with error code if any scenario failed
failed_count = sum(1 for r in results if not r['success'])
sys.exit(1 if failed_count > 0 else 0)
# Require scenario argument if not listing or running all
if not args.scenario:
print('Error: --scenario or --all is required. Use --list to see available scenarios.', file=sys.stderr)
sys.exit(1)
# Run single scenario
print(f'Running scenario: {args.scenario}')
result = run_single_scenario(args.scenario, args.csv)
if not result['success']:
print(f"Error: {result['error']}", file=sys.stderr)
sys.exit(1)
print(
json.dumps(
{
'experiment_run_id': experiment_run_id,
's3_object_name': uploaded_file_name,
'workflow_id': workflow_id,
'scenario': result['scenario'],
'experiment_run_id': result['experiment_run_id'],
's3_object_name': result['s3_object_name'],
'workflow_id': result['workflow_id'],
},
indent=2,
)

View File

@@ -1,8 +1,17 @@
"""Unit tests for sientia metrics module."""
import numpy as np
import pandas as pd
from model_manager.sientia.metrics import mae, mse, r2
from model_manager.sientia.metrics import (
mae,
mse,
r2,
rce_drift,
rce_test,
rce_train,
silverman_radius,
)
def test_mse_perfect_predictions():
@@ -200,3 +209,273 @@ def test_r2_with_mixed_positive_negative():
result = r2(real_data, predictions)
assert result == 1.0
# ============================================================================
# Tests for silverman_radius
# ============================================================================
def test_silverman_radius_basic():
"""Test silverman_radius returns a positive float."""
data = np.array([1.0, 2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0, 10.0])
result = silverman_radius(data)
assert isinstance(result, float)
assert result > 0
def test_silverman_radius_uniform_data():
"""Test silverman_radius with uniformly distributed data."""
data = np.linspace(0, 100, 50)
result = silverman_radius(data)
assert result > 0
assert np.isfinite(result)
def test_silverman_radius_normal_distribution():
"""Test silverman_radius with normally distributed data."""
np.random.seed(42)
data = np.random.normal(loc=50, scale=10, size=100)
result = silverman_radius(data)
assert result > 0
assert np.isfinite(result)
def test_silverman_radius_small_dataset():
"""Test silverman_radius with small dataset."""
data = np.array([1.0, 2.0, 3.0])
result = silverman_radius(data)
assert result > 0
# ============================================================================
# Tests for rce_train
# ============================================================================
def test_rce_train_returns_dataframe():
"""Test rce_train returns a DataFrame."""
training_set = pd.DataFrame({'a': [1.0, 2.0, 3.0, 4.0, 5.0], 'b': [2.0, 3.0, 4.0, 5.0, 6.0]})
result = rce_train(training_set, 0.1)
assert isinstance(result, pd.DataFrame)
def test_rce_train_includes_first_vector():
"""Test rce_train always includes the first vector as a prototype."""
training_set = pd.DataFrame({'a': [1.0, 2.0, 3.0], 'b': [1.0, 2.0, 3.0]})
result = rce_train(training_set, 0.1)
assert len(result) >= 1
assert result.iloc[0].tolist() == [1.0, 1.0]
def test_rce_train_with_identical_vectors():
"""Test rce_train with identical vectors returns single prototype."""
training_set = pd.DataFrame({'a': [1.0, 1.0, 1.0], 'b': [2.0, 2.0, 2.0]})
result = rce_train(training_set, 0.1)
# All vectors are identical, so only one prototype should be created
assert len(result) == 1
def test_rce_train_with_distant_vectors():
"""Test rce_train with very distant vectors creates multiple prototypes."""
training_set = pd.DataFrame({'a': [0.0, 100.0, 200.0], 'b': [0.0, 100.0, 200.0]})
result = rce_train(training_set, 0.1)
# Distant vectors should create multiple prototypes
assert len(result) >= 1
# ============================================================================
# Tests for rce_test
# ============================================================================
def test_rce_test_returns_series():
"""Test rce_test returns a pandas Series."""
test_set = pd.DataFrame({'a': [1.5, 2.5], 'b': [1.5, 2.5]})
prototypes = pd.DataFrame({'a': [1.0, 3.0], 'b': [1.0, 3.0]})
result = rce_test(test_set, prototypes)
assert isinstance(result, pd.Series)
assert len(result) == len(test_set)
def test_rce_test_with_exact_match():
"""Test rce_test with test vector matching a prototype."""
test_set = pd.DataFrame({'a': [1.0], 'b': [2.0]})
prototypes = pd.DataFrame({'a': [1.0], 'b': [2.0]})
result = rce_test(test_set, prototypes)
# Distance should be 0 for exact match
assert result.iloc[0] == 0.0
def test_rce_test_multiple_prototypes():
"""Test rce_test finds closest prototype."""
test_set = pd.DataFrame({'a': [1.1], 'b': [1.1]})
prototypes = pd.DataFrame({'a': [1.0, 10.0], 'b': [1.0, 10.0]})
result = rce_test(test_set, prototypes)
# Should find the closest prototype (1.0, 1.0)
assert len(result) == 1
assert np.isfinite(result.iloc[0])
def test_rce_test_signed_distances():
"""Test rce_test returns signed distances."""
test_set = pd.DataFrame({'a': [0.0, 5.0], 'b': [0.0, 5.0]})
prototypes = pd.DataFrame({'a': [2.0], 'b': [2.0]})
result = rce_test(test_set, prototypes)
assert len(result) == 2
# First test vector (0,0) is less than prototype (2,2) - should be negative
# Second test vector (5,5) is greater than prototype (2,2) - should be positive
assert result.iloc[0] < 0
assert result.iloc[1] > 0
# ============================================================================
# Tests for rce_drift
# ============================================================================
def test_rce_drift_returns_series():
"""Test rce_drift returns a pandas Series."""
reference_data = pd.DataFrame(
{
'feature1': [1.0, 2.0, 3.0, 4.0, 5.0],
'feature2': [2.0, 3.0, 4.0, 5.0, 6.0],
'target': [10.0, 20.0, 30.0, 40.0, 50.0],
'prediction': [11.0, 21.0, 31.0, 41.0, 51.0],
}
)
real_data = pd.DataFrame(
{
'feature1': [1.5, 2.5],
'feature2': [2.5, 3.5],
'target': [15.0, 25.0],
'prediction': [16.0, 26.0],
}
)
result = rce_drift(reference_data, real_data, 'target')
assert isinstance(result, pd.Series)
assert len(result) == len(real_data)
def test_rce_drift_with_target_column():
"""Test rce_drift using target column (drops prediction)."""
reference_data = pd.DataFrame(
{
'feature1': [1.0, 2.0, 3.0],
'target': [10.0, 20.0, 30.0],
'prediction': [11.0, 21.0, 31.0],
}
)
real_data = pd.DataFrame(
{
'feature1': [1.5],
'target': [15.0],
'prediction': [16.0],
}
)
result = rce_drift(reference_data, real_data, 'target')
assert isinstance(result, pd.Series)
assert len(result) == 1
def test_rce_drift_with_prediction_column():
"""Test rce_drift using prediction column (drops target)."""
reference_data = pd.DataFrame(
{
'feature1': [1.0, 2.0, 3.0],
'target': [10.0, 20.0, 30.0],
'prediction': [11.0, 21.0, 31.0],
}
)
real_data = pd.DataFrame(
{
'feature1': [1.5],
'target': [15.0],
'prediction': [16.0],
}
)
result = rce_drift(reference_data, real_data, 'prediction')
assert isinstance(result, pd.Series)
assert len(result) == 1
def test_rce_drift_normalized_output():
"""Test rce_drift returns normalized distances."""
reference_data = pd.DataFrame(
{
'feature1': [1.0, 2.0, 3.0, 4.0, 5.0],
'target': [10.0, 20.0, 30.0, 40.0, 50.0],
'prediction': [10.0, 20.0, 30.0, 40.0, 50.0],
}
)
real_data = pd.DataFrame(
{
'feature1': [2.5, 3.5],
'target': [25.0, 35.0],
'prediction': [25.0, 35.0],
}
)
result = rce_drift(reference_data, real_data, 'target')
# Result should be a Series with same length as real_data
assert isinstance(result, pd.Series)
assert len(result) == len(real_data)
def test_rce_drift_handles_common_columns():
"""Test rce_drift correctly handles common columns between datasets."""
reference_data = pd.DataFrame(
{
'feature1': [1.0, 2.0, 3.0],
'feature2': [2.0, 3.0, 4.0],
'extra_ref': [100.0, 200.0, 300.0],
'target': [10.0, 20.0, 30.0],
'prediction': [11.0, 21.0, 31.0],
}
)
real_data = pd.DataFrame(
{
'feature1': [1.5],
'feature2': [2.5],
'extra_real': [150.0],
'target': [15.0],
'prediction': [16.0],
}
)
result = rce_drift(reference_data, real_data, 'target')
# Should work with only common columns
assert isinstance(result, pd.Series)
assert len(result) == 1

View File

@@ -81,13 +81,15 @@ def test_linear_regression_model_fit():
def test_linear_regression_model_fit_without_variable_columns():
"""Test LinearRegressionModel fit raises AssertionError without variable_columns."""
"""Test LinearRegressionModel fit infers variable_columns when not set."""
model = LinearRegressionModel(target_variable='target')
data = pd.DataFrame({'var1': [1, 2, 3], 'target': [3, 5, 7]})
with raises(AssertionError, match='variable_columns must be set before fitting'):
model.fit(data)
# Model should infer variable_columns from data (all columns except target)
result = model.fit(data)
assert result is model
assert model.variable_columns == ['var1']
def test_linear_regression_model_predict_without_clipping():
@@ -196,7 +198,7 @@ def test_data_preprocessor_init_with_custom_steps_order():
assert 'Normalization' in preprocessor.steps_order
assert 'Feature Creation' in preprocessor.steps_order
assert len(preprocessor.steps_order) == 7
assert len(preprocessor.steps_order) == 8 # Now includes RANGE_SELECTION step
def test_data_preprocessor_get_scaler():
@@ -690,3 +692,358 @@ def test_data_preprocessor_transform_all_steps():
result = preprocessor.transform(test_x)
assert isinstance(result, pd.DataFrame)
# ============================================================================
# Additional tests for coverage - LinearRegressionModel
# ============================================================================
def test_linear_regression_model_fit_without_target_variable():
"""Test fit raises error when target_variable is not set."""
model = LinearRegressionModel()
data = pd.DataFrame({'var1': [1, 2, 3], 'target': [3, 5, 7]})
with raises(ValueError, match='target_variable must be set before fitting'):
model.fit(data)
def test_linear_regression_model_fit_with_missing_columns():
"""Test fit raises error when variable_columns are missing from data."""
model = LinearRegressionModel(
target_variable='target', variable_columns=['var1', 'var_missing']
)
data = pd.DataFrame({'var1': [1, 2, 3], 'target': [3, 5, 7]})
with raises(ValueError, match='Columns not found in input data'):
model.fit(data)
def test_linear_regression_model_fit_with_missing_target():
"""Test fit raises error when target_variable is not in data."""
model = LinearRegressionModel(target_variable='missing_target', variable_columns=['var1'])
data = pd.DataFrame({'var1': [1, 2, 3], 'other': [3, 5, 7]})
with raises(ValueError, match='Target variable missing_target not found in input data'):
model.fit(data)
def test_linear_regression_model_fit_with_inf_values():
"""Test fit handles infinite values by converting to NaN."""
model = LinearRegressionModel(target_variable='target', variable_columns=['var1', 'var2'])
# Include some inf values that will be converted to NaN and rows dropped
data = pd.DataFrame(
{
'var1': [1.0, 2.0, 3.0, np.inf, 5.0],
'var2': [2.0, 3.0, 4.0, 5.0, 6.0],
'target': [3.0, 5.0, 7.0, 9.0, 11.0],
}
)
model.fit(data)
# Model should fit successfully after removing row with inf
assert model.weights is not None
assert 'var1' in model.variable_columns
assert 'var2' in model.variable_columns
def test_linear_regression_model_fit_polynomial():
"""Test fit with polynomial features."""
model = LinearRegressionModel(target_variable='target', variable_columns=['var1'], degree=2)
data = pd.DataFrame({'var1': [1, 2, 3, 4, 5], 'target': [3, 5, 7, 9, 11]})
model.fit(data)
assert model.poly_feature_names is not None
assert len(model.poly_feature_names) > 1
def test_linear_regression_model_predict_polynomial():
"""Test predict with polynomial features."""
model = LinearRegressionModel(target_variable='target', variable_columns=['var1'], degree=2)
train_data = pd.DataFrame({'var1': [1, 2, 3, 4, 5], 'target': [1, 4, 9, 16, 25]})
model.fit(train_data)
test_data = pd.DataFrame({'var1': [6, 7]})
predictions = model.predict(test_data)
assert isinstance(predictions, np.ndarray)
assert len(predictions) == 2
def test_linear_regression_model_create_poly_features_degree_1():
"""Test create_poly_features returns input unchanged when degree <= 1."""
model = LinearRegressionModel(degree=1)
data = pd.DataFrame({'var1': [1, 2, 3], 'var2': [4, 5, 6]})
result = model.create_poly_features(data, fit=True)
pd.testing.assert_frame_equal(result, data)
def test_linear_regression_model_create_poly_features_not_fitted():
"""Test create_poly_features raises error when not fitted and fit=False."""
model = LinearRegressionModel(degree=2)
data = pd.DataFrame({'var1': [1, 2, 3], 'var2': [4, 5, 6]})
with raises(ValueError, match='PolynomialFeatures not fitted'):
model.create_poly_features(data, fit=False)
def test_linear_regression_model_create_poly_features_transform():
"""Test create_poly_features with fit=False after fitting."""
model = LinearRegressionModel(degree=2)
train_data = pd.DataFrame({'var1': [1, 2, 3], 'var2': [4, 5, 6]})
model.create_poly_features(train_data, fit=True)
test_data = pd.DataFrame({'var1': [4, 5], 'var2': [7, 8]})
result = model.create_poly_features(test_data, fit=False)
assert isinstance(result, pd.DataFrame)
assert len(result.columns) > 2
def test_linear_regression_model_get_regressor():
"""Test get_regressor returns the underlying LinearRegression model."""
model = LinearRegressionModel()
regressor = model.get_regressor()
from sklearn.linear_model import LinearRegression
assert isinstance(regressor, LinearRegression)
# ============================================================================
# Additional tests for coverage - DataPreprocessor
# ============================================================================
@patch('model_manager.sientia.models.treat_nan')
def test_data_preprocessor_treat_discontinuities_linear_interpolation(mock_treat_nan):
"""Test treat_discontinuities with 'linear interpolation' treatment."""
preprocessor = DataPreprocessor(nan_treatment='linear interpolation')
data = pd.DataFrame({'col1': [1, np.nan, 3]})
expected_data = pd.DataFrame({'col1': [1.0, 2.0, 3.0]})
mock_treat_nan.return_value = expected_data
result = preprocessor.treat_discontinuities(data)
# Should map 'linear interpolation' to 'fill linear'
mock_treat_nan.assert_called_once_with(data, 'fill linear')
pd.testing.assert_frame_equal(result, expected_data)
def test_data_preprocessor_range_selection_with_start_date():
"""Test range_selection filters by start_date."""
preprocessor = DataPreprocessor(start_date='2023-01-02')
data = pd.DataFrame(
{'col1': [1, 2, 3]},
index=pd.to_datetime(['2023-01-01', '2023-01-02', '2023-01-03']),
)
result = preprocessor.range_selection(data)
assert len(result) == 2
assert result.index[0] == pd.Timestamp('2023-01-02')
def test_data_preprocessor_range_selection_with_end_date():
"""Test range_selection filters by end_date."""
preprocessor = DataPreprocessor(end_date='2023-01-02')
data = pd.DataFrame(
{'col1': [1, 2, 3]},
index=pd.to_datetime(['2023-01-01', '2023-01-02', '2023-01-03']),
)
result = preprocessor.range_selection(data)
assert len(result) == 2
assert result.index[-1] == pd.Timestamp('2023-01-02')
def test_data_preprocessor_range_selection_with_invalid_start_date():
"""Test range_selection handles invalid start_date gracefully."""
preprocessor = DataPreprocessor(start_date='invalid-date')
data = pd.DataFrame(
{'col1': [1, 2, 3]},
index=pd.to_datetime(['2023-01-01', '2023-01-02', '2023-01-03']),
)
result = preprocessor.range_selection(data)
# Should skip filtering and return original data
assert len(result) == 3
def test_data_preprocessor_range_selection_with_invalid_end_date():
"""Test range_selection handles invalid end_date gracefully."""
preprocessor = DataPreprocessor(end_date='invalid-date')
data = pd.DataFrame(
{'col1': [1, 2, 3]},
index=pd.to_datetime(['2023-01-01', '2023-01-02', '2023-01-03']),
)
result = preprocessor.range_selection(data)
# Should skip filtering and return original data
assert len(result) == 3
def test_data_preprocessor_range_selection_with_removed_intervals():
"""Test range_selection removes specified intervals."""
preprocessor = DataPreprocessor(removed_intervals=[['2023-01-02', '2023-01-03']])
data = pd.DataFrame(
{'col1': [1, 2, 3, 4, 5]},
index=pd.to_datetime(
['2023-01-01', '2023-01-02', '2023-01-03', '2023-01-04', '2023-01-05']
),
)
result = preprocessor.range_selection(data)
assert len(result) == 3
assert pd.Timestamp('2023-01-02') not in result.index
assert pd.Timestamp('2023-01-03') not in result.index
def test_data_preprocessor_range_selection_with_invalid_interval():
"""Test range_selection handles invalid interval dates gracefully."""
preprocessor = DataPreprocessor(removed_intervals=[['invalid', 'dates']])
data = pd.DataFrame(
{'col1': [1, 2, 3]},
index=pd.to_datetime(['2023-01-01', '2023-01-02', '2023-01-03']),
)
result = preprocessor.range_selection(data)
# Should skip invalid interval and return original data
assert len(result) == 3
def test_data_preprocessor_range_selection_with_short_interval():
"""Test range_selection skips intervals with less than 2 elements."""
preprocessor = DataPreprocessor(removed_intervals=[['2023-01-02']])
data = pd.DataFrame(
{'col1': [1, 2, 3]},
index=pd.to_datetime(['2023-01-01', '2023-01-02', '2023-01-03']),
)
result = preprocessor.range_selection(data)
# Should skip short interval and return original data
assert len(result) == 3
def test_data_preprocessor_predict():
"""Test predict method removes target and preserves feature order."""
preprocessor = DataPreprocessor(
target_variable='target',
input_columns=['var1', 'var2'],
steps_order=['Discontinuity Treatment'],
)
train_x = pd.DataFrame({'var1': [1, 2, 3], 'var2': [2, 3, 4], 'target': [3, 5, 7]})
preprocessor.fit(train_x)
test_x = pd.DataFrame({'var1': [4, 5], 'var2': [5, 6], 'target': [9, 11]})
result = preprocessor.predict(test_x)
assert 'target' not in result.columns
assert 'var1' in result.columns
assert 'var2' in result.columns
def test_data_preprocessor_predict_without_target():
"""Test predict when target is not in transformed data."""
preprocessor = DataPreprocessor(
target_variable='target',
input_columns=['var1', 'var2'],
steps_order=['Discontinuity Treatment'],
)
train_x = pd.DataFrame({'var1': [1, 2, 3], 'var2': [2, 3, 4], 'target': [3, 5, 7]})
preprocessor.fit(train_x)
# Include target in test data so transform works, predict will remove it
test_x = pd.DataFrame({'var1': [4, 5], 'var2': [5, 6], 'target': [9, 11]})
result = preprocessor.predict(test_x)
# Target should be removed by predict
assert 'target' not in result.columns
assert 'var1' in result.columns
assert 'var2' in result.columns
def test_data_preprocessor_predict_preserves_feature_order():
"""Test predict preserves feature order from fit."""
preprocessor = DataPreprocessor(
target_variable='target',
input_columns=['var1', 'var2'],
steps_order=['Discontinuity Treatment'],
)
train_x = pd.DataFrame({'var1': [1, 2, 3], 'var2': [2, 3, 4], 'target': [3, 5, 7]})
preprocessor.fit(train_x)
# Test data has columns in different order
test_x = pd.DataFrame({'var2': [5, 6], 'var1': [4, 5], 'target': [9, 11]})
result = preprocessor.predict(test_x)
# Should have columns in same order as during fit
assert list(result.columns) == ['var1', 'var2']
# ============================================================================
# Additional tests for 100% coverage
# ============================================================================
def test_data_preprocessor_predict_target_not_in_columns():
"""Test predict when target_variable is not in transformed data columns."""
preprocessor = DataPreprocessor(
target_variable='target',
input_columns=['var1', 'var2'],
steps_order=['Discontinuity Treatment'],
)
train_x = pd.DataFrame({'var1': [1, 2, 3], 'var2': [2, 3, 4], 'target': [3, 5, 7]})
preprocessor.fit(train_x)
# Test data without target column - predict should still work
test_x = pd.DataFrame({'var1': [4, 5], 'var2': [5, 6]})
# Need to add target for transform to work, then it gets removed
test_x['target'] = [9, 11]
# Manually remove target before calling predict to test the branch
preprocessor_copy = DataPreprocessor(
target_variable='nonexistent_target',
input_columns=['var1', 'var2'],
steps_order=['Discontinuity Treatment'],
)
preprocessor_copy.fit(train_x.rename(columns={'target': 'nonexistent_target'}))
test_x_no_target = pd.DataFrame({'var1': [4, 5], 'var2': [5, 6], 'nonexistent_target': [9, 11]})
result = preprocessor_copy.predict(test_x_no_target)
assert 'var1' in result.columns
assert 'var2' in result.columns
def test_data_preprocessor_predict_without_fitted_feature_order():
"""Test predict when _fitted_feature_order is None."""
preprocessor = DataPreprocessor(
target_variable='target',
input_columns=['var1', 'var2'],
steps_order=['Discontinuity Treatment'],
)
train_x = pd.DataFrame({'var1': [1, 2, 3], 'var2': [2, 3, 4], 'target': [3, 5, 7]})
preprocessor.fit(train_x)
# Manually set _fitted_feature_order to None to test the branch
preprocessor._fitted_feature_order = None
test_x = pd.DataFrame({'var1': [4, 5], 'var2': [5, 6], 'target': [9, 11]})
result = preprocessor.predict(test_x)
# Should still work, just without reordering
assert 'target' not in result.columns
assert 'var1' in result.columns
assert 'var2' in result.columns

View File

@@ -24,15 +24,13 @@ def test_load_html_from_file_success(tmp_path):
assert content == '<p>Hello</p>'
def test_load_html_from_file_missing_file(capsys):
def test_load_html_from_file_missing_file():
result = reports.load_html_from_file('non-existent.html')
captured = capsys.readouterr()
assert result is None
assert 'File not found: non-existent.html' in captured.out
def test_load_html_from_file_os_error(monkeypatch, capsys):
def test_load_html_from_file_os_error(monkeypatch):
def fake_open(*_args, **_kwargs):
raise OSError('boom')
@@ -40,9 +38,7 @@ def test_load_html_from_file_os_error(monkeypatch, capsys):
result = reports.load_html_from_file('path.html')
captured = capsys.readouterr()
assert result is None
assert 'Error reading file: boom' in captured.out
def test_inject_content_replaces_section():
@@ -57,15 +53,15 @@ def test_inject_content_replaces_section():
assert section.find('span').text == 'new'
def test_inject_content_missing_section(capsys):
def test_inject_content_missing_section():
main_html = "<html><body><div id='other'>keep</div></body></html>"
result = reports.inject_content(main_html, 'missing', '<p>ignored</p>')
captured = capsys.readouterr()
assert "Section with id 'missing' not found" in captured.out
# Content should be unchanged when section is missing
soup = BeautifulSoup(result, 'html.parser')
assert soup.find(id='other') is not None
assert soup.find(id='other').text == 'keep'
def test_reports_init_sets_defaults(stub_color_options):

View File

@@ -8,8 +8,8 @@ def valid_train_params_dict():
"""Create a valid dictionary for TrainModelParams."""
return {
'variable_columns': ['var1', 'var2'],
'lag_train': 5,
'lag_val': 3,
'lag_train': {'var1': 5, 'var2': 5},
'lag_val': {'var1': 3, 'var2': 3},
'target_variable': 'target',
'rem_static_win': True,
'low_lim': {'var1': 0.0, 'var2': 1.0},
@@ -26,6 +26,15 @@ def valid_train_params_dict():
'experiment_run_id': 1,
'experiment_name': 'test_experiment',
'removed_intervals': [],
'model_name': 'Linear Regression',
'degree': 1,
'interaction_only': False,
'nan_treatment': 'drop',
'start_date': None,
'end_date': None,
'scaler_name': 'Standard Scaler',
'support_filters': {},
'static_threshold': None,
}
@@ -36,8 +45,8 @@ def test_train_model_params_from_dict_success(valid_train_params_dict):
params = TrainModelParams.from_dict(valid_train_params_dict)
assert params.variable_columns == ['var1', 'var2']
assert params.lag_train == 5
assert params.lag_val == 3
assert params.lag_train == {'var1': 5, 'var2': 5}
assert params.lag_val == {'var1': 3, 'var2': 3}
assert params.target_variable == 'target'
assert params.rem_static_win is True
assert params.low_lim == {'var1': 0.0, 'var2': 1.0}
@@ -118,9 +127,9 @@ def test_train_model_params_from_dict_wrong_type(valid_train_params_dict):
"""Test from_dict raises TypeError when field has wrong type."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['lag_train'] = 'not_an_int'
valid_train_params_dict['lag_train'] = 'not_a_dict'
with pytest.raises(TypeError, match='lag_train must be of type int, but got str'):
with pytest.raises(TypeError, match='lag_train must be of type dict, but got str'):
TrainModelParams.from_dict(valid_train_params_dict)
@@ -179,10 +188,10 @@ def test_validate_business_rules_negative_lag_train(valid_train_params_dict):
"""Test validate_business_rules raises error when lag_train is negative."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['lag_train'] = -1
valid_train_params_dict['lag_train'] = {'var1': -1, 'var2': 5}
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='lag_train must be positive, got -1'):
with pytest.raises(ValueError, match='lag_train for var1 must be non-negative, got -1'):
params.validate_business_rules()
@@ -190,10 +199,10 @@ def test_validate_business_rules_negative_lag_val(valid_train_params_dict):
"""Test validate_business_rules raises error when lag_val is negative."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['lag_val'] = -2
valid_train_params_dict['lag_val'] = {'var1': 3, 'var2': -2}
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='lag_val must be positive, got -2'):
with pytest.raises(ValueError, match='lag_val for var2 must be non-negative, got -2'):
params.validate_business_rules()
@@ -204,7 +213,7 @@ def test_validate_business_rules_negative_window(valid_train_params_dict):
valid_train_params_dict['window'] = -5
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='window must be positive, got -5'):
with pytest.raises(ValueError, match='window must be non-negative, got -5'):
params.validate_business_rules()
@@ -334,7 +343,7 @@ def test_validate_business_rules_zero_lag_train(valid_train_params_dict):
"""Test validate_business_rules accepts lag_train = 0."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['lag_train'] = 0
valid_train_params_dict['lag_train'] = {'var1': 0, 'var2': 0}
params = TrainModelParams.from_dict(valid_train_params_dict)
params.validate_business_rules() # Should not raise
@@ -344,7 +353,7 @@ def test_validate_business_rules_zero_lag_val(valid_train_params_dict):
"""Test validate_business_rules accepts lag_val = 0."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['lag_val'] = 0
valid_train_params_dict['lag_val'] = {'var1': 0, 'var2': 0}
params = TrainModelParams.from_dict(valid_train_params_dict)
params.validate_business_rules() # Should not raise
@@ -369,3 +378,283 @@ def test_validate_business_rules_empty_limits(valid_train_params_dict):
params = TrainModelParams.from_dict(valid_train_params_dict)
params.validate_business_rules() # Should not raise
# ============================================================================
# Additional tests for 100% coverage
# ============================================================================
def test_validate_business_rules_degree_less_than_1(valid_train_params_dict):
"""Test validate_business_rules raises error when degree < 1."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['degree'] = 0
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='degree must be at least 1, got 0'):
params.validate_business_rules()
def test_validate_business_rules_invalid_nan_treatment(valid_train_params_dict):
"""Test validate_business_rules raises error for invalid nan_treatment."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['nan_treatment'] = 'invalid_treatment'
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='nan_treatment must be one of'):
params.validate_business_rules()
def test_validate_business_rules_invalid_scaler_name(valid_train_params_dict):
"""Test validate_business_rules raises error for invalid scaler_name."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['scaler_name'] = 'Invalid Scaler'
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='scaler_name must be one of'):
params.validate_business_rules()
def test_validate_business_rules_invalid_model_name(valid_train_params_dict):
"""Test validate_business_rules raises error for invalid model_name."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['model_name'] = 'Invalid Model'
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='model_name must be one of'):
params.validate_business_rules()
def test_validate_business_rules_polynomial_regression_degree_less_than_2(valid_train_params_dict):
"""Test validate_business_rules raises error for Polynomial Regression with degree < 2."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['model_name'] = 'Polynomial Regression'
valid_train_params_dict['degree'] = 1
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='degree must be at least 2 for Polynomial Regression'):
params.validate_business_rules()
def test_validate_business_rules_polynomial_regression_without_scaler(valid_train_params_dict):
"""Test validate_business_rules raises error for Polynomial Regression without scaler."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['model_name'] = 'Polynomial Regression'
valid_train_params_dict['degree'] = 2
valid_train_params_dict['scaler_name'] = 'None'
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='scaler_name must be set'):
params.validate_business_rules()
def test_validate_business_rules_linear_regression_degree_not_1(valid_train_params_dict):
"""Test validate_business_rules raises error for Linear Regression with degree != 1."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['model_name'] = 'Linear Regression'
valid_train_params_dict['degree'] = 2
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='degree must be 1 for Linear Regression, got 2'):
params.validate_business_rules()
def test_validate_business_rules_removed_intervals_not_list(valid_train_params_dict):
"""Test validate_business_rules raises error when removed_intervals item is not list."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['removed_intervals'] = ['not_a_list']
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='removed_intervals\\[0\\] must be a list or tuple'):
params.validate_business_rules()
def test_validate_business_rules_removed_intervals_too_short(valid_train_params_dict):
"""Test validate_business_rules raises error when removed_intervals item has < 2 elements."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['removed_intervals'] = [['only_one_element']]
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='removed_intervals\\[0\\] must have at least 2 elements'):
params.validate_business_rules()
def test_validate_business_rules_start_date_wrong_type(valid_train_params_dict):
"""Test validate_business_rules raises error when start_date is not a string."""
from model_manager.utils.models.train_model_params import TrainModelParams
# Create params normally first, then modify start_date to bypass from_dict validation
params = TrainModelParams.from_dict(valid_train_params_dict)
params.start_date = 12345 # type: ignore
with pytest.raises(TypeError, match='start_date must be a string, got int'):
params.validate_business_rules()
def test_validate_business_rules_end_date_wrong_type(valid_train_params_dict):
"""Test validate_business_rules raises error when end_date is not a string."""
from model_manager.utils.models.train_model_params import TrainModelParams
# Create params normally first, then modify end_date to bypass from_dict validation
params = TrainModelParams.from_dict(valid_train_params_dict)
params.end_date = 12345 # type: ignore
with pytest.raises(TypeError, match='end_date must be a string, got int'):
params.validate_business_rules()
def test_validate_business_rules_empty_target_variable(valid_train_params_dict):
"""Test validate_business_rules raises error when target_variable is empty."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['target_variable'] = ''
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='target_variable cannot be empty or whitespace'):
params.validate_business_rules()
def test_validate_business_rules_whitespace_target_variable(valid_train_params_dict):
"""Test validate_business_rules raises error when target_variable is whitespace."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['target_variable'] = ' '
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='target_variable cannot be empty or whitespace'):
params.validate_business_rules()
def test_validate_business_rules_valid_removed_intervals(valid_train_params_dict):
"""Test validate_business_rules accepts valid removed_intervals."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['removed_intervals'] = [['2023-01-01', '2023-01-02']]
params = TrainModelParams.from_dict(valid_train_params_dict)
params.validate_business_rules() # Should not raise
def test_validate_business_rules_valid_start_and_end_date(valid_train_params_dict):
"""Test validate_business_rules accepts valid start_date and end_date."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['start_date'] = '2023-01-01'
valid_train_params_dict['end_date'] = '2023-12-31'
params = TrainModelParams.from_dict(valid_train_params_dict)
params.validate_business_rules() # Should not raise
def test_validate_business_rules_polynomial_regression_valid(valid_train_params_dict):
"""Test validate_business_rules accepts valid Polynomial Regression config."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['model_name'] = 'Polynomial Regression'
valid_train_params_dict['degree'] = 2
valid_train_params_dict['scaler_name'] = 'Standard Scaler'
params = TrainModelParams.from_dict(valid_train_params_dict)
params.validate_business_rules() # Should not raise
def test_validate_business_rules_static_threshold_valid(valid_train_params_dict):
"""Test validate_business_rules accepts valid static_threshold when rem_static_win is True."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['rem_static_win'] = True
valid_train_params_dict['static_threshold'] = 500
params = TrainModelParams.from_dict(valid_train_params_dict)
params.validate_business_rules() # Should not raise
def test_validate_business_rules_static_threshold_min_valid(valid_train_params_dict):
"""Test validate_business_rules accepts static_threshold = 1."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['rem_static_win'] = True
valid_train_params_dict['static_threshold'] = 1
params = TrainModelParams.from_dict(valid_train_params_dict)
params.validate_business_rules() # Should not raise
def test_validate_business_rules_static_threshold_max_valid(valid_train_params_dict):
"""Test validate_business_rules accepts static_threshold = 1000."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['rem_static_win'] = True
valid_train_params_dict['static_threshold'] = 1000
params = TrainModelParams.from_dict(valid_train_params_dict)
params.validate_business_rules() # Should not raise
def test_validate_business_rules_static_threshold_below_min(valid_train_params_dict):
"""Test validate_business_rules raises error when static_threshold < 1."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['rem_static_win'] = True
valid_train_params_dict['static_threshold'] = 0
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='static_threshold must be between 1 and 1000, got 0'):
params.validate_business_rules()
def test_validate_business_rules_static_threshold_above_max(valid_train_params_dict):
"""Test validate_business_rules raises error when static_threshold > 1000."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['rem_static_win'] = True
valid_train_params_dict['static_threshold'] = 1001
params = TrainModelParams.from_dict(valid_train_params_dict)
with pytest.raises(ValueError, match='static_threshold must be between 1 and 1000, got 1001'):
params.validate_business_rules()
def test_validate_business_rules_static_threshold_none_when_rem_static_win_true(
valid_train_params_dict,
):
"""Test validate_business_rules accepts None static_threshold when rem_static_win is True."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['rem_static_win'] = True
valid_train_params_dict['static_threshold'] = None
params = TrainModelParams.from_dict(valid_train_params_dict)
params.validate_business_rules() # Should not raise - None is allowed
def test_validate_business_rules_static_threshold_ignored_when_rem_static_win_false(
valid_train_params_dict,
):
"""Test validate_business_rules ignores static_threshold when rem_static_win is False."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['rem_static_win'] = False
valid_train_params_dict['static_threshold'] = 5000 # Invalid value, but should be ignored
params = TrainModelParams.from_dict(valid_train_params_dict)
params.validate_business_rules() # Should not raise - validation skipped
def test_from_dict_static_threshold_type_error(valid_train_params_dict):
"""Test from_dict raises TypeError when static_threshold has wrong type."""
from model_manager.utils.models.train_model_params import TrainModelParams
valid_train_params_dict['static_threshold'] = 'not_an_int'
with pytest.raises(TypeError, match='static_threshold must be of type int, but got str'):
TrainModelParams.from_dict(valid_train_params_dict)

View File

@@ -14,8 +14,8 @@ def sample_params():
"""Create sample TrainModelParams for testing."""
return TrainModelParams(
variable_columns=['var1', 'var2'],
lag_train=5,
lag_val=3,
lag_train={'var1': 5, 'var2': 5},
lag_val={'var1': 3, 'var2': 3},
target_variable='target',
rem_static_win=True,
low_lim={'var1': 0.0, 'var2': 0.0},
@@ -32,6 +32,15 @@ def sample_params():
experiment_run_id=123,
experiment_name='test-experiment',
removed_intervals=[],
model_name='Linear Regression',
degree=1,
interaction_only=False,
nan_treatment='drop',
start_date=None,
end_date=None,
scaler_name='Standard Scaler',
support_filters={},
static_threshold=None,
)
@@ -175,11 +184,11 @@ def test_train_model_result_is_dataclass(sample_params, sample_dataframes):
def test_train_model_result_field_count():
"""Test that TrainModelResult has exactly 19 fields."""
"""Test that TrainModelResult has exactly 20 fields."""
from dataclasses import fields
result_fields = fields(TrainModelResult)
assert len(result_fields) == 19
assert len(result_fields) == 20
field_names = {f.name for f in result_fields}
expected_fields = {
@@ -192,6 +201,7 @@ def test_train_model_result_field_count():
'regr',
'scaler_dict',
'y_pred',
'y_train_pred',
'mse_val',
'mae_val',
'r2_val',

View File

@@ -55,6 +55,8 @@ def mock_train_result():
result.params.include_ar = False
result.params.train_size = 80
result.params.removed_intervals = []
result.params.rem_static_win = True
result.params.static_threshold = None
result.run_name = 'test_run'
result.run_dir = '/tmp/test_run' # noqa: S108
result.report_path = '/tmp/test_run/report.html' # noqa: S108
@@ -308,6 +310,23 @@ def test_init_artifacts_data_none_y_pred(mock_model_serving_class, mock_logger,
repo._init_artifacts_data(mock_train_result)
@patch('model_manager.utils.repository.model_repository.ModelServing')
def test_init_artifacts_data_none_y_train_pred(
mock_model_serving_class, mock_logger, mock_train_result
):
"""Test _init_artifacts_data raises ValueError when y_train_pred is None."""
from model_manager.utils.repository.model_repository import ModelRepository
repo = ModelRepository(
url='http://mlflow.test', username='user', password='pass', logger=mock_logger
)
mock_train_result.y_train_pred = None
with pytest.raises(ValueError, match='Training predictions .* are None'):
repo._init_artifacts_data(mock_train_result)
@patch('model_manager.utils.repository.model_repository.ModelServing')
@patch('model_manager.utils.repository.model_repository.datetime')
@patch('model_manager.utils.repository.model_repository.makedirs')
@@ -474,6 +493,70 @@ def test_save_run_with_equation(
assert '/tmp/test_run/model_equation.json' in logged_artifacts # noqa: S108
@patch('model_manager.utils.repository.model_repository.ModelServing')
@patch('model_manager.utils.repository.model_repository.path.exists')
def test_save_run_with_static_threshold_value(
mock_exists, mock_model_serving_class, mock_logger, mock_train_result
):
"""Test _save_run logs static_threshold when rem_static_win is True and value is set."""
from model_manager.utils.repository.model_repository import ModelRepository
mock_model_serving_instance = MagicMock()
mock_model_serving_class.return_value = mock_model_serving_instance
repo = ModelRepository(
url='http://mlflow.test', username='user', password='pass', logger=mock_logger
)
# Set static_threshold to a specific value
mock_train_result.params.rem_static_win = True
mock_train_result.params.static_threshold = 500
mock_train_result.equation_path = None
# Mock all path.exists calls to return True
mock_exists.return_value = True
repo._save_run(mock_train_result)
# Verify static_threshold was logged with the correct value
log_param_calls = {
call[0][0]: call[0][1] for call in mock_model_serving_instance.log_param.call_args_list
}
assert log_param_calls['static_threshold'] == 500
@patch('model_manager.utils.repository.model_repository.ModelServing')
@patch('model_manager.utils.repository.model_repository.path.exists')
def test_save_run_with_rem_static_win_false(
mock_exists, mock_model_serving_class, mock_logger, mock_train_result
):
"""Test _save_run logs static_threshold as None when rem_static_win is False."""
from model_manager.utils.repository.model_repository import ModelRepository
mock_model_serving_instance = MagicMock()
mock_model_serving_class.return_value = mock_model_serving_instance
repo = ModelRepository(
url='http://mlflow.test', username='user', password='pass', logger=mock_logger
)
# Set rem_static_win to False
mock_train_result.params.rem_static_win = False
mock_train_result.params.static_threshold = 500 # Should be ignored
mock_train_result.equation_path = None
# Mock all path.exists calls to return True
mock_exists.return_value = True
repo._save_run(mock_train_result)
# Verify static_threshold was logged as None
log_param_calls = {
call[0][0]: call[0][1] for call in mock_model_serving_instance.log_param.call_args_list
}
assert log_param_calls['static_threshold'] is None
@patch('model_manager.utils.repository.model_repository.ModelServing')
@patch('model_manager.utils.repository.model_repository.path.exists')
def test_save_run_without_equation(

View File

@@ -33,8 +33,8 @@ def sample_params():
experiment_name='test_experiment',
target_variable='target',
variable_columns=['var1', 'var2', 'var3'],
lag_train=0,
lag_val=0,
lag_train={'var1': 0, 'var2': 0, 'var3': 0},
lag_val={'var1': 0, 'var2': 0, 'var3': 0},
rem_static_win=False,
low_lim={},
upp_lim={},
@@ -48,6 +48,15 @@ def sample_params():
line_separator=',',
decimal_separator='.',
removed_intervals=[],
model_name='Linear Regression',
degree=1,
interaction_only=False,
nan_treatment='drop',
start_date=None,
end_date=None,
scaler_name='None',
support_filters={},
static_threshold=None,
)
@@ -106,8 +115,8 @@ class TestExtractModelEquation:
experiment_name='test',
target_variable='y',
variable_columns=['x'],
lag_train=0,
lag_val=0,
lag_train={'x': 0},
lag_val={'x': 0},
rem_static_win=False,
low_lim={},
upp_lim={},
@@ -121,6 +130,15 @@ class TestExtractModelEquation:
line_separator=',',
decimal_separator='.',
removed_intervals=[],
model_name='Linear Regression',
degree=1,
interaction_only=False,
nan_treatment='drop',
start_date=None,
end_date=None,
scaler_name='None',
support_filters={},
static_threshold=None,
)
# Mock model with single coefficient
@@ -177,6 +195,7 @@ class TestInitDataPreprocessor:
def test_init_preprocessor_with_scaler(self, training_repo, sample_params):
"""Test preprocessor initialization with scaler enabled."""
sample_params.use_scaler = True
sample_params.scaler_name = 'Standard Scaler'
preprocessor = training_repo._init_data_preprocessor(sample_params)
assert preprocessor.scaler_name == 'Standard Scaler'
@@ -184,6 +203,7 @@ class TestInitDataPreprocessor:
def test_init_preprocessor_without_scaler(self, training_repo, sample_params):
"""Test preprocessor initialization without scaler."""
sample_params.use_scaler = False
sample_params.scaler_name = 'None'
preprocessor = training_repo._init_data_preprocessor(sample_params)
assert preprocessor.scaler_name == 'None'
@@ -203,22 +223,75 @@ class TestInitDataPreprocessor:
assert preprocessor.ar_var is None
def test_init_preprocessor_with_static_removal(self, training_repo, sample_params):
"""Test preprocessor with static window removal enabled."""
"""Test preprocessor with static window removal enabled and no static_threshold."""
sample_params.rem_static_win = True
sample_params.static_threshold = None
preprocessor = training_repo._init_data_preprocessor(sample_params)
assert preprocessor.static_threshold == 1
def test_init_preprocessor_lag_configuration(self, training_repo, sample_params):
"""Test preprocessor lag configuration."""
sample_params.lag_train = 5
sample_params.lag_val = 3
def test_init_preprocessor_with_static_removal_custom_threshold(
self, training_repo, sample_params
):
"""Test preprocessor with static window removal and custom static_threshold."""
sample_params.rem_static_win = True
sample_params.static_threshold = 500
preprocessor = training_repo._init_data_preprocessor(sample_params)
# Check that lag dictionaries are created correctly
for col in sample_params.variable_columns:
assert preprocessor.lag_train[col] == 5
assert preprocessor.lag_transform[col] == 3
assert preprocessor.static_threshold == 500
def test_init_preprocessor_without_static_removal_ignores_threshold(
self, training_repo, sample_params
):
"""Test preprocessor without static removal ignores static_threshold."""
sample_params.rem_static_win = False
sample_params.static_threshold = 500
preprocessor = training_repo._init_data_preprocessor(sample_params)
assert preprocessor.static_threshold is None
def test_init_preprocessor_lag_configuration(self, training_repo, sample_params):
"""Test preprocessor lag configuration."""
sample_params.lag_train = {'var1': 5, 'var2': 5, 'var3': 5}
sample_params.lag_val = {'var1': 3, 'var2': 3, 'var3': 3}
preprocessor = training_repo._init_data_preprocessor(sample_params)
# Check that lag dictionaries are passed correctly
assert preprocessor.lag_train == {'var1': 5, 'var2': 5, 'var3': 5}
assert preprocessor.lag_transform == {'var1': 3, 'var2': 3, 'var3': 3}
class TestGetStaticThreshold:
"""Tests for _get_static_threshold method."""
def test_get_static_threshold_rem_static_win_false(self, training_repo, sample_params):
"""Test returns None when rem_static_win is False."""
sample_params.rem_static_win = False
sample_params.static_threshold = 500
result = training_repo._get_static_threshold(sample_params)
assert result is None
def test_get_static_threshold_rem_static_win_true_with_value(
self, training_repo, sample_params
):
"""Test returns static_threshold value when rem_static_win is True and value is set."""
sample_params.rem_static_win = True
sample_params.static_threshold = 500
result = training_repo._get_static_threshold(sample_params)
assert result == 500
def test_get_static_threshold_rem_static_win_true_with_none(self, training_repo, sample_params):
"""Test returns 1 when rem_static_win is True and static_threshold is None."""
sample_params.rem_static_win = True
sample_params.static_threshold = None
result = training_repo._get_static_threshold(sample_params)
assert result == 1
class TestInitScalerDict:
@@ -308,8 +381,14 @@ class TestAfterTrainCalculation:
y_train = pd.Series([100, 200, 300], index=[0, 1, 2], name='target')
y_test = pd.Series([400, 500], index=[3, 4], name='target')
# Mock predict to return a simple array
sample_linear_model.predict = MagicMock(return_value=np.array([450.0, 550.0]))
# Mock predict to return arrays with correct length based on input
def mock_predict(data):
if len(data) == 3: # x_train
return np.array([150.0, 250.0, 350.0])
else: # x_test
return np.array([450.0, 550.0])
sample_linear_model.predict = MagicMock(side_effect=mock_predict)
return TrainModelResult(
params=sample_params,
@@ -392,13 +471,24 @@ class TestAfterTrainCalculation:
y_train = pd.Series([100, 200, 300], index=[0, 1, 2], name='target')
y_test = pd.Series([400, 500], index=[3, 4], name='target')
# Mock predict to return a simple array
sample_linear_model.predict = MagicMock(return_value=np.array([450.0, 550.0]))
# Mock predict to return arrays with correct length based on input
def mock_predict(data):
if len(data) == 3: # x_train
return np.array([150.0, 250.0, 350.0])
else: # x_test
return np.array([450.0, 550.0])
sample_linear_model.predict = MagicMock(side_effect=mock_predict)
# Create mock scaler with denormalize methods
mock_scaler = MagicMock()
mock_scaler.denormalize_single_input = MagicMock(side_effect=lambda x, col: x * 2)
mock_scaler.denormalize_predictions = MagicMock(return_value=np.array([900.0, 1100.0]))
# denormalize_predictions needs to return correct length based on input
def mock_denormalize_predictions(arr, col):
return arr * 2
mock_scaler.denormalize_predictions = MagicMock(side_effect=mock_denormalize_predictions)
# Create mock preprocessor
mock_process_data = MagicMock()
@@ -436,8 +526,14 @@ class TestAfterTrainCalculation:
y_train = pd.Series([100, 200, 300], index=[0, 1, 2], name='target')
y_test = pd.Series([400, 500], index=[3, 4], name='target')
# Mock predict
sample_linear_model.predict = MagicMock(return_value=np.array([450.0, 550.0]))
# Mock predict to return arrays with correct length based on input
def mock_predict(data):
if len(data) == 3: # x_train
return np.array([150.0, 250.0, 350.0])
else: # x_test
return np.array([450.0, 550.0])
sample_linear_model.predict = MagicMock(side_effect=mock_predict)
# Create mock sklearn scaler (without denormalize methods)
mock_scaler = MagicMock()
@@ -489,6 +585,7 @@ class TestTrain:
"""
return BytesIO(csv_content.encode('utf-8'))
@patch.object(TrainingRepository, '_configure_datetime_index', lambda self, df: df)
@patch('model_manager.utils.repository.training_repository.split_train_test')
@patch('model_manager.utils.repository.training_repository.load_data')
def test_train_basic_workflow(
@@ -534,6 +631,7 @@ class TestTrain:
# Verify split was called
assert mock_split_train_test.called
@patch.object(TrainingRepository, '_configure_datetime_index', lambda self, df: df)
@patch('model_manager.utils.repository.training_repository.split_train_test')
@patch('model_manager.utils.repository.training_repository.load_data')
def test_train_with_scaler(
@@ -563,6 +661,7 @@ class TestTrain:
assert result is not None
assert result.scaler_dict is not None
@patch.object(TrainingRepository, '_configure_datetime_index', lambda self, df: df)
@patch('model_manager.utils.repository.training_repository.split_train_test')
@patch('model_manager.utils.repository.training_repository.load_data')
def test_train_with_shuffle_enabled(
@@ -593,6 +692,7 @@ class TestTrain:
call_kwargs = mock_split_train_test.call_args[1]
assert call_kwargs['shuffle'] is True
@patch.object(TrainingRepository, '_configure_datetime_index', lambda self, df: df)
@patch('model_manager.utils.repository.training_repository.split_train_test')
@patch('model_manager.utils.repository.training_repository.load_data')
def test_train_with_different_train_size(
@@ -623,6 +723,7 @@ class TestTrain:
call_kwargs = mock_split_train_test.call_args[1]
assert call_kwargs['train_size'] == 0.7
@patch.object(TrainingRepository, '_configure_datetime_index', lambda self, df: df)
@patch('model_manager.utils.repository.training_repository.split_train_test')
@patch('model_manager.utils.repository.training_repository.load_data')
def test_train_raises_on_empty_data_after_transform(
@@ -643,6 +744,7 @@ class TestTrain:
with pytest.raises(ValueError, match='Data view is empty after transformation'):
training_repo.train(sample_csv_data, sample_params)
@patch.object(TrainingRepository, '_configure_datetime_index', lambda self, df: df)
@patch('model_manager.utils.repository.training_repository.split_train_test')
@patch('model_manager.utils.repository.training_repository.load_data')
def test_train_logs_success(
@@ -679,6 +781,7 @@ class TestTrain:
'Model trained successfully' in str(call) for call in mock_logger.info.call_args_list
)
@patch.object(TrainingRepository, '_configure_datetime_index', lambda self, df: df)
@patch('model_manager.utils.repository.training_repository.split_train_test')
@patch('model_manager.utils.repository.training_repository.load_data')
def test_train_with_custom_separators(
@@ -709,6 +812,7 @@ class TestTrain:
# Verify load_data was called with custom separators
mock_load_data.assert_called_once_with(sample_csv_data, ';', ',')
@patch.object(TrainingRepository, '_configure_datetime_index', lambda self, df: df)
@patch('model_manager.utils.repository.training_repository.split_train_test')
@patch('model_manager.utils.repository.training_repository.load_data')
def test_train_result_contains_all_fields(
@@ -743,3 +847,228 @@ class TestTrain:
assert hasattr(result, 'regr')
assert hasattr(result, 'scaler_dict')
assert result.params == sample_params
# ============================================================================
# Tests for _configure_datetime_index
# ============================================================================
class TestConfigureDatetimeIndex:
"""Tests for _configure_datetime_index method."""
@pytest.fixture
def training_repo(self, mock_logger):
"""Create a TrainingRepository instance."""
return TrainingRepository(logger=mock_logger)
def test_configure_datetime_index_already_datetime(self, training_repo):
"""Test _configure_datetime_index when index is already DatetimeIndex."""
data = pd.DataFrame(
{'var1': [1, 2, 3], 'var2': [4, 5, 6]},
index=pd.to_datetime(['2023-01-01', '2023-01-02', '2023-01-03']),
)
result = training_repo._configure_datetime_index(data)
assert isinstance(result.index, pd.DatetimeIndex)
assert len(result) == 3
def test_configure_datetime_index_with_timestamp_column(self, training_repo):
"""Test _configure_datetime_index with 'timestamp' column."""
data = pd.DataFrame(
{
'timestamp': ['2023-01-01', '2023-01-02', '2023-01-03'],
'var1': [1, 2, 3],
'var2': [4, 5, 6],
}
)
result = training_repo._configure_datetime_index(data)
assert isinstance(result.index, pd.DatetimeIndex)
assert 'timestamp' not in result.columns
def test_configure_datetime_index_with_date_column(self, training_repo):
"""Test _configure_datetime_index with 'date' column."""
data = pd.DataFrame(
{
'date': ['2023-01-01', '2023-01-02', '2023-01-03'],
'var1': [1, 2, 3],
}
)
result = training_repo._configure_datetime_index(data)
assert isinstance(result.index, pd.DatetimeIndex)
assert 'date' not in result.columns
def test_configure_datetime_index_with_datetime_column(self, training_repo):
"""Test _configure_datetime_index with 'datetime' column."""
data = pd.DataFrame(
{
'datetime': ['2023-01-01', '2023-01-02', '2023-01-03'],
'var1': [1, 2, 3],
}
)
result = training_repo._configure_datetime_index(data)
assert isinstance(result.index, pd.DatetimeIndex)
assert 'datetime' not in result.columns
def test_configure_datetime_index_first_column_datetime(self, training_repo):
"""Test _configure_datetime_index when first column looks like datetime."""
data = pd.DataFrame(
{
'my_date': ['2023-01-01', '2023-01-02', '2023-01-03'],
'var1': [1, 2, 3],
}
)
result = training_repo._configure_datetime_index(data)
assert isinstance(result.index, pd.DatetimeIndex)
assert 'my_date' not in result.columns
@pytest.mark.filterwarnings('ignore::UserWarning')
def test_configure_datetime_index_no_timestamp_column(self, training_repo):
"""Test _configure_datetime_index when no timestamp column found."""
data = pd.DataFrame(
{
'var1': ['text_a', 'text_b', 'text_c'],
'var2': ['text_d', 'text_e', 'text_f'],
}
)
result = training_repo._configure_datetime_index(data)
# Should return original data unchanged (no valid datetime columns)
assert 'var1' in result.columns
assert 'var2' in result.columns
@pytest.mark.filterwarnings('ignore::UserWarning')
def test_configure_datetime_index_invalid_timestamp_column(self, training_repo):
"""Test _configure_datetime_index with invalid timestamp values."""
data = pd.DataFrame(
{
'timestamp': ['not_a_date', 'also_not', 'nope'],
'var1': [1, 2, 3],
}
)
result = training_repo._configure_datetime_index(data)
# Should skip invalid column and try first column
assert 'var1' in result.columns
@pytest.mark.filterwarnings('ignore::UserWarning')
def test_configure_datetime_index_invalid_first_column(self, training_repo):
"""Test _configure_datetime_index when first column is not datetime."""
data = pd.DataFrame(
{
'var1': ['a', 'b', 'c'],
'var2': [1, 2, 3],
}
)
result = training_repo._configure_datetime_index(data)
# Should return original data unchanged
assert 'var1' in result.columns
assert 'var2' in result.columns
def test_configure_datetime_index_first_column_all_nan(self, training_repo):
"""Test _configure_datetime_index when first column has all NaN values."""
data = pd.DataFrame(
{
'first_col': [np.nan, np.nan, np.nan],
'var1': [1, 2, 3],
}
)
result = training_repo._configure_datetime_index(data)
# Should return original data unchanged (first column has no valid values)
assert 'first_col' in result.columns
assert 'var1' in result.columns
# ============================================================================
# Tests for _init_data_preprocessor with removed_intervals
# ============================================================================
class TestInitDataPreprocessorWithRemovedIntervals:
"""Tests for _init_data_preprocessor with removed_intervals."""
@pytest.fixture
def training_repo(self, mock_logger):
"""Create a TrainingRepository instance."""
return TrainingRepository(logger=mock_logger)
def test_init_data_preprocessor_with_removed_intervals(self, training_repo, sample_params):
"""Test _init_data_preprocessor with removed_intervals."""
sample_params.removed_intervals = [
['2023-01-01', '2023-01-02'],
['2023-02-01', '2023-02-02'],
]
preprocessor = training_repo._init_data_preprocessor(sample_params)
assert preprocessor is not None
assert preprocessor.removed_intervals is not None
assert len(preprocessor.removed_intervals) == 2
def test_init_data_preprocessor_with_tuple_intervals(self, training_repo, sample_params):
"""Test _init_data_preprocessor with tuple intervals."""
sample_params.removed_intervals = [('2023-01-01', '2023-01-02')]
preprocessor = training_repo._init_data_preprocessor(sample_params)
assert preprocessor is not None
assert preprocessor.removed_intervals is not None
# ============================================================================
# Tests for _extract_model_equation with polynomial features
# ============================================================================
class TestExtractModelEquationPolynomial:
"""Tests for _extract_model_equation with polynomial features."""
@pytest.fixture
def training_repo(self, mock_logger):
"""Create a TrainingRepository instance."""
return TrainingRepository(logger=mock_logger)
def test_extract_model_equation_polynomial(self, training_repo, sample_params):
"""Test _extract_model_equation with polynomial features."""
# Create a mock regressor with polynomial features
mock_regr = MagicMock()
mock_regr.regr.coef_ = np.array([0.5, 0.3, 0.2])
mock_regr.regr.intercept_ = 1.0
mock_regr.poly_feature_names = ['var1', 'var2', 'var1^2']
sample_params.degree = 2
result = training_repo._extract_model_equation(mock_regr, sample_params)
assert 'equation_string' in result
assert 'latex_equation' in result
assert 'var1' in result['equation_string']
def test_extract_model_equation_linear(self, training_repo, sample_params):
"""Test _extract_model_equation with linear features."""
mock_regr = MagicMock()
mock_regr.regr.coef_ = np.array([0.5, 0.3])
mock_regr.regr.intercept_ = 1.0
mock_regr.poly_feature_names = None
sample_params.degree = 1
result = training_repo._extract_model_equation(mock_regr, sample_params)
assert 'equation_string' in result
assert 'latex_equation' in result

View File

@@ -1,5 +0,0 @@
- Criar um gráfico no grafana para cada nova atividade.
- Atualizar o .github/workflows/quality-gate.yml para usar os pipelines genéricos do github;
Criar um workflow para fazer o deploy no suse.
Criar um workflow para criar o release no github.