Pipeline optimizer#

Main PipelineOptimizer class implementation.

class spectoprep.pipeline.optimizer.PipelineOptimizer(X_train, y_train, preprocessing_steps=None, X_test=None, y_test=None, cv_method='group_shuffle_split', n_splits=3, test_size=0.3, n_groups_out=2, random_state=42, groups=None, max_pipeline_length=5, n_jobs=-1, allowed_preprocess_combinations=None, log_level='INFO')[source]#

Bases: object

A class for optimizing machine learning pipelines using Bayesian optimization. It precomputes possible pipeline configurations and then searches over both the pipeline configuration (encoded as an index) and the hyperparameters.

Parameters:

  • X_train (NDArray)

  • y_train (NDArray)

  • preprocessing_steps (list[str] | str | None)

  • X_test (NDArray | None)

  • y_test (NDArray | None)

  • cv_method (str)

  • n_splits (int)

  • test_size (float)

  • n_groups_out (int)

  • random_state (int)

  • groups (NDArray | None)

  • max_pipeline_length (int)

  • n_jobs (int)

  • allowed_preprocess_combinations (int | list[int] | tuple[int, ...] | None)

  • log_level (str)

__init__(X_train, y_train, preprocessing_steps=None, X_test=None, y_test=None, cv_method='group_shuffle_split', n_splits=3, test_size=0.3, n_groups_out=2, random_state=42, groups=None, max_pipeline_length=5, n_jobs=-1, allowed_preprocess_combinations=None, log_level='INFO')[source]#

Initialize the PipelineOptimizer.

Parameters:

  • X_train (NDArray) – Training features.

  • y_train (NDArray) – Training targets.

  • preprocessing_steps (list[str] | str | None) – List of preprocessing steps to use.

  • X_test (NDArray | None) – Test features (optional).

  • y_test (NDArray | None) – Test targets (optional).

  • cv_method (str) – One of “group_shuffle_split”, “group_kfold”, or “leave_p_group_out”.

  • n_splits (int) – Number of CV splits.

  • test_size (float) – Test set fraction (if using GroupShuffleSplit).

  • n_groups_out (int) – Number of groups left out (if using LeavePGroupsOut).

  • random_state (int) – Random seed.

  • groups (NDArray | None) – Optional group labels; if None, one group per sample is used.

  • max_pipeline_length (int) – Maximum number of steps in pipeline.

  • n_jobs (int) – Number of parallel jobs for compatible estimators.

  • allowed_preprocess_combinations (int | list[int] | tuple[int, ...] | None) – Allowed lengths for preprocessing combinations.

  • log_level (str) – Logging level (INFO, DEBUG, WARNING, ERROR).

bayes_objective(**params)[source]#

Objective function for Bayesian optimization.

The score is the negative mean RMSE under group-aware cross-validation on the training data only. Any supplied test set is deliberately never touched here: using it to select preprocessing or hyperparameters would leak the held-out data and inflate reported performance. The test set is used only for final reporting in get_best_pipeline_predictions().

Parameters:

**params – Parameters proposed by the Bayesian optimizer.

Returns:

Negative cross-validated RMSE, or a large penalty (-1e6) if no valid predictions could be produced.

Return type:

float

bayesian_optimize(init_points=10, n_iter=50, acquisition_function='ei')[source]#

Run Bayesian optimization to find the best pipeline configuration and hyperparameters.

Parameters:

  • init_points (int) – Number of random initial points

  • n_iter (int) – Number of Bayesian optimization iterations

  • acquisition_function (str) – Acquisition function for Bayesian optimization

Returns:

  • Dict of best parameters

  • Fitted Pipeline with best configuration

Return type:

Tuple containing

get_best_pipeline_predictions(best_pipeline)[source]#

Get predictions using the best pipeline.

Parameters:

best_pipeline (Pipeline) – Fitted pipeline object

Returns:

  • Predictions array

  • RMSE score

  • R² score

Return type:

Tuple containing

get_all_tested_pipelines()[source]#

Get details of all tested pipeline configurations.

Returns:

List of dictionaries with pipeline details

Return type:

list[dict]

print_evaluated_pipelines()[source]#

Print details for all evaluated pipelines from the Bayesian optimizer.

This method assumes that bayesian_optimize() has been run and that self.optimizer exists.

Return type:

None

export_best_pipeline(file_path)[source]#

Export the best pipeline configuration and hyperparameters to a file.

Parameters:

file_path (str) – Path to save the export file

Raises:

AttributeError – If optimizer hasn’t been run yet

Return type:

None

summarize_optimization()[source]#

Generate a summary of the optimization results.

Returns:

Dictionary containing optimization summary metrics

Return type:

dict

Configuration#

Configuration constants for the ML Pipeline Optimizer.

Builder utilities#

Functions for building preprocessing steps from parameters.

spectoprep.pipeline.builder.build_preprocessor_from_bayes(name, params, X_train_shape, random_state, n_jobs)[source]#

Build a transformer from bayesian optimization parameters.

Parameters:

  • name (str) – Name of the transformer

  • params (dict) – Dictionary of parameters for the transformer

  • X_train_shape (tuple) – Shape of the training data

  • random_state (int) – Random state for reproducibility

  • n_jobs (int) – Number of CPU cores to use

Returns:

The constructed transformer object

Return type:

object

Pipeline helpers#

Utility functions for pipeline optimization.

spectoprep.pipeline.utils.choose_nearest(value, allowed)[source]#

Choose the nearest allowed value to the given value.

Parameters:

  • value (float) – The input value

  • allowed (list[float]) – List of allowed values

Returns:

The nearest allowed value

Return type:

float

spectoprep.pipeline.utils.is_valid_pipeline(pipeline, incompatible_sets)[source]#

Check if a pipeline configuration is valid based on incompatibility rules.

Parameters:

  • pipeline (tuple[str, ...]) – Tuple of preprocessing step names

  • incompatible_sets (list[list[str]]) – List of sets, each containing mutually incompatible steps

Returns:

True if the pipeline is valid, False otherwise

Return type:

bool

spectoprep.pipeline.utils.generate_pipeline_configurations(preprocessing_steps, incompatible_sets, max_length=5, allowed_lengths=None)[source]#

Generate all valid pipeline configurations based on available steps and incompatibility rules.

Parameters:

  • preprocessing_steps (list[str]) – List of available preprocessing step names

  • incompatible_sets (list[list[str]]) – List of sets, each containing mutually incompatible steps

  • max_length (int) – Maximum number of steps in a pipeline

  • allowed_lengths (int | list[int] | tuple[int, ...] | None) – Specific pipeline lengths to allow (e.g., [1, 2])

Returns:

List of tuples, each representing a valid pipeline configuration

Return type:

list[tuple[str, …]]

spectoprep.pipeline.utils.calculate_rmse(y_true, y_pred)[source]#

Compute Root Mean Square Error (RMSE).

Parameters:

  • y_true (ndarray) – True target values

  • y_pred (ndarray) – Predicted target values

Returns:

RMSE value

Return type:

float

spectoprep.pipeline.utils.calculate_r2(y_true, y_pred)[source]#

Compute the R² score.

Parameters:

  • y_true (ndarray) – True target values

  • y_pred (ndarray) – Predicted target values

Returns:

R² value

Return type:

float