from __future__ import annotations import logging import time from dataclasses import dataclass from typing import Any import numpy as np import pandas as pd from sklearn.base import BaseEstimator from sklearn.compose import ColumnTransformer from sklearn.ensemble import ( ExtraTreesClassifier, ExtraTreesRegressor, RandomForestClassifier, RandomForestRegressor, ) from sklearn.impute import SimpleImputer from sklearn.inspection import permutation_importance from sklearn.linear_model import LinearRegression, LogisticRegression from sklearn.metrics import ( accuracy_score, balanced_accuracy_score, f1_score, mean_absolute_error, mean_squared_error, r2_score, roc_auc_score, ) from sklearn.model_selection import cross_val_score, train_test_split from sklearn.pipeline import Pipeline from sklearn.preprocessing import OneHotEncoder, StandardScaler from datapilot.config import Settings from datapilot.schemas import ( CriticDecision, ExplainabilityResult, ModelFailure, ModelResult, TaskType, ) from datapilot.tuning import tune_random_forest @dataclass class TrainingBundle: pipeline: Pipeline results: list[ModelResult] best_model: str test_features: pd.DataFrame test_target: pd.Series retry_number: int failures: list[ModelFailure] logger = logging.getLogger(__name__) def _preprocessor(features: pd.DataFrame, settings: Settings) -> ColumnTransformer: numeric = features.select_dtypes(include=np.number).columns.tolist() categorical = [column for column in features.columns if column not in numeric] numeric_pipeline = Pipeline( [ ("imputer", SimpleImputer(strategy="median")), ("scaler", StandardScaler()), ] ) categorical_pipeline = Pipeline( [ ("imputer", SimpleImputer(strategy="most_frequent")), ( "encoder", OneHotEncoder( handle_unknown="infrequent_if_exist", min_frequency=2, max_categories=settings.max_categories_per_feature, sparse_output=True, ), ), ] ) return ColumnTransformer( [ ("numeric", numeric_pipeline, numeric), ("categorical", categorical_pipeline, categorical), ], remainder="drop", verbose_feature_names_out=False, ) def _candidate_models(task: TaskType, settings: Settings, retry: int) -> dict[str, BaseEstimator]: if task == TaskType.classification: models: dict[str, BaseEstimator] = { "Logistic Regression": LogisticRegression( max_iter=1_000, class_weight="balanced", random_state=settings.random_state ), "Random Forest": RandomForestClassifier( n_estimators=180 + retry * 80, min_samples_leaf=max(1, 2 - retry), class_weight="balanced", n_jobs=-1, random_state=settings.random_state, ), "Extra Trees": ExtraTreesClassifier( n_estimators=180 + retry * 80, class_weight="balanced", n_jobs=-1, random_state=settings.random_state, ), } else: models = { "Linear Regression": LinearRegression(), "Random Forest": RandomForestRegressor( n_estimators=180 + retry * 80, min_samples_leaf=max(1, 2 - retry), n_jobs=-1, random_state=settings.random_state, ), "Extra Trees": ExtraTreesRegressor( n_estimators=180 + retry * 80, n_jobs=-1, random_state=settings.random_state, ), } try: if task == TaskType.classification: from xgboost import XGBClassifier models["XGBoost"] = XGBClassifier( n_estimators=160 + retry * 60, max_depth=4 + retry, learning_rate=0.07, eval_metric="logloss", n_jobs=-1, random_state=settings.random_state, ) else: from xgboost import XGBRegressor models["XGBoost"] = XGBRegressor( n_estimators=160 + retry * 60, max_depth=4 + retry, learning_rate=0.07, n_jobs=-1, random_state=settings.random_state, ) except ImportError: pass return models def train_models( frame: pd.DataFrame, target: str, task: TaskType, settings: Settings, retry_number: int = 0, ) -> TrainingBundle: if target not in frame.columns: raise ValueError(f"Target column '{target}' does not exist.") clean = frame.dropna(subset=[target]).drop_duplicates().reset_index(drop=True) features = clean.drop(columns=[target]).copy() labels = clean[target].copy() if features.empty: raise ValueError("No usable feature columns remain after removing the target.") if labels.nunique(dropna=True) < 2: raise ValueError("The target must contain at least two distinct non-null values.") categorical = features.select_dtypes(exclude=np.number) estimated_width = len(features.select_dtypes(include=np.number).columns) + sum( min(int(categorical[column].nunique(dropna=True)), settings.max_categories_per_feature) for column in categorical.columns ) if estimated_width > settings.max_encoded_features: raise ValueError( f"Estimated encoded width {estimated_width:,} exceeds the safe limit of " f"{settings.max_encoded_features:,}. Reduce high-cardinality columns or increase " "MAX_ENCODED_FEATURES after reviewing memory capacity." ) stratify = ( labels if task == TaskType.classification and labels.value_counts().min() >= 2 else None ) x_train, x_test, y_train, y_test = train_test_split( features, labels, test_size=settings.test_size, random_state=settings.random_state, stratify=stratify, ) results: list[ModelResult] = [] failures: list[ModelFailure] = [] cv = min(5, max(2, int(len(x_train) / 20))) if task == TaskType.classification: smallest_class = int(y_train.value_counts().min()) cv = min(cv, smallest_class) if smallest_class >= 2 else 0 scoring = "balanced_accuracy" else: scoring = "r2" if cv < 2: raise ValueError("The training partition is too small for reliable cross-validation.") candidates = _candidate_models(task, settings, retry_number) tuned_parameters = tune_random_forest( task, _preprocessor(x_train, settings), x_train, y_train, settings, cv, ) if tuned_parameters: candidates["Random Forest"].set_params(**tuned_parameters) for name, estimator in candidates.items(): pipeline = Pipeline( [("preprocessor", _preprocessor(x_train, settings)), ("model", estimator)] ) started = time.perf_counter() try: cv_scores = ( cross_val_score(pipeline, x_train, y_train, scoring=scoring, cv=cv, n_jobs=1) if cv >= 2 else np.array([]) ) result = ModelResult( name=name, primary_metric="balanced_accuracy" if task == TaskType.classification else "r2", primary_score=float(cv_scores.mean()), metrics={}, cross_validation_mean=float(cv_scores.mean()) if len(cv_scores) else None, cross_validation_std=float(cv_scores.std()) if len(cv_scores) else None, training_seconds=round(time.perf_counter() - started, 3), selection_score=float(cv_scores.mean()) if len(cv_scores) else None, ) results.append(result) except Exception as exc: logger.warning( "Candidate %s failed during cross-validation: %s", name, type(exc).__name__ ) failures.append( ModelFailure( name=name, stage="cross_validation", exception_category=type(exc).__name__, sanitized_error="Candidate failed during cross-validation; inspect structured logs.", training_seconds=round(time.perf_counter() - started, 3), expected=isinstance(exc, (ValueError, TypeError)), ) ) if not results: raise RuntimeError("Every candidate model failed; review the data types and target.") results.sort(key=lambda item: item.selection_score or float("-inf"), reverse=True) best = results[0] final_pipeline = Pipeline( [ ("preprocessor", _preprocessor(x_train, settings)), ("model", candidates[best.name]), ] ) final_pipeline.fit(x_train, y_train) predictions = final_pipeline.predict(x_test) final_metrics = _metrics(task, y_test, predictions, final_pipeline, x_test) final_score = ( final_metrics["balanced_accuracy"] if task == TaskType.classification else final_metrics["r2"] ) best.final_test_score = float(final_score) best.final_test_metrics = final_metrics best.metrics = final_metrics return TrainingBundle( pipeline=final_pipeline, results=results, best_model=best.name, test_features=x_test, test_target=y_test, retry_number=retry_number, failures=failures, ) def _metrics( task: TaskType, truth: pd.Series, predictions: np.ndarray, pipeline: Pipeline, features: pd.DataFrame, ) -> dict[str, float]: if task == TaskType.classification: metrics = { "accuracy": round(float(accuracy_score(truth, predictions)), 4), "balanced_accuracy": round(float(balanced_accuracy_score(truth, predictions)), 4), "f1_weighted": round(float(f1_score(truth, predictions, average="weighted")), 4), } if truth.nunique() == 2 and hasattr(pipeline, "predict_proba"): probabilities = pipeline.predict_proba(features)[:, 1] metrics["roc_auc"] = round(float(roc_auc_score(truth, probabilities)), 4) return metrics return { "r2": round(float(r2_score(truth, predictions)), 4), "rmse": round(float(mean_squared_error(truth, predictions) ** 0.5), 4), "mae": round(float(mean_absolute_error(truth, predictions)), 4), } def critic_decision(bundle: TrainingBundle, task: TaskType, settings: Settings) -> CriticDecision: best = bundle.results[0] threshold = ( settings.min_classification_score if task == TaskType.classification else settings.min_regression_score ) reasons: list[str] = [] quality_score = best.cross_validation_mean if best.cross_validation_mean is not None else -1.0 if quality_score < threshold: reasons.append( f"Training CV {best.primary_metric} {quality_score:.3f} is below {threshold:.3f}." ) if ( best.cross_validation_mean is not None and best.final_test_score is not None and abs(best.final_test_score - best.cross_validation_mean) > 0.2 ): reasons.append("Holdout and cross-validation scores diverge by more than 0.20.") approved = not reasons or bundle.retry_number >= settings.max_critic_retries if not reasons: reasons.append("Performance and validation consistency passed the configured quality gate.") elif approved: reasons.append("Retry budget exhausted; result is retained with an explicit limitation.") return CriticDecision( approved=approved, score=quality_score, threshold=threshold, reasons=reasons, retry_number=bundle.retry_number, ) def explain_model(bundle: TrainingBundle) -> ExplainabilityResult: sample_size = min(300, len(bundle.test_features)) features = bundle.test_features.iloc[:sample_size] target = bundle.test_target.iloc[:sample_size] try: transformed = bundle.pipeline.named_steps["preprocessor"].transform(features) transformed_names = bundle.pipeline.named_steps["preprocessor"].get_feature_names_out() estimator = bundle.pipeline.named_steps["model"] import shap explainer = shap.Explainer(estimator, transformed) values = explainer(transformed) raw = np.asarray(values.values) if raw.ndim == 3: raw = np.abs(raw).mean(axis=(0, 2)) else: raw = np.abs(raw).mean(axis=0) importance = _top_importance(transformed_names, raw) return ExplainabilityResult( method="SHAP", feature_importance=importance, caveats=["SHAP values explain this fitted model, not causal effects."], ) except Exception: permutation = permutation_importance( bundle.pipeline, features, target, n_repeats=5, random_state=42, n_jobs=1, ) importance = _top_importance(features.columns, np.abs(permutation.importances_mean)) return ExplainabilityResult( method="Permutation importance", feature_importance=importance, caveats=[ "Permutation importance can dilute importance among correlated features.", "Feature importance is predictive, not causal.", ], ) def _top_importance(names: Any, values: np.ndarray, limit: int = 15) -> dict[str, float]: pairs = sorted( zip([str(name) for name in names], values.tolist(), strict=False), key=lambda item: item[1], reverse=True, )[:limit] return {name: round(float(value), 6) for name, value in pairs}