From 94e7a7a505200b51fd42c3040da91c8af58374be Mon Sep 17 00:00:00 2001 From: Sebithaz-dev Date: Tue, 27 Oct 2026 18:40:13 -0300 Subject: [PATCH 1/3] feat(model): usa HistGradientBoosting con split temporal y umbral F1 --- README.md | 62 +++++++++--- config/model.yaml | 26 ++++- ml/config.py | 29 +++++- ml/features.py | 26 +++++ ml/predict.py | 9 +- ml/train.py | 215 +++++++++++++++++++++++++++++++++-------- requirements.txt | 2 +- tests/test_config.py | 73 +++++++++++++- tests/test_features.py | 61 ++++++++++++ tests/test_health.py | 2 +- tests/test_predict.py | 85 +++++++++------- tests/test_train.py | 120 +++++++++++++++++++++++ 12 files changed, 613 insertions(+), 97 deletions(-) create mode 100644 ml/features.py create mode 100644 tests/test_features.py create mode 100644 tests/test_train.py diff --git a/README.md b/README.md index a4b183c..2ad045f 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ [![Security & Quality](https://github.com/Sebithaz-dev/Model-Training-S3/actions/workflows/security.yml/badge.svg)](https://github.com/Sebithaz-dev/Model-Training-S3/actions/workflows/security.yml) [![Quality Gate](https://sonarcloud.io/api/project_badges/measure?project=Sebithaz-dev_Model-Training-S3&metric=alert_status)](https://sonarcloud.io/summary/new_code?id=Sebithaz-dev_Model-Training-S3) -Pipeline de entrenamiento automatizado para clasificar la duración de viajes de Citi Bike NYC en **corto** (< 15 min) o **largo** (≥ 15 min), utilizando datos curados desde AWS Glue y subiendo artefactos a S3. +Pipeline de entrenamiento automatizado para clasificar la duración de viajes de Citi Bike NYC en **corto** (≤ 15 min) o **largo** (> 15 min), utilizando datos curados desde AWS Glue y subiendo artefactos a S3. --- @@ -131,7 +131,7 @@ Antes del feature engineering se valida el schema del Parquet contra el contrato python -m ml.predict ``` -Ejecuta ejemplos de prueba con un Passenger (`Customer`, 5pm → largo) vs un Suscriptor (`Subscriber`, 8am → corto). +Ejecuta ejemplos de prueba con un Passenger (`Customer`, 5pm → largo) vs un Suscriptor (`Subscriber`, 6am → corto). ### Health check del pipeline @@ -178,20 +178,36 @@ Parquet con columnas: ### Feature Engineering -- `start_station_freq` — frequency encoding con `log1p` -- `usertype` / `gender` — `LabelEncoder` (fiteado sobre training set) -- Escalado con `StandardScaler` +- `start_station_freq` — frequency encoding con `log1p` (fiteado solo con train) +- `usertype` / `gender` — `LabelEncoder` (fiteado solo con train; categorías nuevas → `-1`) +- Features cíclicas: `hour_sin/cos`, `month_sin/cos`, `dayofweek_sin/cos` +- Flags horarios: `is_rush_hour` (7-9 y 17-19), `is_night` (22-5) +- Escalado con `StandardScaler` (fiteado solo con train) ### Target - **1** (`largo`) si `tripduration / 60 > 15` (> 900 segundos) - **0** (`corto`) en caso contrario -- Umbral de decisión del modelo: **0.65** +- El umbral de decisión se optimiza en validación maximizando **F1** y se persiste en `transformers.pkl` / `metricas.json` (el valor configurado en YAML queda como respaldo) + +### Split + +- **Temporal** por defecto: ordena por `starttime` y usa 70% train, 10% validación y 20% test. +- El modo `random` sigue disponible en `config/model.yaml` para comparar contra el baseline. ### Modelo ```python -LogisticRegression(random_state=42, max_iter=1000, class_weight="balanced") +HistGradientBoostingClassifier( + class_weight="balanced", + max_iter=300, + learning_rate=0.08, + max_leaf_nodes=31, + min_samples_leaf=20, + l2_regularization=0.1, + early_stopping=True, + random_state=42, +) ``` ### Salidas (subidas a S3) @@ -199,26 +215,43 @@ LogisticRegression(random_state=42, max_iter=1000, class_weight="balanced") | Archivo | Contenido | |---|---| | `modelo.pkl` | Modelo entrenado (joblib) | -| `transformers.pkl` | LabelEncoders + Scaler + station_freq + feature_cols | +| `transformers.pkl` | LabelEncoders + Scaler + station_freq + feature_cols + umbral | | `metricas.json` | Métricas de evaluación + umbral + features usadas | --- ## Métricas reportadas -Las métricas se calculan sobre el test set (80/20 split estratificado) y se guardan en `metricas.json`: +Las métricas se calculan sobre el test set temporal (70/10/20) y se guardan en `metricas.json`: | Métrica | Descripción | |---|---| | `accuracy` | Exactitud global | | `recall` | Recall clase positiva (`largo`) | | `precision` | Precisión clase positiva | -| `f1` | F1-score clase positiva | +| `f1_score` | F1-score clase positiva | | `roc_auc` | Área bajo la curva ROC | +| `pr_auc` | Área bajo la curva Precision-Recall | | `gini` | Coeficiente Gini (`2 * roc_auc - 1`) | -| `confusion_matrix` | Matriz [[TN, FP], [FN, TP]] | -| `train_size` / `test_size` | Tamaño de cada split | -| `proportion` | Proporción de viajes largos en test | +| `matriz_confusion` | Matriz [[TN, FP], [FN, TP]] | +| `train_size` / `validation_size` / `test_size` | Tamaño de cada split | +| `proporcion_largos` | Proporción de viajes largos en test | +| `umbral_decision` | Umbral F1 elegido en validación | + +### Últimas métricas (test temporal) + +| Métrica | Valor | +|---|---| +| ROC AUC | 0.6505 | +| PR AUC | 0.4455 | +| Gini | 0.3009 | +| Accuracy | 0.5500 | +| Recall (largo) | 0.7035 | +| Precision (largo) | 0.3473 | +| F1 Score (largo) | 0.4650 | +| Umbral decisión | 0.46 | +| Train / Val / Test | 331,401 / 47,343 / 94,686 | +| Matriz de confusión | [[33,557, 34,806], [7,804, 18,519]] | --- @@ -404,11 +437,14 @@ Reglas: │ ├── __init__.py │ ├── config.py # Carga de configuración YAML │ ├── data_validation.py # Validación de schema y calidad de datos +│ ├── features.py # Features temporales compartidas train/predict │ ├── health.py # Health check (config, artefactos, S3) │ ├── train.py # Entrenamiento del modelo │ ├── predict.py # Predicción local (CLI demo) │ └── s3_utils.py # Subida/descarga desde S3 ├── tests/ # Tests unitarios (pytest) +│ ├── test_features.py # Features temporales +│ ├── test_train.py # Split, umbral y pipeline │ └── test_health.py # Tests del health check ├── pyproject.toml # Config de ruff y pytest ├── requirements.txt # Dependencias runtime diff --git a/config/model.yaml b/config/model.yaml index c868f10..fc31152 100644 --- a/config/model.yaml +++ b/config/model.yaml @@ -5,19 +5,31 @@ target: # Umbral en segundos: viaje_largo = 1 si tripduration > trip_long_threshold_seconds trip_long_threshold_seconds: 900 - # Umbral de decision sobre la probabilidad del modelo + # Umbral de decision de respaldo; si tune_threshold=true se reemplaza por + # el umbral que maximiza threshold_metric en el set de validacion. decision_threshold: 0.65 + tune_threshold: true + threshold_metric: f1 split: + # temporal: ordena por starttime y entrena con el pasado, evalua con el futuro. + # random: split aleatorio estratificado (util para comparar contra el baseline). + strategy: temporal test_size: 0.2 + validation_size: 0.1 random_state: 42 stratify: true model: - class: sklearn.linear_model.LogisticRegression + class: sklearn.ensemble.HistGradientBoostingClassifier params: class_weight: balanced - max_iter: 1000 + max_iter: 300 + learning_rate: 0.08 + max_leaf_nodes: 31 + min_samples_leaf: 20 + l2_regularization: 0.1 + early_stopping: true random_state: 42 features: @@ -28,6 +40,14 @@ features: - is_weekend - age - start_station_freq + - hour_sin + - hour_cos + - month_sin + - month_cos + - dayofweek_sin + - dayofweek_cos + - is_rush_hour + - is_night categorical: - usertype - gender diff --git a/ml/config.py b/ml/config.py index dc4d2ad..c1c665c 100644 --- a/ml/config.py +++ b/ml/config.py @@ -20,7 +20,11 @@ class ModelConfig: trip_long_threshold_seconds: int decision_threshold: float + tune_threshold: bool + threshold_metric: str + split_strategy: str test_size: float + validation_size: float random_state: int stratify: bool model_class: str @@ -55,10 +59,14 @@ def load_config(path: str | os.PathLike | None = None) -> ModelConfig: features = data["features"] outliers = data["outliers"] - return ModelConfig( + cfg = ModelConfig( trip_long_threshold_seconds=int(target["trip_long_threshold_seconds"]), decision_threshold=float(target["decision_threshold"]), + tune_threshold=bool(target.get("tune_threshold", True)), + threshold_metric=str(target.get("threshold_metric", "f1")), + split_strategy=str(split.get("strategy", "temporal")), test_size=float(split["test_size"]), + validation_size=float(split.get("validation_size", 0.1)), random_state=int(split["random_state"]), stratify=bool(split["stratify"]), model_class=str(model["class"]), @@ -68,3 +76,22 @@ def load_config(path: str | os.PathLike | None = None) -> ModelConfig: age_min=int(outliers["age_min"]), age_max=int(outliers["age_max"]), ) + _validate_config(cfg) + return cfg + + +def _validate_config(cfg: ModelConfig) -> None: + if not 0 <= cfg.decision_threshold <= 1: + raise ValueError("target.decision_threshold debe estar entre 0 y 1") + if cfg.threshold_metric not in {"f1", "recall", "precision"}: + raise ValueError( + "target.threshold_metric debe ser f1, recall o precision" + ) + if cfg.split_strategy not in {"temporal", "random"}: + raise ValueError("split.strategy debe ser temporal o random") + if not 0 < cfg.test_size < 1: + raise ValueError("split.test_size debe estar entre 0 y 1") + if not 0 < cfg.validation_size < 1: + raise ValueError("split.validation_size debe estar entre 0 y 1") + if cfg.test_size + cfg.validation_size >= 1: + raise ValueError("test_size + validation_size debe ser menor que 1") diff --git a/ml/features.py b/ml/features.py new file mode 100644 index 0000000..dc9debf --- /dev/null +++ b/ml/features.py @@ -0,0 +1,26 @@ +"""Ingenieria de features temporales compartida por entrenamiento e inferencia.""" +from __future__ import annotations + +import numpy as np +import pandas as pd + + +def add_time_features(df: pd.DataFrame) -> pd.DataFrame: + """Agrega features ciclicas y flags de horario. + + Las formulas deben coincidir exactamente con las usadas por el dashboard + (``app/services/ml_service.py``) para evitar skew entre train y serving. + """ + out = df.copy() + out["hour_sin"] = np.sin(2 * np.pi * out["hour"] / 24) + out["hour_cos"] = np.cos(2 * np.pi * out["hour"] / 24) + out["month_sin"] = np.sin(2 * np.pi * (out["month"] - 1) / 12) + out["month_cos"] = np.cos(2 * np.pi * (out["month"] - 1) / 12) + out["dayofweek_sin"] = np.sin(2 * np.pi * out["dayofweek"] / 7) + out["dayofweek_cos"] = np.cos(2 * np.pi * out["dayofweek"] / 7) + out["is_rush_hour"] = ( + ((out["hour"] >= 7) & (out["hour"] <= 9)) + | ((out["hour"] >= 17) & (out["hour"] <= 19)) + ).astype(int) + out["is_night"] = ((out["hour"] <= 5) | (out["hour"] >= 22)).astype(int) + return out diff --git a/ml/predict.py b/ml/predict.py index 3f61230..4f2832f 100644 --- a/ml/predict.py +++ b/ml/predict.py @@ -6,6 +6,8 @@ import numpy as np import pandas as pd +from ml.features import add_time_features + warnings.filterwarnings("ignore") RUTA_ARTEFACTOS = os.path.join(os.path.dirname(__file__), "artefactos") @@ -38,6 +40,7 @@ def predecir(self, datos): df["start_station_freq"] = np.log1p( df["start_station_id"].map(self.station_freq).fillna(0) ) + df = add_time_features(df) for col in ["usertype", "gender"]: le = self.label_encoders[col] df[col] = ( @@ -53,6 +56,7 @@ def predecir(self, datos): "probabilidad_largo": round(float(prob), 4), "prediccion": pred, "etiqueta": "largo" if pred else "corto", + "umbral_decision": round(float(self.decision_threshold), 4), } @@ -71,6 +75,7 @@ def predecir(self, datos): "probabilidad_largo": "float - Probabilidad de viaje largo (0.0 - 1.0)", "prediccion": "int - 1 si viaje largo, 0 si corto", "etiqueta": "str - 'largo' | 'corto'", + "umbral_decision": "float - Umbral usado para clasificar", } @@ -91,7 +96,7 @@ def main(): "start_station_id": 3160, "usertype": "Subscriber", "gender": "Male", - "hour": 8, + "hour": 6, "month": 7, "dayofweek": 3, "is_weekend": 0, @@ -100,7 +105,7 @@ def main(): print("\nEjemplo 1 (Customer, 5pm):") print(json.dumps(predictor.predecir(ejemplo_1), indent=2)) - print("\nEjemplo 2 (Subscriber, 8am):") + print("\nEjemplo 2 (Subscriber, 6am):") print(json.dumps(predictor.predecir(ejemplo_2), indent=2)) diff --git a/ml/train.py b/ml/train.py index 434a585..d1b651e 100644 --- a/ml/train.py +++ b/ml/train.py @@ -7,12 +7,18 @@ import joblib import numpy as np import pandas as pd -from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score +from sklearn.metrics import ( + average_precision_score, + classification_report, + confusion_matrix, + roc_auc_score, +) from sklearn.model_selection import train_test_split from sklearn.preprocessing import LabelEncoder, StandardScaler from ml.config import ModelConfig, load_config from ml.data_validation import validate_dataframe +from ml.features import add_time_features warnings.filterwarnings("ignore") RUTA_ARTEFACTOS = os.path.join(os.path.dirname(__file__), "artefactos") @@ -27,81 +33,196 @@ def _resolve_model(cfg: ModelConfig): return cls(**cfg.model_params) +def _split_temporal(df: pd.DataFrame, cfg: ModelConfig) -> tuple[pd.DataFrame, ...]: + df = df.sort_values("starttime").reset_index(drop=True) + n_total = len(df) + n_test = int(n_total * cfg.test_size) + n_val = int(n_total * cfg.validation_size) + n_train = n_total - n_val - n_test + if min(n_train, n_val, n_test) <= 0: + raise ValueError( + "El dataset es demasiado chico para el split temporal configurado" + ) + train = df.iloc[:n_train].copy() + val = df.iloc[n_train : n_train + n_val].copy() + test = df.iloc[n_train + n_val :].copy() + return train, val, test + + +def _split_random(df: pd.DataFrame, cfg: ModelConfig) -> tuple[pd.DataFrame, ...]: + stratify = df["viaje_largo"] if cfg.stratify else None + train_val, test = train_test_split( + df, + test_size=cfg.test_size, + random_state=cfg.random_state, + stratify=stratify, + ) + val_fraction = cfg.validation_size / (1 - cfg.test_size) + stratify_train = train_val["viaje_largo"] if cfg.stratify else None + train, val = train_test_split( + train_val, + test_size=val_fraction, + random_state=cfg.random_state, + stratify=stratify_train, + ) + return train.copy(), val.copy(), test.copy() + + +def _apply_station_freq( + df: pd.DataFrame, station_freq: dict[int, int] +) -> pd.DataFrame: + out = df.copy() + out["start_station_freq"] = np.log1p( + out["start_station_id"].map(station_freq).fillna(0) + ) + return out + + +def _encode_value(value: str, encoder: LabelEncoder) -> int: + return encoder.transform([value])[0] if value in encoder.classes_ else -1 + + +def _transform_categorical( + df: pd.DataFrame, encoders: dict[str, LabelEncoder], columns: list[str] +) -> pd.DataFrame: + out = df.copy() + for col in columns: + encoder = encoders[col] + out[col] = out[col].astype(str).map( + lambda value, enc=encoder: _encode_value(value, enc) + ) + return out + + +def _elegir_umbral( + y_true: np.ndarray, y_prob: np.ndarray, metric: str = "f1" +) -> float: + if metric not in {"f1", "recall", "precision"}: + raise ValueError("metric debe ser f1, recall o precision") + + best_threshold = 0.5 + best_score = -1.0 + for threshold in np.linspace(0.01, 0.99, 99): + y_pred = (y_prob >= threshold).astype(int) + tn, fp, fn, tp = confusion_matrix(y_true, y_pred, labels=[0, 1]).ravel() + precision = tp / (tp + fp) if (tp + fp) else 0.0 + recall = tp / (tp + fn) if (tp + fn) else 0.0 + f1 = 2 * precision * recall / (precision + recall) if (precision + recall) else 0.0 + score = {"f1": f1, "recall": recall, "precision": precision}[metric] + if score > best_score: + best_score = score + best_threshold = float(threshold) + return best_threshold + + def entrenar(ruta_parquet, config: ModelConfig | None = None): cfg = config or load_config() os.makedirs(RUTA_ARTEFACTOS, exist_ok=True) - print(f"[1/7] Cargando datos desde: {ruta_parquet}") + print(f"[1/8] Cargando datos desde: {ruta_parquet}") df = pd.read_parquet(ruta_parquet) print(f" Shape: {df.shape}") - print("[2/7] Validando schema y calidad de datos") + print("[2/8] Validando schema y calidad de datos") validacion = validate_dataframe(df) - for w in validacion.warnings: - print(f" [WARN] {w}") + for warning in validacion.warnings: + print(f" [WARN] {warning}") validacion.raise_if_invalid() print(" Schema OK") umbral_s = cfg.trip_long_threshold_seconds - print(f"[3/7] Creando variable objetivo (viaje_largo > {umbral_s}s)") + print(f"[3/8] Creando variable objetivo (viaje_largo > {umbral_s}s)") df["viaje_largo"] = (df["tripduration"] > umbral_s).astype(int) print(f" Proporcion viajes largos: {df['viaje_largo'].mean():.3f}") - print("[4/7] Ingenieria de features") + print("[4/8] Ingenieria de features") df["starttime"] = pd.to_datetime(df["starttime"], errors="coerce") + df = df.dropna(subset=["starttime"]).copy() df["hour"] = df["starttime"].dt.hour df["month"] = df["starttime"].dt.month df["dayofweek"] = df["starttime"].dt.dayofweek df["is_weekend"] = df["dayofweek"].isin([5, 6]).astype(int) df["age"] = df["year"] - df["birth_year"] df = df[(df["age"] >= cfg.age_min) & (df["age"] <= cfg.age_max)].copy() + df = add_time_features(df) - station_freq = df["start_station_id"].value_counts().to_dict() - df["start_station_freq"] = np.log1p( - df["start_station_id"].map(station_freq).fillna(0) - ) + print(f"[5/8] Split {cfg.split_strategy} (train/val/test)") + if cfg.split_strategy == "temporal": + train, val, test = _split_temporal(df, cfg) + else: + train, val, test = _split_random(df, cfg) + print(f" Train: {len(train)} | Val: {len(val)} | Test: {len(test)}") - print("[5/7] Codificando variables categoricas") + station_freq = train["start_station_id"].value_counts().to_dict() + train = _apply_station_freq(train, station_freq) + val = _apply_station_freq(val, station_freq) + test = _apply_station_freq(test, station_freq) + + print("[6/8] Codificando variables categoricas") label_encoders = {} for col in cfg.categorical_features: - le = LabelEncoder() - df[col] = le.fit_transform(df[col].astype(str)) - label_encoders[col] = le - print(f" {col}: {list(le.classes_)}") + encoder = LabelEncoder() + train[col] = encoder.fit_transform(train[col].astype(str)) + label_encoders[col] = encoder + print(f" {col}: {list(encoder.classes_)}") + val = _transform_categorical(val, label_encoders, cfg.categorical_features) + test = _transform_categorical(test, label_encoders, cfg.categorical_features) feature_cols = cfg.feature_cols - X = df[feature_cols].astype(float) - y = df["viaje_largo"].values + X_train = train[feature_cols].astype(float) + X_val = val[feature_cols].astype(float) + X_test = test[feature_cols].astype(float) + y_train = train["viaje_largo"].values + y_val = val["viaje_largo"].values + y_test = test["viaje_largo"].values scaler = StandardScaler() - X_scaled = scaler.fit_transform(X) - - stratify = y if cfg.stratify else None - X_train, X_test, y_train, y_test = train_test_split( - X_scaled, y, test_size=cfg.test_size, random_state=cfg.random_state, stratify=stratify - ) - print(f" Train: {X_train.shape[0]} | Test: {X_test.shape[0]}") + X_train_s = scaler.fit_transform(X_train) + X_val_s = scaler.transform(X_val) + X_test_s = scaler.transform(X_test) - print(f"[6/7] Entrenando {cfg.model_class}") + print(f"[7/8] Entrenando {cfg.model_class}") modelo = _resolve_model(cfg) - modelo.fit(X_train, y_train) + modelo.fit(X_train_s, y_train) - y_prob = modelo.predict_proba(X_test)[:, 1] - y_pred = (y_prob >= cfg.decision_threshold).astype(int) + y_val_prob = modelo.predict_proba(X_val_s)[:, 1] + if cfg.tune_threshold: + decision_threshold = _elegir_umbral( + y_val, y_val_prob, metric=cfg.threshold_metric + ) + print( + f" Umbral optimizado ({cfg.threshold_metric}): " + f"{decision_threshold:.4f}" + ) + else: + decision_threshold = cfg.decision_threshold + print(f" Umbral fijo: {decision_threshold:.4f}") + + decision_threshold = round(decision_threshold, 4) + y_prob = modelo.predict_proba(X_test_s)[:, 1] + y_pred = (y_prob >= decision_threshold).astype(int) print("\n Classification Report:") separador = chr(10) + " " print( " " - + classification_report(y_test, y_pred, target_names=["corto", "largo"]) + + classification_report( + y_test, + y_pred, + labels=[0, 1], + target_names=["corto", "largo"], + zero_division=0, + ) .rstrip() .replace(chr(10), separador) ) roc = roc_auc_score(y_test, y_prob) + pr_auc = average_precision_score(y_test, y_prob) print(f" ROC-AUC: {roc:.4f}") - cm = confusion_matrix(y_test, y_pred) + print(f" PR-AUC: {pr_auc:.4f}") + cm = confusion_matrix(y_test, y_pred, labels=[0, 1]) print(f" Confusion Matrix:\n {cm}") - print(f"[7/7] Guardando artefactos en {RUTA_ARTEFACTOS}/") + print(f"[8/8] Guardando artefactos en {RUTA_ARTEFACTOS}/") joblib.dump(modelo, os.path.join(RUTA_ARTEFACTOS, "modelo.pkl")) joblib.dump( { @@ -109,15 +230,25 @@ def entrenar(ruta_parquet, config: ModelConfig | None = None): "scaler": scaler, "station_freq": station_freq, "feature_cols": feature_cols, - "decision_threshold": cfg.decision_threshold, + "decision_threshold": decision_threshold, + "threshold_metric": cfg.threshold_metric if cfg.tune_threshold else "fixed", + "split_strategy": cfg.split_strategy, }, os.path.join(RUTA_ARTEFACTOS, "transformers.pkl"), ) tn, fp, fn, tp = cm.ravel() - report = classification_report(y_test, y_pred, target_names=["corto", "largo"], output_dict=True) + report = classification_report( + y_test, + y_pred, + labels=[0, 1], + target_names=["corto", "largo"], + output_dict=True, + zero_division=0, + ) metricas = { "roc_auc": round(roc, 4), + "pr_auc": round(pr_auc, 4), "gini": round(2 * roc - 1, 4), "accuracy": round(report["accuracy"], 4), "recall": round(report["largo"]["recall"], 4), @@ -125,10 +256,13 @@ def entrenar(ruta_parquet, config: ModelConfig | None = None): "f1_score": round(report["largo"]["f1-score"], 4), "matriz_confusion": [[int(tn), int(fp)], [int(fn), int(tp)]], "umbral_largo_segundos": umbral_s, - "umbral_decision": cfg.decision_threshold, - "proporcion_largos": round(y.mean(), 3), - "train_size": X_train.shape[0], - "test_size": X_test.shape[0], + "umbral_decision": round(decision_threshold, 4), + "threshold_metric": cfg.threshold_metric if cfg.tune_threshold else "fixed", + "split_strategy": cfg.split_strategy, + "proporcion_largos": round(y_test.mean(), 3), + "train_size": len(train), + "validation_size": len(val), + "test_size": len(test), "features": feature_cols, "modelo": cfg.model_class, "model_params": cfg.model_params, @@ -145,14 +279,15 @@ def entrenar(ruta_parquet, config: ModelConfig | None = None): print(f"\nSubiendo artefactos a s3://{bucket}/{prefix}") upload_artifacts(RUTA_ARTEFACTOS, bucket, prefix) else: - print('\n[S3] S3_BUCKET_NAME no configurado, artefactos solo locales.') + print("\n[S3] S3_BUCKET_NAME no configurado, artefactos solo locales.") if __name__ == "__main__": parser = argparse.ArgumentParser(description="Entrenar clasificador Citibike") parser.add_argument("--input", required=True, help="Ruta al archivo .parquet") parser.add_argument( - "--config", default=None, + "--config", + default=None, help="Ruta al archivo YAML de configuracion (default: config/model.yaml o $MODEL_CONFIG)", ) args = parser.parse_args() diff --git a/requirements.txt b/requirements.txt index 6ff6f68..1007571 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,6 +1,6 @@ pandas>=2.0 pyarrow -scikit-learn>=1.3 +scikit-learn==1.9.0 joblib>=1.3 boto3>=1.34 numpy diff --git a/tests/test_config.py b/tests/test_config.py index 738beac..c455c76 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -9,9 +9,13 @@ def test_load_default_config(): assert isinstance(cfg, ModelConfig) assert cfg.trip_long_threshold_seconds == 900 assert cfg.decision_threshold == 0.65 + assert cfg.tune_threshold is True + assert cfg.threshold_metric == "f1" + assert cfg.split_strategy == "temporal" assert cfg.test_size == 0.2 + assert cfg.validation_size == 0.1 assert cfg.random_state == 42 - assert cfg.model_class == "sklearn.linear_model.LogisticRegression" + assert cfg.model_class == "sklearn.ensemble.HistGradientBoostingClassifier" assert cfg.model_params["class_weight"] == "balanced" @@ -20,6 +24,8 @@ def test_feature_cols_combines_numeric_and_categorical(): assert cfg.feature_cols == [*cfg.numeric_features, *cfg.categorical_features] assert "usertype" in cfg.feature_cols assert "hour" in cfg.feature_cols + assert "hour_sin" in cfg.feature_cols + assert "is_rush_hour" in cfg.feature_cols def test_missing_config_file_raises(tmp_path): @@ -34,8 +40,12 @@ def test_custom_config_override(tmp_path): target: trip_long_threshold_seconds: 600 decision_threshold: 0.5 + tune_threshold: false + threshold_metric: recall split: + strategy: random test_size: 0.3 + validation_size: 0.1 random_state: 7 stratify: false model: @@ -54,8 +64,69 @@ def test_custom_config_override(tmp_path): cfg = load_config(cfg_file) assert cfg.trip_long_threshold_seconds == 600 assert cfg.decision_threshold == 0.5 + assert cfg.tune_threshold is False + assert cfg.threshold_metric == "recall" + assert cfg.split_strategy == "random" assert cfg.test_size == 0.3 + assert cfg.validation_size == 0.1 assert cfg.random_state == 7 assert cfg.stratify is False assert cfg.age_min == 18 assert cfg.feature_cols == ["hour", "age", "usertype"] + + +def test_invalid_split_strategy_raises(tmp_path): + cfg_file = tmp_path / "invalid.yaml" + cfg_file.write_text( + """ +target: + trip_long_threshold_seconds: 900 + decision_threshold: 0.5 +split: + strategy: inventado + test_size: 0.2 + random_state: 42 + stratify: true +model: + class: sklearn.linear_model.LogisticRegression + params: {} +features: + numeric: [hour] + categorical: [usertype] +outliers: + age_min: 0 + age_max: 90 +""", + encoding="utf-8", + ) + with pytest.raises(ValueError, match="split.strategy"): + load_config(cfg_file) + + +def test_invalid_split_sizes_raises(tmp_path): + cfg_file = tmp_path / "invalid_sizes.yaml" + cfg_file.write_text( + """ +target: + trip_long_threshold_seconds: 900 + decision_threshold: 0.5 +split: + strategy: temporal + test_size: 0.7 + validation_size: 0.4 + random_state: 42 + stratify: true +model: + class: sklearn.linear_model.LogisticRegression + params: {} +features: + numeric: [hour] + categorical: [usertype] +outliers: + age_min: 0 + age_max: 90 +""", + encoding="utf-8", + ) + with pytest.raises(ValueError, match="test_size \\+ validation_size"): + load_config(cfg_file) diff --git a/tests/test_features.py b/tests/test_features.py new file mode 100644 index 0000000..6b409d4 --- /dev/null +++ b/tests/test_features.py @@ -0,0 +1,61 @@ +"""Tests para ml.features.""" +import numpy as np +import pandas as pd +import pytest + +from ml.features import add_time_features + + +def test_add_time_features_crea_columnas_y_flags(): + df = pd.DataFrame( + { + "hour": [0, 8, 17, 22], + "month": [1, 7, 12, 6], + "dayofweek": [0, 3, 5, 6], + } + ) + + out = add_time_features(df) + + assert { + "hour_sin", + "hour_cos", + "month_sin", + "month_cos", + "dayofweek_sin", + "dayofweek_cos", + "is_rush_hour", + "is_night", + } <= set(out.columns) + assert out.loc[0, "hour_sin"] == pytest.approx(0.0) + assert out.loc[0, "hour_cos"] == pytest.approx(1.0) + assert out.loc[0, "is_night"] == 1 + assert out.loc[0, "is_rush_hour"] == 0 + assert out.loc[1, "is_rush_hour"] == 1 + assert out.loc[2, "is_rush_hour"] == 1 + assert out.loc[3, "is_night"] == 1 + assert out.loc[0, "month_sin"] == pytest.approx(0.0) + assert out.loc[0, "month_cos"] == pytest.approx(1.0) + assert out.loc[0, "dayofweek_sin"] == pytest.approx(0.0) + assert out.loc[0, "dayofweek_cos"] == pytest.approx(1.0) + + +def test_add_time_features_no_muta_el_dataframe_original(): + df = pd.DataFrame({"hour": [8], "month": [7], "dayofweek": [3]}) + add_time_features(df) + assert "hour_sin" not in df.columns + + +def test_add_time_features_rangos_ciclicos(): + df = pd.DataFrame( + { + "hour": list(range(24)), + "month": [1] * 24, + "dayofweek": [0] * 24, + } + ) + out = add_time_features(df) + assert np.all(out["hour_sin"].between(-1, 1)) + assert np.all(out["hour_cos"].between(-1, 1)) + assert out["is_rush_hour"].isin([0, 1]).all() + assert out["is_night"].isin([0, 1]).all() diff --git a/tests/test_health.py b/tests/test_health.py index 4e39101..0eed7ac 100644 --- a/tests/test_health.py +++ b/tests/test_health.py @@ -10,7 +10,7 @@ def test_check_config_ok(): result = health.check_config() assert result["status"] == "ok" - assert result["model_class"] == "sklearn.linear_model.LogisticRegression" + assert result["model_class"] == "sklearn.ensemble.HistGradientBoostingClassifier" def test_check_artifacts_missing(tmp_path): diff --git a/tests/test_predict.py b/tests/test_predict.py index e073f11..c310e0b 100644 --- a/tests/test_predict.py +++ b/tests/test_predict.py @@ -6,6 +6,25 @@ from ml.predict import Predictor +FEATURE_COLS = [ + "hour", + "month", + "dayofweek", + "is_weekend", + "age", + "start_station_freq", + "hour_sin", + "hour_cos", + "month_sin", + "month_cos", + "dayofweek_sin", + "dayofweek_cos", + "is_rush_hour", + "is_night", + "usertype", + "gender", +] + @pytest.fixture def artefactos(tmp_path, monkeypatch): @@ -15,12 +34,7 @@ def artefactos(tmp_path, monkeypatch): artefactos_dir = tmp_path / "artefactos" artefactos_dir.mkdir() - # Modelo trivial con 8 features (igual que feature_cols). - feature_cols = [ - "hour", "month", "dayofweek", "is_weekend", "age", - "start_station_freq", "usertype", "gender", - ] - n_features = len(feature_cols) + n_features = len(FEATURE_COLS) modelo = LogisticRegression() rng = np.random.default_rng(42) X = rng.standard_normal((20, n_features)) @@ -36,10 +50,7 @@ def artefactos(tmp_path, monkeypatch): "label_encoders": {"usertype": le_usertype, "gender": le_gender}, "scaler": scaler, "station_freq": {3160: 10}, - "feature_cols": [ - "hour", "month", "dayofweek", "is_weekend", "age", - "start_station_freq", "usertype", "gender", - ], + "feature_cols": FEATURE_COLS, "decision_threshold": 0.65, }, artefactos_dir / "transformers.pkl", @@ -51,21 +62,30 @@ def artefactos(tmp_path, monkeypatch): return artefactos_dir +def _payload(**overrides): + datos = { + "start_station_id": 3160, + "usertype": "Customer", + "gender": "Male", + "hour": 17, + "month": 7, + "dayofweek": 3, + "is_weekend": 0, + "age": 32, + } + datos.update(overrides) + return datos + + def test_predict_returns_expected_shape(artefactos): p = Predictor() - resultado = p.predecir( - { - "start_station_id": 3160, - "usertype": "Customer", - "gender": "Male", - "hour": 17, - "month": 7, - "dayofweek": 3, - "is_weekend": 0, - "age": 32, - } - ) - assert set(resultado) == {"probabilidad_largo", "prediccion", "etiqueta"} + resultado = p.predecir(_payload()) + assert set(resultado) == { + "probabilidad_largo", + "prediccion", + "etiqueta", + "umbral_decision", + } assert resultado["prediccion"] in (0, 1) assert resultado["etiqueta"] in ("corto", "largo") @@ -73,21 +93,16 @@ def test_predict_returns_expected_shape(artefactos): def test_predict_unknown_category_maps_to_minus_one(artefactos): """Un valor categorico no visto en entrenamiento no rompe la prediccion.""" p = Predictor() - resultado = p.predecir( - { - "start_station_id": 9999, - "usertype": "Alien", - "gender": "Robot", - "hour": 8, - "month": 3, - "dayofweek": 1, - "is_weekend": 0, - "age": 40, - } - ) + resultado = p.predecir(_payload(usertype="Alien", gender="Robot")) assert resultado["prediccion"] in (0, 1) def test_predict_uses_decision_threshold(artefactos): p = Predictor() assert p.decision_threshold == 0.65 + + +def test_predict_exposes_decision_threshold(artefactos): + p = Predictor() + resultado = p.predecir(_payload()) + assert resultado["umbral_decision"] == 0.65 diff --git a/tests/test_train.py b/tests/test_train.py new file mode 100644 index 0000000..84600b5 --- /dev/null +++ b/tests/test_train.py @@ -0,0 +1,120 @@ +"""Tests para el pipeline de entrenamiento.""" +import json +from dataclasses import replace + +import joblib +import numpy as np +import pandas as pd +import pytest + +from ml import train +from ml.config import load_config + + +def _dataset(n: int = 120) -> pd.DataFrame: + starttime = pd.date_range("2018-01-01", periods=n, freq="h") + pares = np.arange(n) % 2 == 0 + return pd.DataFrame( + { + "tripduration": np.where(pares, 600, 1200), + "starttime": starttime, + "start_station_id": np.arange(n) % 3 + 1, + "usertype": np.where(pares, "Subscriber", "Customer"), + "gender": np.where(pares, "Male", "Female"), + "birth_year": np.where(pares, 1980, 1990), + "year": 2018, + "hour": starttime.hour, + "month": starttime.month, + "dayofweek": starttime.dayofweek, + "is_weekend": starttime.dayofweek.isin([5, 6]).astype(int), + "viaje_largo": (~pares).astype(int), + } + ) + + +def _config(): + base = load_config() + return replace( + base, + model_params={ + "class_weight": "balanced", + "max_iter": 10, + "learning_rate": 0.1, + "max_leaf_nodes": 5, + "min_samples_leaf": 2, + "early_stopping": False, + "random_state": 42, + }, + ) + + +def test_split_temporal_respeta_orden_y_tamanos(): + df = _dataset(100) + config = load_config() + + train_df, val_df, test_df = train._split_temporal(df, config) + + assert (len(train_df), len(val_df), len(test_df)) == (70, 10, 20) + assert train_df["starttime"].max() < val_df["starttime"].min() + assert val_df["starttime"].max() < test_df["starttime"].min() + + +def test_split_random_respeta_tamanos(): + df = _dataset(100) + config = replace(load_config(), split_strategy="random") + + train_df, val_df, test_df = train._split_random(df, config) + + assert (len(train_df), len(val_df), len(test_df)) == (70, 10, 20) + assert set(train_df.index).isdisjoint(val_df.index) + assert set(train_df.index).isdisjoint(test_df.index) + + +def test_elegir_umbral_por_metrica(): + y_true = np.array([0, 0, 0, 1, 1, 1]) + y_prob = np.array([0.1, 0.2, 0.3, 0.7, 0.8, 0.9]) + + umbral_f1 = train._elegir_umbral(y_true, y_prob, "f1") + umbral_recall = train._elegir_umbral(y_true, y_prob, "recall") + umbral_precision = train._elegir_umbral(y_true, y_prob, "precision") + + assert 0.3 < umbral_f1 < 0.7 + assert umbral_recall <= 0.1 + assert 0.3 < umbral_precision <= 0.4 + + +def test_elegir_umbral_metrica_invalida(): + with pytest.raises(ValueError, match="metric"): + train._elegir_umbral( + np.array([0, 1]), np.array([0.2, 0.8]), "accuracy" + ) + + +def test_entrenar_genera_artefactos(tmp_path, monkeypatch): + dataset_path = tmp_path / "datos.parquet" + _dataset().to_parquet(dataset_path, index=False) + artefactos_dir = tmp_path / "artefactos" + monkeypatch.setattr(train, "RUTA_ARTEFACTOS", str(artefactos_dir)) + monkeypatch.delenv("S3_BUCKET_NAME", raising=False) + + train.entrenar(dataset_path, _config()) + + for nombre in ("modelo.pkl", "transformers.pkl", "metricas.json"): + assert (artefactos_dir / nombre).exists() + + with (artefactos_dir / "metricas.json").open(encoding="utf-8") as f: + metricas = json.load(f) + transformers = joblib.load(artefactos_dir / "transformers.pkl") + modelo = joblib.load(artefactos_dir / "modelo.pkl") + + assert hasattr(modelo, "predict_proba") + assert metricas["split_strategy"] == "temporal" + assert metricas["train_size"] == 84 + assert metricas["validation_size"] == 12 + assert metricas["test_size"] == 24 + assert metricas["modelo"] == "sklearn.ensemble.HistGradientBoostingClassifier" + assert "pr_auc" in metricas + assert "hour_sin" in metricas["features"] + assert 0 <= metricas["umbral_decision"] <= 1 + assert transformers["decision_threshold"] == metricas["umbral_decision"] + assert transformers["feature_cols"] == metricas["features"] From 70be5cf3e7e1ed92703462173357911d8f07b6d5 Mon Sep 17 00:00:00 2001 From: Sebithaz-dev Date: Tue, 27 Oct 2026 18:43:05 -0300 Subject: [PATCH 2/3] fix(config): reemplaza comparaciones encadenadas para Sonar --- ml/config.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/ml/config.py b/ml/config.py index c1c665c..b66bdfd 100644 --- a/ml/config.py +++ b/ml/config.py @@ -81,7 +81,7 @@ def load_config(path: str | os.PathLike | None = None) -> ModelConfig: def _validate_config(cfg: ModelConfig) -> None: - if not 0 <= cfg.decision_threshold <= 1: + if cfg.decision_threshold < 0 or cfg.decision_threshold > 1: raise ValueError("target.decision_threshold debe estar entre 0 y 1") if cfg.threshold_metric not in {"f1", "recall", "precision"}: raise ValueError( @@ -89,9 +89,9 @@ def _validate_config(cfg: ModelConfig) -> None: ) if cfg.split_strategy not in {"temporal", "random"}: raise ValueError("split.strategy debe ser temporal o random") - if not 0 < cfg.test_size < 1: + if cfg.test_size <= 0 or cfg.test_size >= 1: raise ValueError("split.test_size debe estar entre 0 y 1") - if not 0 < cfg.validation_size < 1: + if cfg.validation_size <= 0 or cfg.validation_size >= 1: raise ValueError("split.validation_size debe estar entre 0 y 1") if cfg.test_size + cfg.validation_size >= 1: raise ValueError("test_size + validation_size debe ser menor que 1") From be87994fe2249b4a03027ed3b6ddfa39e15af398 Mon Sep 17 00:00:00 2001 From: Sebithaz-dev Date: Tue, 27 Oct 2026 18:45:28 -0300 Subject: [PATCH 3/3] fix(config): mueve validacion de fracciones a helper --- ml/config.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/ml/config.py b/ml/config.py index b66bdfd..b4ccc32 100644 --- a/ml/config.py +++ b/ml/config.py @@ -80,6 +80,11 @@ def load_config(path: str | os.PathLike | None = None) -> ModelConfig: return cfg +def _ensure_fraction(value: float, name: str) -> None: + if value <= 0 or value >= 1: + raise ValueError(f"{name} debe estar entre 0 y 1") + + def _validate_config(cfg: ModelConfig) -> None: if cfg.decision_threshold < 0 or cfg.decision_threshold > 1: raise ValueError("target.decision_threshold debe estar entre 0 y 1") @@ -89,9 +94,7 @@ def _validate_config(cfg: ModelConfig) -> None: ) if cfg.split_strategy not in {"temporal", "random"}: raise ValueError("split.strategy debe ser temporal o random") - if cfg.test_size <= 0 or cfg.test_size >= 1: - raise ValueError("split.test_size debe estar entre 0 y 1") - if cfg.validation_size <= 0 or cfg.validation_size >= 1: - raise ValueError("split.validation_size debe estar entre 0 y 1") + _ensure_fraction(cfg.test_size, "split.test_size") + _ensure_fraction(cfg.validation_size, "split.validation_size") if cfg.test_size + cfg.validation_size >= 1: raise ValueError("test_size + validation_size debe ser menor que 1")