Source code for ml_grid.pipeline.grid_search_cross_validate

import time
import logging
import multiprocessing
import joblib
import warnings
import os
import sys
from typing import Any, Dict, List, Optional, Union

import numpy as np
import pandas as pd
import tensorflow as tf

# Lazy import torch only when needed to handle ImportError gracefully
try:
    import torch
except ImportError:
[docs] torch = None
from IPython.display import clear_output from scikeras.wrappers import KerasClassifier import sklearn from sklearn import metrics from pandas.testing import assert_index_equal from xgboost.core import XGBoostError from ml_grid.model_classes.AutoKerasClassifierWrapper import AutoKerasClassifierWrapper from ml_grid.model_classes.H2OAutoMLClassifier import H2OAutoMLClassifier from ml_grid.model_classes.FLAMLClassifierWrapper import FLAMLClassifierWrapper from ml_grid.model_classes.H2OGBMClassifier import H2OGBMClassifier from ml_grid.model_classes.H2ODRFClassifier import H2ODRFClassifier from ml_grid.model_classes.H2OGAMClassifier import H2OGAMClassifier from ml_grid.model_classes.H2ODeepLearningClassifier import H2ODeepLearningClassifier from ml_grid.model_classes.H2OGLMClassifier import H2OGLMClassifier from ml_grid.model_classes.H2ONaiveBayesClassifier import H2ONaiveBayesClassifier from ml_grid.model_classes.H2ORuleFitClassifier import H2ORuleFitClassifier from ml_grid.model_classes.H2OXGBoostClassifier import H2OXGBoostClassifier from ml_grid.model_classes.H2OStackedEnsembleClassifier import ( H2OStackedEnsembleClassifier, ) from ml_grid.model_classes.NeuralNetworkKerasClassifier import NeuralNetworkClassifier # from sklearn.utils.testing import ignore_warnings from sklearn.exceptions import ConvergenceWarning from sklearn.model_selection import ( ParameterGrid, RepeatedKFold, KFold, cross_validate, ) from ml_grid.model_classes.keras_classifier_class import KerasClassifierClass from ml_grid.pipeline.hyperparameter_search import HyperparameterSearch from ml_grid.util.debug_print_statements import debug_print_statements_class from ml_grid.util.global_params import global_parameters from ml_grid.util.project_score_save import project_score_save_class from ml_grid.util.validate_parameters import validate_parameters_helper from sklearn.preprocessing import MinMaxScaler, StandardScaler from ml_grid.util.bayes_utils import is_skopt_space from skopt.space import Categorical # Global flag to ensure TensorFlow/GPU setup runs only once per process _TF_INITIALIZED = False # Define H2O model types at module level for reuse
[docs] H2O_MODEL_TYPES = ( H2OAutoMLClassifier, H2OGBMClassifier, H2ODRFClassifier, H2OGAMClassifier, H2ODeepLearningClassifier, H2OGLMClassifier, H2ONaiveBayesClassifier, H2ORuleFitClassifier, H2OXGBoostClassifier, H2OStackedEnsembleClassifier, )
# Disable TF Traceback Filtering to reduce overhead in Keras model building try: tf.debugging.disable_traceback_filtering() except (AttributeError, ImportError): pass
[docs] def is_sklearn_version_supported(): """Check if scikit-learn version supports required CV features.""" from packaging import version try: sklearn_version = sklearn.__version__ return version.parse(sklearn_version) >= version.parse("0.24") except Exception: return True # Assume supported if version check fails
[docs] class grid_search_crossvalidate: def __init__( self, algorithm_implementation: Any, parameter_space: Union[Dict, List[Dict]], method_name: str, ml_grid_object: Any, sub_sample_parameter_val: int = 100, project_score_save_class_instance: Optional[project_score_save_class] = None, ): """Initializes and runs a cross-validated hyperparameter search. Performs grid, randomized, or Bayesian search for hyperparameters and logs results. Args: algorithm_implementation (Any): The scikit-learn compatible estimator instance. parameter_space (Union[Dict, List[Dict]]): The dictionary or list of dictionaries defining the hyperparameter search space. method_name (str): The name of the algorithm method. ml_grid_object (Any): The main pipeline object containing all data (X_train, y_train, etc.) and parameters for the current iteration. sub_sample_parameter_val (int, optional): A value used to limit the number of iterations in a randomized search. Defaults to 100. project_score_save_class_instance (Optional[project_score_save_class], optional): An instance of the score saving class. Defaults to None. """ # Set warning filters warnings.filterwarnings("ignore", category=UserWarning) warnings.filterwarnings("ignore", category=ConvergenceWarning) warnings.filterwarnings("ignore", category=FutureWarning)
[docs] self.logger = logging.getLogger("ml_grid")
[docs] self.global_params = global_parameters
[docs] self.verbose = self.global_params.verbose
[docs] self.method_name = method_name # Store for use in CV optimizations
if self.verbose < 8: self.logger.debug("Clearing output.") clear_output(wait=True)
[docs] self.project_score_save_class_instance = project_score_save_class_instance
[docs] self.sub_sample_param_space_pct = self.global_params.sub_sample_param_space_pct
random_grid_search = self.global_params.random_grid_search
[docs] self.sub_sample_parameter_val = sub_sample_parameter_val
# Detect nested parallelism: force n_jobs=1 if running inside a worker process if multiprocessing.current_process().daemon: self.global_params.grid_n_jobs = 1 grid_n_jobs = 1 else: grid_n_jobs = self.global_params.grid_n_jobs # Configure GPU usage and job limits for specific models is_gpu_model = ( "keras" in method_name.lower() or "xgb" in method_name.lower() or "catboost" in method_name.lower() or "neural" in method_name.lower() ) is_h2o_model = isinstance(algorithm_implementation, H2O_MODEL_TYPES) global _TF_INITIALIZED if is_gpu_model or is_h2o_model: grid_n_jobs = 1 # Disable H2O Progress Bar to save time if is_h2o_model: try: import h2o h2o.no_progress() except ImportError: pass except Exception: pass # One-time TF/GPU Setup if is_gpu_model and not _TF_INITIALIZED: try: # --- FIX for libdevice error --- # Set XLA_FLAGS to point to the CUDA toolkit installed by pip. # This is crucial for XLA to find the libdevice library for GPU compilation. if "XLA_FLAGS" not in os.environ: # Find site-packages directory site_packages_path = next( (p for p in sys.path if "site-packages" in p), None ) if site_packages_path: # The 'nvidia-cuda-nvcc-cu12' package installs the compiler toolkit here. # XLA needs this path to find the 'nvvm/libdevice' directory. cuda_path = os.path.join( site_packages_path, "nvidia", "cuda_nvcc" ) if os.path.exists(cuda_path): self.logger.info( f"Found CUDA compiler toolkit at {cuda_path}. Setting XLA_FLAGS." ) os.environ["XLA_FLAGS"] = ( f"--xla_gpu_cuda_data_dir={cuda_path}" ) else: self.logger.warning( "Could not find 'nvidia/cuda_nvcc' directory. Falling back to site-packages root. " "Install 'nvidia-cuda-nvcc-cu12' for a reliable setup." ) gpu_devices = tf.config.experimental.list_physical_devices("GPU") if gpu_devices: for device in gpu_devices: try: tf.config.experimental.set_memory_growth(device, True) except RuntimeError: pass else: # Explicitly set CPU as the visible device for TensorFlow to avoid CUDA init errors tf.config.set_visible_devices([], "GPU") tf.config.run_functions_eagerly(False) except Exception as e: self.logger.warning(f"Could not configure GPU for TensorFlow: {e}") finally: _TF_INITIALIZED = True
[docs] self.metric_list = self.global_params.metric_list
[docs] self.error_raise = self.global_params.error_raise
if self.verbose >= 3: self.logger.info(f"Cross-validating {method_name}")
[docs] self.global_parameters = global_parameters
[docs] self.ml_grid_object_iter = ml_grid_object
[docs] self.X_train = self.ml_grid_object_iter.X_train
[docs] self.y_train = self.ml_grid_object_iter.y_train
[docs] self.X_test = self.ml_grid_object_iter.X_test
[docs] self.y_test = self.ml_grid_object_iter.y_test
[docs] self.X_test_orig = self.ml_grid_object_iter.X_test_orig
[docs] self.y_test_orig = self.ml_grid_object_iter.y_test_orig
# Ensure X_train is a DataFrame and always normalize column names to strings # This matches the behavior for test data below, ensuring consistent feature naming if not isinstance(self.X_train, pd.DataFrame): self.X_train = pd.DataFrame(self.X_train) self.X_train = self.X_train.rename(columns=str) # Ensure y_train is a Series with aligned index if not isinstance(self.y_train, (pd.Series, pd.DataFrame)): self.y_train = pd.Series(self.y_train, index=self.X_train.index) # Enforce consistent target variable name for H2O compatibility self.y_train.name = "outcome" # Ensure test data types match training data (DataFrames/Series with aligned indices) # This prevents crashes in SVC branch that assumes .columns attribute on X_test if not isinstance(self.X_test, pd.DataFrame): self.X_test = pd.DataFrame(self.X_test).rename(columns=str) else: self.X_test = self.X_test.rename(columns=str) if not isinstance(self.y_test, (pd.Series, pd.DataFrame)): self.y_test = pd.Series(self.y_test, index=self.X_test.index) if not isinstance(self.X_test_orig, pd.DataFrame): self.X_test_orig = pd.DataFrame(self.X_test_orig).rename(columns=str) else: self.X_test_orig = self.X_test_orig.rename(columns=str) if not isinstance(self.y_test_orig, (pd.Series, pd.DataFrame)): self.y_test_orig = pd.Series(self.y_test_orig, index=self.X_test_orig.index) # Drop ID column if present if "client_idcode" in self.X_train.columns: self.logger.debug("Dropping 'client_idcode' from training data.") self.X_train = self.X_train.drop(columns=["client_idcode"], errors="ignore") if isinstance(self.X_test, pd.DataFrame): self.X_test = self.X_test.drop( columns=["client_idcode"], errors="ignore" ) if isinstance(self.X_test_orig, pd.DataFrame): self.X_test_orig = self.X_test_orig.drop( columns=["client_idcode"], errors="ignore" ) max_param_space_iter_value = ( # hard limit on param space exploration self.global_params.max_param_space_iter_value ) # Allow local override for max_param_space_iter_value if ( self.ml_grid_object_iter.local_param_dict.get("max_param_space_iter_value") is not None ): max_param_space_iter_value = self.ml_grid_object_iter.local_param_dict.get( "max_param_space_iter_value" ) if "svc" in method_name.lower(): self.logger.info( "Applying StandardScaler for SVC to prevent convergence issues." ) scaler = StandardScaler() self.X_train = pd.DataFrame( scaler.fit_transform(self.X_train), columns=self.X_train.columns, index=self.X_train.index, ) self.X_test = pd.DataFrame( scaler.transform(self.X_test), columns=self.X_test.columns, index=self.X_test.index, ) self.X_test_orig = pd.DataFrame( scaler.transform(self.X_test_orig), columns=self.X_test_orig.columns, index=self.X_test_orig.index, ) # Optimize y_test and y_test_orig to reduce metric calculation overhead self.y_test = self._optimize_y(self.y_test) self.y_test_orig = self._optimize_y(self.y_test_orig) # Track training set size for adaptive CV / parameter adjustments self._n_samples_train = len(self.X_train) # Use faster CV strategy in test mode if getattr(self.global_parameters, "test_mode", False): self.logger.info("Test mode enabled. Using fast KFold(n_splits=2) for CV.") self.cv = KFold(n_splits=2, shuffle=True, random_state=1) else: # Allow explicit override to the simple RepeatedKFold strategy if desired. if getattr(self.global_params, "cv_strategy", None) == "repeated": self.logger.debug("Using RepeatedKFold (cv_strategy='repeated')") self.cv = RepeatedKFold(n_splits=2, n_repeats=2, random_state=1) else: # Adaptive CV strategy based on data characteristics. NOTE: to keep # cross-algorithm score comparisons fair, this strategy is a function # of the dataset only (n_samples, class balance) and NOT of the # algorithm being evaluated, so every algorithm run against a given # dataset in this pipeline gets the same fold structure. cv_strategy = self._get_adaptive_cv_strategy() self.cv = cv_strategy if hasattr(cv_strategy, "get_n_splits"): try: n_splits = cv_strategy.get_n_splits() except TypeError: n_splits = getattr(cv_strategy, "n_splits", "N/A") self.logger.info( f"Using adaptive CV: {type(cv_strategy).__name__}(n_splits={n_splits}) " "for production runs" ) start = time.time() current_algorithm = algorithm_implementation # Silence verbose models to keep logs clean and reduce I/O overhead if hasattr(current_algorithm, "set_params"): if "catboost" in method_name.lower(): ml_grid_object.logger.info( "Silencing CatBoost verbose output and file writing." ) current_algorithm.set_params(verbose=0, allow_writing_files=False) elif "xgb" in method_name.lower(): ml_grid_object.logger.info("Silencing XGBoost verbose output.") current_algorithm.set_params(verbosity=0) elif "keras" in method_name.lower() or "neural" in method_name.lower(): ml_grid_object.logger.info("Silencing Keras verbose output.") try: current_algorithm.set_params(verbose=0) except Exception: pass # Check for GPU availability and set device for torch-based models if "simbsig" in str(type(algorithm_implementation)): if torch is None: self.logger.warning( "torch not installed, cannot configure GPU for simbsig" ) elif not torch.cuda.is_available(): self.logger.info( "No CUDA GPU detected. Forcing simbsig model to use CPU." ) if hasattr(current_algorithm, "set_params"): current_algorithm.set_params(device="cpu") else: self.logger.info( "CUDA GPU detected. Allowing simbsig model to use GPU." ) self.logger.debug(f"Algorithm implementation: {algorithm_implementation}") parameters = parameter_space # Keep a reference to the original if ml_grid_object.verbose >= 3: self.logger.debug( f"algorithm_implementation: {algorithm_implementation}, type: {type(algorithm_implementation)}" ) # Validate parameters if not self.global_params.bayessearch: parameters = validate_parameters_helper( algorithm_implementation=algorithm_implementation, parameters=parameter_space, ml_grid_object=ml_grid_object, ) # Ensure list-based parameters are wrapped in Categorical for Bayesian search if self.global_params.bayessearch: self.logger.debug("Validating parameter space for Bayesian search...") def _is_simple_categorical(val): """ Determines if a value is a list of simple, hashable choices suitable for wrapping in skopt.space.Categorical. """ # A list is only considered a categorical choice if it has more than one item. # Single-item lists are treated as fixed parameters by BayesSearchCV. if not isinstance(val, (list, np.ndarray)) or len(val) <= 1: return False try: for item in val: hash(item) return True except TypeError: return False if isinstance(parameter_space, list): for i, space in enumerate(parameter_space): new_space = {} for key, value in space.items(): if _is_simple_categorical(value) and not is_skopt_space(value): self.logger.info( f"Auto-correcting param '{key}' for BayesSearch: wrapping list in Categorical." ) new_space[key] = Categorical(value) else: new_space[key] = value parameter_space[i] = new_space elif isinstance(parameter_space, dict): new_parameter_space = {} for key, value in parameter_space.items(): if _is_simple_categorical(value) and not is_skopt_space(value): self.logger.info( f"Auto-correcting param '{key}' for BayesSearch: wrapping list in Categorical." ) new_parameter_space[key] = Categorical(value) else: new_parameter_space[key] = value parameter_space = new_parameter_space parameters = parameter_space # Determine n_iter try: n_iter_v = getattr(self.global_params, "n_iter", 2) if n_iter_v is None: n_iter_v = 2 n_iter_v = int(n_iter_v) except (ValueError, TypeError): self.logger.warning( "Invalid or missing n_iter in global_params. Defaulting to 2." ) n_iter_v = 2 # Allow local override from run_params/local_param_dict local_n_iter = self.ml_grid_object_iter.local_param_dict.get("n_iter") if local_n_iter is not None: try: n_iter_v = int(local_n_iter) self.logger.info( f"Overriding global n_iter with local value: {n_iter_v}" ) except (ValueError, TypeError): self.logger.warning( f"Invalid local n_iter value: {local_n_iter}. Ignoring override." ) if max_param_space_iter_value is not None: if n_iter_v > max_param_space_iter_value: self.logger.info( f"Capping n_iter ({n_iter_v}) to max_param_space_iter_value ({max_param_space_iter_value})" ) n_iter_v = max_param_space_iter_value # Log grid size or iterations # Check for skopt space to avoid ParameterGrid errors is_bayes_space = False if isinstance(parameter_space, list): for space in parameter_space: if isinstance(space, dict) and any( is_skopt_space(v) for v in space.values() ): is_bayes_space = True break elif isinstance(parameter_space, dict): if any(is_skopt_space(v) for v in parameter_space.values()): is_bayes_space = True if ( not self.global_params.bayessearch and not random_grid_search and not is_bayes_space ): try: pg = len(ParameterGrid(parameter_space)) self.logger.info(f"Parameter grid size: {pg}") except TypeError: self.logger.warning( "Could not calculate ParameterGrid size (likely skopt objects)." ) pg = "N/A" else: self.logger.info(f"Using n_iter={n_iter_v} for search.") pg = "N/A" # Dynamically adjust KNN parameter space for small datasets if "kneighbors" in method_name.lower() or "simbsig" in method_name.lower(): self._adjust_knn_parameters(parameter_space) self.logger.debug( "Adjusted KNN n_neighbors parameter space to prevent errors on small CV folds." ) # Check if dataset is too small for CatBoost if "catboost" in method_name.lower(): min_samples_required = 10 # CatBoost needs a reasonable amount of data if len(self.X_train) < min_samples_required: self.logger.warning( f"Dataset too small for CatBoost ({len(self.X_train)} samples < {min_samples_required} required). " f"Skipping {method_name}." ) # Return early with default scores self.grid_search_cross_validate_score_result = 0.5 return # Adjust CatBoost parameters for small datasets if "catboost" in method_name.lower(): self._adjust_catboost_parameters(parameter_space) self.logger.debug( "Adjusted CatBoost subsample parameter space to prevent errors on small CV folds." ) # Adjust XGBoost parameters for small datasets if "xgb" in method_name.lower(): self._adjust_xgboost_parameters(parameter_space) self.logger.debug( "Adjusted XGBoost max_bin parameter space to prevent errors on small CV folds." ) # Force sequential search for H2O/GPU models original_grid_n_jobs = self.global_parameters.grid_n_jobs if is_gpu_model or is_h2o_model: self.global_parameters.grid_n_jobs = 1 try: # Instantiate and run the hyperparameter grid/random search search = HyperparameterSearch( algorithm=current_algorithm, parameter_space=parameters, # Use the validated/modified parameters method_name=method_name, global_params=self.global_parameters, sub_sample_pct=self.sub_sample_param_space_pct, # Explore 50% of the parameter space max_iter=n_iter_v, # Maximum iterations for randomized search ml_grid_object=ml_grid_object, cv=self.cv, ) if self.global_parameters.verbose >= 3: self.logger.debug("Running hyperparameter search") # Define default scores early to handle timeouts in search phase default_scores = { "test_accuracy": np.array([0.5]), "test_f1": np.array([0.5]), "test_auc": np.array([0.5]), "fit_time": np.array([0]), "score_time": np.array([0]), "train_score": np.array([0.5]), "test_recall": np.array([0.5]), } failed = False scores = None # Initialize start_time early start_time = time.time() try: # Verify initial index alignment try: assert_index_equal(self.X_train.index, self.y_train.index) ml_grid_object.logger.debug( "Index alignment PASSED before search.run_search" ) except AssertionError: ml_grid_object.logger.error( "Index alignment FAILED before search.run_search" ) raise # Ensure y_train is a Series for consistency if not isinstance(self.y_train, pd.Series): ml_grid_object.logger.error( f"y_train is not a pandas Series, but {type(self.y_train)}. Converting to Series." ) self.y_train = pd.Series(self.y_train, index=self.X_train.index) # Reset indices for integer-based indexing X_train_reset = self.X_train.reset_index(drop=True) y_train_reset = self.y_train.reset_index(drop=True) ml_grid_object.logger.debug( f"X_train index after reset: {X_train_reset.index[:5]}" ) ml_grid_object.logger.debug( f"y_train index after reset: {y_train_reset.index[:5]}" ) # Convert y to numpy for ALL models y_train_search = self._optimize_y(y_train_reset) # Pass DataFrame to H2O and FLAML, which need column info. # Other models get a numpy array for performance. is_flaml_model = isinstance(current_algorithm, FLAMLClassifierWrapper) if not is_h2o_model and not is_flaml_model: X_train_search = X_train_reset.values else: X_train_search = X_train_reset # Skip parameter validation overhead with sklearn.config_context(skip_parameter_validation=True): # Pass reset data to search if is_h2o_model: try: import h2o h2o.no_progress() except Exception: pass # Force threading backend for search with joblib.parallel_backend("threading"): current_algorithm = search.run_search( X_train_search, y_train_search ) except TimeoutError: self.logger.warning("Timeout occurred during hyperparameter search.") failed = "Timeout" scores = default_scores except KeyboardInterrupt: if "catboost" in method_name.lower(): self.logger.warning( "KeyboardInterrupt detected during hyperparameter search. " "This is likely a signal handling artifact (e.g. from CatBoost) triggered by the timeout. " "Treating as a timeout." ) failed = "KeyboardInterrupt" scores = default_scores else: raise except Exception as e: if "dual coefficients or intercepts are not finite" in str(e): self.logger.warning( f"SVC failed to fit due to data issues: {e}. Returning default score." ) self.grid_search_cross_validate_score_result = 0.5 return self.logger.error( f"An exception occurred during hyperparameter search for {method_name}: {e}", exc_info=True, ) raise e finally: # Restore the original grid_n_jobs setting self.global_parameters.grid_n_jobs = original_grid_n_jobs if not failed and self.global_parameters.verbose >= 3: self.logger.debug("Fitting final model") if not failed and self.y_train.nunique() < 2: raise ValueError( "Only one class present in y_train. ROC AUC score is not defined " "in that case. grid_search_cross_validate>>>cross_validate" ) if not failed and self.global_parameters.verbose >= 1: self.logger.info("Getting cross validation scores") self.logger.debug( f"X_train shape: {self.X_train.shape}, y_train shape: {self.y_train.shape}" ) self.logger.debug(f"y_train value counts:\n{self.y_train.value_counts()}") # Set a time threshold in seconds time_threshold = 60 # For example, 60 seconds keras_model_types = (NeuralNetworkClassifier, KerasClassifierClass) is_h2o_model = isinstance(current_algorithm, H2O_MODEL_TYPES) is_keras_model = isinstance(current_algorithm, keras_model_types) is_flaml_model = isinstance(current_algorithm, FLAMLClassifierWrapper) is_autokeras_model = isinstance(current_algorithm, AutoKerasClassifierWrapper) # H2O and Keras models require single-threaded execution for CV final_cv_n_jobs = ( 1 if is_h2o_model or is_keras_model or is_flaml_model or is_autokeras_model else grid_n_jobs ) if final_cv_n_jobs == 1: self.logger.debug( "H2O, Keras, FLAML, or AutoKeras model detected. Forcing n_jobs=1 for final cross-validation." ) try: if failed: raise TimeoutError # H2O, FLAML and AutoKeras require pandas DataFrame to handle categorical features correctly. if is_h2o_model or is_flaml_model or is_autokeras_model: X_train_final = self.X_train # Pass DataFrame directly y_train_final = self._optimize_y(self.y_train) else: # Use to_numpy with explicit dtype=float to ensure consistent numeric types X_train_final = self.X_train.to_numpy(dtype=float, copy=True) y_train_final = self._optimize_y(self.y_train) scores = None # Check for user override to force second CV force_second_cv = self.ml_grid_object_iter.local_param_dict.get( "force_second_cv", getattr(self.global_params, "force_second_cv", False) ) if force_second_cv: self.logger.info( "force_second_cv is True. Skipping cached result extraction to run fresh cross-validation." ) # Check if we can reuse results from HyperparameterSearch if ( not force_second_cv and hasattr(current_algorithm, "cv_results_") and hasattr(current_algorithm, "best_index_") ): try: self.logger.info( "Using cached cross-validation results from HyperparameterSearch." ) results = current_algorithm.cv_results_ index = current_algorithm.best_index_ n_splits = self.cv.get_n_splits() temp_scores = {} # Extract fit and score times if "split0_fit_time" in results: temp_scores["fit_time"] = np.array( [ results[f"split{k}_fit_time"][index] for k in range(n_splits) ] ) else: # Fallback: Use mean time repeated if split times are missing (e.g. BayesSearchCV) temp_scores["fit_time"] = np.full( n_splits, results["mean_fit_time"][index] ) if "split0_score_time" in results: temp_scores["score_time"] = np.array( [ results[f"split{k}_score_time"][index] for k in range(n_splits) ] ) else: # Fallback: Use mean score time. default_times = np.zeros(index + 1) temp_scores["score_time"] = np.full( n_splits, results.get("mean_score_time", default_times)[index], ) # Extract metric scores for metric in self.metric_list: # Test scores - handle both sklearn cross_validate format and GridSearchCV format test_key = f"test_{metric}" split_test_key = f"split0_test_{metric}" if split_test_key in results: # GridSearchCV format: extract from split{k}_test_{metric} temp_scores[test_key] = np.array( [ results[f"split{k}_test_{metric}"][index] for k in range(n_splits) ] ) elif test_key in results: # cross_validate format: use result array directly temp_scores[test_key] = results[test_key] else: # Default fallback with mean score repeated fallback_value = results.get( test_key.replace("test_", "mean_test_"), 0.5 ) temp_scores[test_key] = np.full(n_splits, fallback_value) self.logger.warning( f"CV metric '{test_key}' not found in cached results. " f"Falling back to default score ({fallback_value}). " f"This may indicate misnamed metrics or missing scorers." ) # Train scores (if available) - same dual format handling. # NOTE: this block must stay INSIDE the `for metric` loop so # every metric in metric_list gets its train score extracted, # not just whichever metric happened to be last in the list. train_key = f"train_{metric}" split_train_key = f"split0_train_{metric}" if split_train_key in results: temp_scores[train_key] = np.array( [ results[f"split{k}_train_{metric}"][index] for k in range(n_splits) ] ) elif train_key in results: temp_scores[train_key] = results[train_key] else: # Train scores missing - log for debugging self.logger.warning( f"CV train metric '{train_key}' not found in cached results. " "This may indicate misnamed metrics or incomplete cv_results_" ) scores = temp_scores except Exception as e: self.logger.warning( f"Could not extract cached CV results: {e}. Falling back to standard CV." ) scores = None if scores is None: if isinstance( current_algorithm, (KerasClassifier, KerasClassifierClass) ): self.logger.debug("Fitting Keras model with internal CV handling.") y_train_values = ( self.y_train.values if hasattr(self.y_train, "values") else self.y_train ) X_train_values = ( self.X_train.values if hasattr(self.X_train, "values") else self.X_train ) current_algorithm.fit( X_train_values, y_train_values, cv=self.cv, verbose=0 ) # Since fit already did the CV, create a dummy scores dictionary. # Use defensive pattern for y_test (may be ndarray from _optimize_y) y_test_value = ( self.y_test.values if hasattr(self.y_test, "values") else self.y_test ) scores = { "test_roc_auc": [ current_algorithm.score(self.X_test, y_test_value) ] } else: # Skip parameter validation overhead with sklearn.config_context(skip_parameter_validation=True): # Ensure H2O progress is disabled before CV if is_h2o_model: try: import h2o h2o.no_progress() except Exception: pass # Always use threading backend backend = "threading" # Opt-in path: a genuinely parallel CV loop with correctly # resolved scorers and no fold-count-breaking early stopping. # Off by default; existing sklearn.cross_validate remains the # default, proven path. Enable via # global_parameters.use_optimized_cv = True once validated. use_optimized_cv = getattr( self.global_parameters, "use_optimized_cv", False ) and not (is_h2o_model or is_flaml_model or is_autokeras_model) with joblib.parallel_backend(backend): if use_optimized_cv: try: scores = self._optimized_cross_validate( current_algorithm, X_train_final, y_train_final, scoring=self.metric_list, cv=self.cv, n_jobs=final_cv_n_jobs, error_score=self.error_raise, ) except Exception as e: self.logger.warning( f"_optimized_cross_validate failed ({e}). " "Falling back to standard sklearn cross_validate." ) scores = None if scores is None: scores = cross_validate( current_algorithm, X_train_final, y_train_final, # Use optimized y (numpy for sklearn, Series for H2O) scoring=self.metric_list, cv=self.cv, n_jobs=final_cv_n_jobs, # Use adjusted n_jobs pre_dispatch="2*n_jobs", error_score=self.error_raise, # Raise error if cross-validation fails ) # Pre-compile the predict function for Keras/TF models if isinstance( current_algorithm, ( KerasClassifier, KerasClassifierClass, NeuralNetworkClassifier, ), ): try: self.logger.debug( "Pre-compiling TensorFlow predict function to avoid retracing." ) n_features = self.X_train.shape[1] # Define an input signature that allows for variable batch size. input_signature = [ tf.TensorSpec( shape=(None, n_features), dtype=tf.float32 ) ] # Access the underlying Keras model via .model_ current_algorithm.model_.predict.get_concrete_function( input_signature ) except Exception as e: self.logger.warning( f"Could not pre-compile TF function. Performance may be impacted. Error: {e}" ) except XGBoostError as e: if "cuda" in str(e).lower() or "memory" in str(e).lower(): self.logger.warning( "GPU memory error detected during cross-validation, falling back to CPU..." ) current_algorithm.set_params(tree_method="hist") try: scores = cross_validate( current_algorithm, X_train_final, y_train_final, # Use optimized y scoring=self.metric_list, cv=self.cv, n_jobs=final_cv_n_jobs, # Use adjusted n_jobs pre_dispatch="2*n_jobs", error_score=self.error_raise, # Raise error if cross-validation fails ) except Exception as e: self.logger.error( f"An unexpected error occurred during cross-validation attempt 2: {e}", exc_info=True, ) self.logger.warning("Returning default scores") failed = True scores = default_scores # Use default scores for other errors except ValueError as e: # Handle specific ValueError if AdaBoostClassifier fails due to poor performance if ( "BaseClassifier in AdaBoostClassifier ensemble is worse than random" in str(e) ): self.logger.warning(f"AdaBoostClassifier failed: {e}") self.logger.warning( "Skipping AdaBoostClassifier due to poor base classifier performance." ) # Set default scores if the AdaBoostClassifier fails failed = True scores = default_scores # Use default scores else: self.logger.error( f"An unexpected ValueError occurred during cross-validation: {e}", exc_info=True, ) failed = True scores = default_scores # Use default scores for other errors except RuntimeError as e: self.logger.error( f"A RuntimeError occurred during cross-validation (often H2O related): {e}", exc_info=True, ) self.logger.warning("Returning default scores.") failed = True scores = default_scores except TimeoutError: self.logger.warning("Timeout occurred during cross-validation.") failed = "Timeout" scores = default_scores except KeyboardInterrupt: if "catboost" in method_name.lower(): self.logger.warning( "KeyboardInterrupt detected during cross-validation. " "This is likely a signal handling artifact (e.g. from CatBoost) triggered by the timeout. " "Treating as a timeout." ) failed = "KeyboardInterrupt" scores = default_scores else: raise except Exception as e: # Catch any other general exceptions and log them self.logger.error( f"An unexpected error occurred during cross-validation: {e}", exc_info=True, ) failed = True scores = default_scores # Use default scores if an error occurs # End the timer end_time = time.time() # Calculate elapsed time elapsed_time = end_time - start_time if self.global_parameters.verbose >= 1: # Print a warning if the execution time exceeds the threshold if elapsed_time > time_threshold: self.logger.warning( f"Cross-validation took too long ({elapsed_time:.2f} seconds). " "Consider optimizing the parameters or reducing CV folds." ) else: self.logger.info( f"Cross-validation for {method_name} completed in {elapsed_time:.2f} seconds." ) if self.global_parameters.verbose >= 4: debug_print_statements_class(scores).debug_print_scores() plot_auc = False if plot_auc: # This was passing a classifier trained on the test dataset.... self.logger.debug("Plotting AUC is disabled.") try: best_pred_orig = current_algorithm.predict(self.X_test) # exp except Exception: best_pred_orig = np.zeros(len(self.X_test)) # Call the update_score_log method on the provided instance if self.project_score_save_class_instance: self.project_score_save_class_instance.update_score_log( ml_grid_object=ml_grid_object, scores=scores, best_pred_orig=best_pred_orig, current_algorithm=current_algorithm, method_name=method_name, pg=pg, start=start, n_iter_v=n_iter_v, failed=failed, ) else: self.logger.warning( "No project_score_save_class_instance provided. Skipping score logging." ) # calculate metric for optimisation try: y_test_np = ( self.y_test.values if hasattr(self.y_test, "values") else self.y_test ) auc = metrics.roc_auc_score(y_test_np, best_pred_orig) except Exception: auc = 0.5
[docs] self.grid_search_cross_validate_score_result = auc
self._shutdown_h2o_if_needed(current_algorithm) def _get_adaptive_cv_strategy(self): """ Dynamically selects the optimal cross-validation strategy based on data characteristics. This depends only on dataset properties (size, class balance) and NOT on the algorithm being evaluated, so that every algorithm run against a given dataset uses an identical fold structure and their resulting scores remain directly comparable. Optimizations applied: 1. For imbalanced binary datasets, uses StratifiedKFold to maintain class distribution 2. For small datasets (n_samples < 100), reduces n_splits to prevent sparse folds 3. For larger datasets, maintains standard 5-fold CV 4. Falls back to plain KFold if stratification isn't applicable/possible Returns: CV strategy object (StratifiedKFold or KFold) """ n_samples = self._n_samples_train # Detect class imbalance y_values = ( self.y_train if isinstance(self.y_train, pd.Series) else pd.Series(self.y_train) ) y_unique = y_values.unique() n_classes = len(y_unique) is_binary = n_classes == 2 class_ratio = 1.0 value_counts = None if is_binary: value_counts = y_values.value_counts() if len(value_counts) >= 2: class_ratio = min(value_counts) / max(value_counts) # Determine n_splits based on dataset size. # Smaller datasets need fewer splits to avoid very small training folds. if n_samples < 30: n_splits = 2 elif n_samples < 100: n_splits = 3 elif n_samples < 500: n_splits = 5 else: n_splits = 5 # Check global cv_strategy parameter to override auto-selection if getattr(self.global_params, "cv_strategy", None) == "standard": self.logger.debug("Using standard K-fold (cv_strategy='standard')") return KFold(n_splits=n_splits, shuffle=True, random_state=1) # For binary classification, prefer StratifiedKFold to preserve class # distribution across folds, provided there are enough samples per class. if is_binary and is_sklearn_version_supported(): try: from sklearn.model_selection import StratifiedKFold min_class_samples = min(value_counts) if value_counts is not None else 0 if min_class_samples >= n_splits: self.logger.debug( f"Using StratifiedKFold for binary classification " f"(class ratio={class_ratio:.2f}, n_samples={n_samples})" ) return StratifiedKFold( n_splits=n_splits, shuffle=True, random_state=1 ) else: self.logger.debug( "Not enough samples per class for StratifiedKFold " f"(min_class_samples={min_class_samples} < n_splits={n_splits}). " "Falling back to KFold." ) except Exception as e: self.logger.debug(f"StratifiedKFold unavailable, falling back: {e}") # Fallback to standard KFold return KFold(n_splits=n_splits, shuffle=True, random_state=1) def _calculate_convergence_threshold(self, n_samples, n_splits): """ Calculates a variance threshold used only for informational convergence logging (NOT for early-stopping / truncating folds, which would break cross-algorithm score comparability). Args: n_samples: Number of training samples n_splits: Number of CV splits Returns: float: Variance threshold for convergence logging """ base_threshold = 0.01 if n_samples < 50: threshold_multiplier = 2.0 elif n_samples < 200: threshold_multiplier = 1.5 elif n_samples < 1000: threshold_multiplier = 1.0 else: threshold_multiplier = 0.8 if n_splits >= 5: split_factor = 0.9 elif n_splits >= 3: split_factor = 1.0 else: split_factor = 1.2 return base_threshold * threshold_multiplier * split_factor def _optimized_cross_validate( self, estimator, X, y, scoring, cv, n_jobs=1, error_score=np.nan, ): """ Parallel cross-validation with correctly resolved scorers. This is an OPT-IN alternative to sklearn's cross_validate, enabled via global_parameters.use_optimized_cv = True. It exists mainly as a hook for future optimizations (e.g. estimator-specific fast paths); today it intentionally mirrors sklearn's own semantics rather than introducing early-stopping/fold-truncation, because truncating folds differently per-algorithm breaks cross-algorithm score comparability, which is the core requirement of this grid search framework. Args: estimator: The estimator to cross-validate (unfitted or fitted; will be cloned per fold). X: Training features. y: Training labels. scoring: List of sklearn scorer name strings (e.g. ["accuracy", "f1"]). cv: A cross-validation splitter. n_jobs: Number of parallel jobs for fold dispatch. error_score: Value to assign if a fold's fit fails. Returns: Dictionary matching sklearn.model_selection.cross_validate's output shape: {"fit_time": ..., "score_time": ..., "test_<metric>": ..., ...} """ # Resolve scorer name strings to actual scorer callables up front. # `scoring` is a list of strings (e.g. self.global_params.metric_list), # NOT a dict, so we must not call .items()/.keys() on it directly. if isinstance(scoring, dict): scorer_dict = scoring else: scorer_dict = {name: metrics.get_scorer(name) for name in scoring} X = sklearn.utils.validation.check_array( X, accept_sparse=True, dtype=None, force_all_finite=False ) y = sklearn.utils.validation.check_array(y, ensure_2d=False, dtype=None) n_samples = X.shape[0] if hasattr(X, "shape") else len(X) n_splits = ( cv.get_n_splits(X, y) if hasattr(cv, "get_n_splits") else getattr(cv, "n_splits", 3) ) # Used only for informational logging, not to truncate folds. convergence_threshold = self._calculate_convergence_threshold( n_samples, n_splits ) def _fit_and_score_fold(train_idx, val_idx): X_train_fold, X_val_fold = X[train_idx], X[val_idx] y_train_fold, y_val_fold = y[train_idx], y[val_idx] fold_estimator = sklearn.base.clone(estimator) fold_fit_start = time.time() try: fold_estimator.fit(X_train_fold, y_train_fold) except Exception as e: self.logger.warning(f"CV fold failed during fit: {e}") fold_result = {"fit_time": 0.0, "score_time": 0.0} for metric_name in scorer_dict: fold_result[f"test_{metric_name}"] = error_score return fold_result fit_time = time.time() - fold_fit_start fold_score_start = time.time() fold_result = {"fit_time": fit_time} for metric_name, scorer in scorer_dict.items(): try: fold_result[f"test_{metric_name}"] = scorer( fold_estimator, X_val_fold, y_val_fold ) except Exception as e: self.logger.warning( f"Score calculation failed for '{metric_name}': {e}" ) fold_result[f"test_{metric_name}"] = error_score fold_result["score_time"] = time.time() - fold_score_start return fold_result splits = list(cv.split(X, y)) fold_results = joblib.Parallel(n_jobs=n_jobs, backend="threading")( joblib.delayed(_fit_and_score_fold)(train_idx, val_idx) for train_idx, val_idx in splits ) # Informational convergence logging only (does not affect fold count). for metric_name in scorer_dict: metric_values = [ fr[f"test_{metric_name}"] for fr in fold_results if isinstance(fr.get(f"test_{metric_name}"), (int, float)) ] if len(metric_values) >= 2: score_variance = np.var(metric_values) if score_variance < convergence_threshold: self.logger.debug( f"CV scores for '{metric_name}' show low variance " f"({score_variance:.6f} < {convergence_threshold:.6f}); " "folds appear converged (all folds still run for comparability)." ) # Assemble sklearn-cross_validate-shaped output. converted_scores = {} converted_scores["fit_time"] = np.array([fr["fit_time"] for fr in fold_results]) converted_scores["score_time"] = np.array( [fr["score_time"] for fr in fold_results] ) for metric_name in scorer_dict: key = f"test_{metric_name}" converted_scores[key] = np.array( [fr.get(key, error_score) for fr in fold_results] ) return converted_scores def _optimize_y(self, y): """Helper to optimize y for sklearn/H2O to reduce type_of_target overhead.""" # Convert to numpy if it's a Series or Categorical if hasattr(y, "dtype") and isinstance(y.dtype, pd.CategoricalDtype): y_opt = y.cat.codes.values elif hasattr(y, "values"): y_opt = y.values else: y_opt = y # Force integer encoding if not pd.api.types.is_integer_dtype(y_opt): try: y_opt = y_opt.astype(int) except (ValueError, TypeError): y_opt, _ = pd.factorize(y_opt, sort=True) y_opt = y_opt.astype(int) # Ensure contiguous array for speed in np.unique and other ops return np.ascontiguousarray(y_opt) def _adjust_knn_parameters(self, parameter_space: Union[Dict, List[Dict]]): """ Dynamically adjusts the 'n_neighbors' parameter for KNN-based models to prevent errors on small datasets during cross-validation. """ self.cv.get_n_splits() # Correctly calculate the training fold size dummy_indices = np.arange(len(self.X_train)) train_indices, _ = next(self.cv.split(dummy_indices)) n_samples_train_fold = len(train_indices) n_samples_test_fold = len(self.X_train) - n_samples_train_fold max_n_neighbors = max(1, n_samples_train_fold) self.logger.debug( f"KNN constraints - train_fold_size={n_samples_train_fold}, " f"test_fold_size={n_samples_test_fold}, max_n_neighbors={max_n_neighbors}" ) def adjust_param(param_value): """Adjusts parameter values for KNN models. Args: param_value: The parameter value to adjust (skopt space, list, or scalar). Returns: Adjusted parameter value suitable for the training fold size. """ if is_skopt_space(param_value): # For skopt.space objects, adjust the upper bound new_high = min(param_value.high, max_n_neighbors) new_low = min(param_value.low, new_high) param_value.high = new_high param_value.low = new_low self.logger.debug( f"Adjusted skopt space: low={new_low}, high={new_high}" ) elif isinstance(param_value, (list, np.ndarray)): # For lists, filter the values new_param_value = [n for n in param_value if n <= max_n_neighbors] if not new_param_value: self.logger.warning( f"All n_neighbors values filtered out. Using [{max_n_neighbors}]" ) return [max_n_neighbors] self.logger.debug(f"Filtered n_neighbors list: {new_param_value}") return new_param_value return param_value if isinstance(parameter_space, list): for params in parameter_space: if "n_neighbors" in params: params["n_neighbors"] = adjust_param(params["n_neighbors"]) elif isinstance(parameter_space, dict) and "n_neighbors" in parameter_space: parameter_space["n_neighbors"] = adjust_param( parameter_space["n_neighbors"] ) def _adjust_catboost_parameters(self, parameter_space: Union[Dict, List[Dict]]): """ Dynamically adjusts the 'subsample' parameter for CatBoost to prevent errors on small datasets during cross-validation. """ n_splits = self.cv.get_n_splits() # Correctly calculate the size of the smallest training fold. n_samples_in_fold = len(self.X_train) - (len(self.X_train) // n_splits) # Ensure n_samples_in_fold is at least 1 to avoid division by zero n_samples_in_fold = max(1, n_samples_in_fold) # If the training fold is extremely small, force subsample to 1.0 # to prevent CatBoost from failing on constant features. if n_samples_in_fold <= 2: min_subsample = 1.0 else: # The minimum subsample value must be > 1/n_samples to ensure at least one sample is chosen min_subsample = 1.0 / n_samples_in_fold def adjust_param(param_value): """Adjusts parameter values for CatBoost models. Args: param_value: The parameter value to adjust (skopt space, list, or scalar). Returns: Adjusted parameter value suitable for the training fold size. """ if is_skopt_space(param_value): # For skopt.space objects (Real), adjust the lower bound new_low = max(param_value.low, min_subsample) # Ensure the new low is not higher than the high if new_low > param_value.high: new_low = param_value.high param_value.low = new_low # If the fold is tiny, force the entire space to be 1.0 if n_samples_in_fold <= 2: param_value.low = param_value.high = 1.0 elif isinstance(param_value, (list, np.ndarray)): # For lists, filter the values new_param_value = [s for s in param_value if s >= min_subsample] if not new_param_value: # If all values are filtered out, use the smallest valid value return [ ( min(p for p in param_value if p > 0) if any(p > 0 for p in param_value) else 1.0 ) ] return new_param_value # If the fold is tiny, force subsample to 1.0 if n_samples_in_fold <= 2: return [1.0] if isinstance(param_value, list) else 1.0 return param_value if isinstance(parameter_space, list): for params in parameter_space: if "subsample" in params: params["subsample"] = adjust_param(params["subsample"]) elif isinstance(parameter_space, dict) and "subsample" in parameter_space: parameter_space["subsample"] = adjust_param(parameter_space["subsample"]) # Also adjust 'rsm' (colsample_bylevel) which can cause the same issue if isinstance(parameter_space, list): for params in parameter_space: if "rsm" in params: params["rsm"] = adjust_param(params["rsm"]) elif isinstance(parameter_space, dict) and "rsm" in parameter_space: parameter_space["rsm"] = adjust_param(parameter_space["rsm"]) def _adjust_xgboost_parameters(self, parameter_space: Union[Dict, List[Dict]]): """ Dynamically adjusts 'max_bin' for XGBoost to prevent errors on small datasets. """ # Ensure max_bin is at least 2 min_max_bin = 2 def adjust_param(param_value): """Adjusts parameter values for XGBoost models. Args: param_value: The parameter value to adjust (skopt space, list, or scalar). Returns: Adjusted parameter value suitable for minimum requirements. """ if is_skopt_space(param_value): # For skopt Integer space if hasattr(param_value, "low"): new_low = max(param_value.low, min_max_bin) if new_low > param_value.high: # If low > high, we must adjust high as well param_value.high = max(param_value.high, min_max_bin) new_low = min(new_low, param_value.high) param_value.low = new_low elif isinstance(param_value, (list, np.ndarray)): # Filter out invalid values new_param_value = [v for v in param_value if v >= min_max_bin] if not new_param_value: return [min_max_bin] return new_param_value # If it's a single scalar (int), ensure it's >= 2 elif isinstance(param_value, (int, float)): if param_value < min_max_bin: return min_max_bin return param_value if isinstance(parameter_space, list): for params in parameter_space: if "max_bin" in params: params["max_bin"] = adjust_param(params["max_bin"]) elif isinstance(parameter_space, dict) and "max_bin" in parameter_space: parameter_space["max_bin"] = adjust_param(parameter_space["max_bin"]) def _shutdown_h2o_if_needed(self, algorithm: Any): """Safely shuts down the H2O cluster if the algorithm is an H2O model.""" # Use the module-level tuple if isinstance(algorithm, H2O_MODEL_TYPES): import h2o cluster = h2o.cluster() if cluster and cluster.is_running(): self.logger.info( "H2O model finished. Leaving cluster running for next H2O model." )
[docs] def dummy_auc() -> float: """Returns a constant AUC score of 0.5. This function is intended as a placeholder or for use in scenarios where a valid AUC score cannot be calculated but a value is required. Returns: float: A constant value of 0.5. """ return 0.5
# Create a scorer using make_scorer # dummy_auc_scorer = make_scorer(dummy_auc)
[docs] def scale_data(X_train: pd.DataFrame) -> pd.DataFrame: """Scales the data to a [0, 1] range if it's not already scaled. Args: X_train (pd.DataFrame): Training features. Returns: pd.DataFrame: Scaled training features. """ # Initialize MinMaxScaler scaler = MinMaxScaler(feature_range=(0, 1)) # Check if data is already scaled min_val = X_train.min().min() max_val = X_train.max().max() # If data is not scaled, then scale it if min_val < 0 or max_val > 1: # Fit and transform the data X_train_scaled = pd.DataFrame( scaler.fit_transform(X_train), columns=X_train.columns ) return X_train_scaled else: # If data is already scaled, return it as is return X_train