Selectors

APPS

Bases: AbstractSelector

Automatic Parallel Portfolio Selector based on the approach by Kashgarani and Kotthoff.

This selector predicts performance distributions for each algorithm and selects a parallel portfolio based on the overlap of each algorithm's predicted runtime distribution with the best algorithm's distribution.

The size of the portfolio is controlled by the p_intersection threshold, which determines the minimum overlap required for an algorithm to be included.

References

Kashgarani, H. J., and Kotthoff, L. (2020). "Automatic Parallel Portfolio Construction." https://par.nsf.gov/biblio/10470673

Source code in asf/selectors/parallel_portfolio_selector.py

class APPS(AbstractSelector):
    """
    Automatic Parallel Portfolio Selector based on the approach by Kashgarani and Kotthoff.

    This selector predicts performance distributions for each algorithm and selects
    a parallel portfolio based on the overlap of each algorithm's predicted runtime
    distribution with the best algorithm's distribution.

    The size of the portfolio is controlled by the p_intersection threshold, which
    determines the minimum overlap required for an algorithm to be included.

    References
    ----------
    Kashgarani, H. J., and Kotthoff, L. (2020).
    "Automatic Parallel Portfolio Construction."
    https://par.nsf.gov/biblio/10470673
    """

    PREFIX = "parallel_portfolio"
    RETURN_TYPE = "portfolio"

    def __init__(
        self,
        model_class: type[AbstractPredictor] = RandomForestRegressorWrapper,
        p_intersection: float = 0.82,
        n_estimators_for_std: int = 10,
        random_state: int = 42,
        use_jackknife: bool = True,
        n_jackknife_folds: int | None = None,
        **kwargs: Any,
    ):
        """
        Initialize the Parallel Portfolio Selector.

        Args:
            model_class: The predictor class to use for performance modeling.
            p_intersection: Threshold for distribution overlap.
                          Higher values -> smaller portfolios.
            n_estimators_for_std: Number of Jackknife iterations (n_instances) or bootstrap samples.
            random_state: Random seed for reproducibility.
            use_jackknife: If True, use Jackknife method; if False, use bootstrap.
            n_jackknife_folds: Number of folds for Jackknife. If None, uses leave-one-out.
                              E.g., 10 means 10-fold cross-validation (10 models per algorithm).
            **kwargs: Additional arguments passed to parent class.
        """
        super().__init__(**kwargs)
        self.model_class = model_class
        self.p_intersection = float(p_intersection)
        self.n_estimators_for_std = int(n_estimators_for_std)
        self.random_state = int(random_state)
        self.use_jackknife = bool(use_jackknife)
        self.n_jackknife_folds = n_jackknife_folds

        self.predictors: (
            List[List[tuple[AbstractPredictor, Any]]] | List[List[AbstractPredictor]]
        ) = []
        self.algorithms: List[str] = []
        self.features_train: pd.DataFrame | None = None
        self.performance_train: pd.DataFrame | None = None

    def _fit(self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs) -> None:
        """
        Train ensemble of performance models for each algorithm.
        Uses either Jackknife or bootstrap method based on use_jackknife flag.

        Args:
            features: DataFrame of instance features.
            performance: DataFrame of algorithm performance (runtimes).
        """
        self.algorithms = list(performance.columns)
        self.features_train = features.copy()
        self.performance_train = performance.copy()

        rng = np.random.RandomState(self.random_state)
        n_instances = len(features)

        if self.use_jackknife:
            # Jackknife method: K-fold or leave-one-out
            if self.n_jackknife_folds is None:
                n_splits = n_instances
            else:
                n_splits = self.n_jackknife_folds

            kfold = KFold(
                n_splits=n_splits, shuffle=True, random_state=self.random_state
            )

            for algo_idx, _ in enumerate(self.algorithms):
                algo_models = []
                algo_performance = performance.iloc[:, algo_idx]

                for fold_idx, (train_idx, test_idx) in enumerate(kfold.split(features)):
                    X_train = features.iloc[train_idx]
                    y_train = algo_performance.iloc[train_idx]

                    model = self.model_class()
                    model.fit(X_train, y_train)
                    # Store (model, test_indices) for this fold
                    algo_models.append((model, test_idx))

                self.predictors.append(algo_models)
        else:
            # Bootstrap method: random sampling with replacement
            for algo_idx, _ in enumerate(self.algorithms):
                algo_models = []
                algo_performance = performance.iloc[:, algo_idx]

                for _ in range(self.n_estimators_for_std):
                    sample_indices = rng.choice(
                        n_instances, size=n_instances, replace=True
                    )
                    X_boot = features.iloc[sample_indices]
                    y_boot = algo_performance.iloc[sample_indices]

                    model = self.model_class()
                    model.fit(X_boot, y_boot)
                    algo_models.append(model)

                self.predictors.append(algo_models)

    def _predict_with_uncertainty(
        self, features: pd.DataFrame
    ) -> tuple[np.ndarray, np.ndarray]:
        """
        Predict mean and standard deviation for each algorithm on each instance.
        For Jackknife: uses leave-one-out predictions to estimate std dev.
        For bootstrap: uses bootstrap ensemble predictions to estimate std dev.

        Args:
            features: DataFrame of instance features.

        Returns:
            means: Array of shape (n_instances, n_algorithms) with mean predictions
            stds: Array of shape (n_instances, n_algorithms) with std dev predictions
        """
        n_instances = len(features)
        n_algorithms = len(self.algorithms)

        if self.use_jackknife:
            # Jackknife: compute mean predictions and std dev across K-fold models
            means = np.zeros((n_instances, n_algorithms))
            stds = np.zeros((n_instances, n_algorithms))

            for algo_idx in range(n_algorithms):
                # For each test instance, collect predictions from all models
                all_predictions = []

                # Type narrowing: ensure we have the right structure
                algo_models = self.predictors[algo_idx]
                if not isinstance(algo_models, list):
                    raise RuntimeError("Expected list of models")

                for model_tuple in algo_models:
                    if isinstance(model_tuple, tuple):
                        model, test_idx = model_tuple
                    else:
                        model = model_tuple
                    predictor = cast(AbstractPredictor, model)
                    pred = predictor.predict(features)
                    all_predictions.append(pred)

                all_predictions = np.array(all_predictions)  # (n_folds, n_test)
                means[:, algo_idx] = np.mean(all_predictions, axis=0)
                stds[:, algo_idx] = np.std(all_predictions, axis=0)

            stds = np.maximum(stds, 1e-6)
            return means, stds
        else:
            # Bootstrap: compute mean and std dev across bootstrap samples
            all_predictions = np.zeros(
                (n_instances, n_algorithms, self.n_estimators_for_std)
            )

            for algo_idx in range(n_algorithms):
                for model_idx, model in enumerate(self.predictors[algo_idx]):
                    all_predictions[:, algo_idx, model_idx] = model.predict(features)

            means = np.mean(all_predictions, axis=2)
            stds = np.std(all_predictions, axis=2)
            stds = np.maximum(stds, 1e-6)

            return means, stds

    def _solve_intersection_vectorized(
        self,
        mu_best: float,
        sigma_best: float,
        mu_candidates: np.ndarray,
        sigma_candidates: np.ndarray,
    ) -> np.ndarray:
        """
        Vectorized computation of PDF intersection points for multiple candidates.

        Args:
            mu_best: Scalar or array of best algorithm means
            sigma_best: Scalar or array of best algorithm stds
            mu_candidates: Array of candidate means (n_candidates,)
            sigma_candidates: Array of candidate stds (n_candidates,)

        Returns:
            Array of intersection points c for each candidate
        """
        # Handle equal variance case
        equal_var = np.abs(sigma_best - sigma_candidates) < 1e-9
        c = np.where(equal_var, (mu_best + mu_candidates) / 2.0, 0.0)

        # Compute for unequal variance cases
        unequal_mask = ~equal_var
        if np.any(unequal_mask):
            sigma_best_arr = np.asarray(sigma_best)
            mu_best_arr = np.asarray(mu_best)

            if sigma_best_arr.ndim == 0:
                sigma_best_ue = cast(float, sigma_best_arr.item())
            else:
                sigma_best_ue = sigma_best_arr[unequal_mask]

            if mu_best_arr.ndim == 0:
                mu_best_ue = cast(float, mu_best_arr.item())
            else:
                mu_best_ue = mu_best_arr[unequal_mask]

            var1 = np.asarray(sigma_best_ue) ** 2
            var2 = np.asarray(sigma_candidates[unequal_mask]) ** 2

            a = 0.5 / var1 - 0.5 / var2
            mu_best_arr = np.asarray(mu_best_ue)
            b = mu_candidates[unequal_mask] / var2 - mu_best_arr / var1
            c_coeff = (
                (mu_best_arr**2) / (2 * var1)
                - (mu_candidates[unequal_mask] ** 2) / (2 * var2)
                - np.log(
                    np.asarray(sigma_candidates[unequal_mask])
                    / np.asarray(sigma_best_ue)
                )
            )

            delta = b**2 - 4 * a * c_coeff
            delta = np.maximum(delta, 0)  # Clamp negative deltas

            sqrt_delta = np.sqrt(delta)
            x1 = (-b - sqrt_delta) / (2 * a)
            x2 = (-b + sqrt_delta) / (2 * a)

            midpoint = (mu_best_ue + mu_candidates[unequal_mask]) / 2.0

            # Choose x closest to midpoint
            c_unequal = np.where(np.abs(x1 - midpoint) < np.abs(x2 - midpoint), x1, x2)

            c[unequal_mask] = c_unequal

        return c

    def _compute_overlap_vectorized(
        self,
        mu_best: float,
        sigma_best: float,
        mu_candidates: np.ndarray,
        sigma_candidates: np.ndarray,
    ) -> np.ndarray:
        """
        Vectorized computation of distribution overlaps.

        Args:
            mu_best: Best algorithm mean
            sigma_best: Best algorithm std
            mu_candidates: Array of candidate means
            sigma_candidates: Array of candidate stds

        Returns:
            Array of overlap values for each candidate
        """
        c = self._solve_intersection_vectorized(
            float(mu_best), float(sigma_best), mu_candidates, sigma_candidates
        )

        # Vectorized CDF computation
        p_cand_left = stats.norm.cdf(c, loc=mu_candidates, scale=sigma_candidates)
        p_best_right = 1.0 - stats.norm.cdf(c, loc=mu_best, scale=sigma_best)

        return p_cand_left + p_best_right

    def _solve_intersection(
        self, mu1: float, sigma1: float, mu2: float, sigma2: float
    ) -> float:
        """
        Solve for the PDF intersection point between two Gaussians.
        Assumes mu1 <= mu2; returns a point between the means when possible.
        """

        if abs(sigma1 - sigma2) < 1e-9:
            return (mu1 + mu2) / 2.0

        var1 = sigma1**2
        var2 = sigma2**2

        a = 0.5 / var1 - 0.5 / var2
        b = mu2 / var2 - mu1 / var1
        c = (mu1**2) / (2 * var1) - (mu2**2) / (2 * var2) - np.log(sigma2 / sigma1)

        delta = b**2 - 4 * a * c
        if delta < 0:
            return (mu1 + mu2) / 2.0

        sqrt_delta = np.sqrt(delta)
        x1 = (-b - sqrt_delta) / (2 * a)
        x2 = (-b + sqrt_delta) / (2 * a)

        midpoint = (mu1 + mu2) / 2.0
        in_range_1 = mu1 <= x1 <= mu2
        in_range_2 = mu1 <= x2 <= mu2

        if in_range_1 and not in_range_2:
            return x1
        if in_range_2 and not in_range_1:
            return x2

        return x1 if abs(x1 - midpoint) < abs(x2 - midpoint) else x2

    def _compute_overlap(
        self,
        mu_best: float,
        sigma_best: float,
        mu_candidate: float,
        sigma_candidate: float,
    ) -> float:
        """
        Compute distribution overlap as in Kashgarani & Kotthoff.
        Overlap = P(candidate <= c) + P(best >= c), where c is the PDF intersection.
        """

        c = self._solve_intersection(mu_best, sigma_best, mu_candidate, sigma_candidate)

        p_cand_left = stats.norm.cdf(c, loc=mu_candidate, scale=sigma_candidate)
        p_best_right = 1.0 - stats.norm.cdf(c, loc=mu_best, scale=sigma_best)

        return p_cand_left + p_best_right

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict parallel portfolio for each instance.

        Returns:
            Dictionary mapping instance names to lists of (algorithm, budget) tuples.
        """
        if features is None:
            raise ValueError("features cannot be None for APPS prediction")

        budget = getattr(self, "budget", None)
        if budget is None:
            budget = float("inf")

        means, stds = self._predict_with_uncertainty(features)
        predictions: dict[str, list[tuple[str, float]]] = {}

        # Vectorized ranking across all instances
        ranks = np.argsort(means, axis=1)  # (n_instances, n_algorithms)
        best_algo_indices = ranks[:, 0]

        mu_best_all = means[np.arange(len(means)), best_algo_indices]
        sigma_best_all = stds[np.arange(len(stds)), best_algo_indices]

        for inst_idx, inst_name in enumerate(features.index):
            mu_best = mu_best_all[inst_idx]
            sigma_best = sigma_best_all[inst_idx]
            best_algo_idx = best_algo_indices[inst_idx]

            portfolio = [self.algorithms[best_algo_idx]]

            # Vectorized overlap computation for all candidates
            mu_candidates = means[inst_idx, :]
            sigma_candidates = stds[inst_idx, :]
            overlaps = self._compute_overlap_vectorized(
                mu_best, sigma_best, mu_candidates, sigma_candidates
            )

            # Find algorithms that meet threshold (excluding the best one)
            for rank_idx in range(1, len(self.algorithms)):
                algo_idx = ranks[inst_idx, rank_idx]
                if overlaps[algo_idx] >= self.p_intersection:
                    portfolio.append(self.algorithms[algo_idx])

            predictions[inst_name] = [(algo, float(budget)) for algo in portfolio]

        return predictions

__init__(model_class=RandomForestRegressorWrapper, p_intersection=0.82, n_estimators_for_std=10, random_state=42, use_jackknife=True, n_jackknife_folds=None, **kwargs)

Initialize the Parallel Portfolio Selector.

Parameters:

Name	Type	Description	Default
model_class	type[AbstractPredictor]	The predictor class to use for performance modeling.	RandomForestRegressorWrapper
p_intersection	float	Threshold for distribution overlap. Higher values -> smaller portfolios.	0.82
n_estimators_for_std	int	Number of Jackknife iterations (n_instances) or bootstrap samples.	10
random_state	int	Random seed for reproducibility.	42
use_jackknife	bool	If True, use Jackknife method; if False, use bootstrap.	True
n_jackknife_folds	int | None	Number of folds for Jackknife. If None, uses leave-one-out. E.g., 10 means 10-fold cross-validation (10 models per algorithm).	None
**kwargs	Any	Additional arguments passed to parent class.	{}

Source code in asf/selectors/parallel_portfolio_selector.py

def __init__(
    self,
    model_class: type[AbstractPredictor] = RandomForestRegressorWrapper,
    p_intersection: float = 0.82,
    n_estimators_for_std: int = 10,
    random_state: int = 42,
    use_jackknife: bool = True,
    n_jackknife_folds: int | None = None,
    **kwargs: Any,
):
    """
    Initialize the Parallel Portfolio Selector.

    Args:
        model_class: The predictor class to use for performance modeling.
        p_intersection: Threshold for distribution overlap.
                      Higher values -> smaller portfolios.
        n_estimators_for_std: Number of Jackknife iterations (n_instances) or bootstrap samples.
        random_state: Random seed for reproducibility.
        use_jackknife: If True, use Jackknife method; if False, use bootstrap.
        n_jackknife_folds: Number of folds for Jackknife. If None, uses leave-one-out.
                          E.g., 10 means 10-fold cross-validation (10 models per algorithm).
        **kwargs: Additional arguments passed to parent class.
    """
    super().__init__(**kwargs)
    self.model_class = model_class
    self.p_intersection = float(p_intersection)
    self.n_estimators_for_std = int(n_estimators_for_std)
    self.random_state = int(random_state)
    self.use_jackknife = bool(use_jackknife)
    self.n_jackknife_folds = n_jackknife_folds

    self.predictors: (
        List[List[tuple[AbstractPredictor, Any]]] | List[List[AbstractPredictor]]
    ) = []
    self.algorithms: List[str] = []
    self.features_train: pd.DataFrame | None = None
    self.performance_train: pd.DataFrame | None = None

AbstractFeatureGenerator

Abstract base class for generating additional features.

Subclasses should implement the methods to define specific feature generation logic based on a set of base features.

Source code in asf/selectors/feature_generator.py

class AbstractFeatureGenerator:
    """
    Abstract base class for generating additional features.

    Subclasses should implement the methods to define specific feature
    generation logic based on a set of base features.
    """

    def __init__(self, **kwargs: Any) -> None:
        """
        Initialize the AbstractFeatureGenerator.

        Parameters
        ----------
        **kwargs : Any
            Additional keyword arguments.
        """
        pass

    def fit(
        self,
        features: pd.DataFrame,
        performance: pd.DataFrame,
        algorithm_features: pd.DataFrame | None = None,
        **kwargs: Any,
    ) -> None:
        """
        Fit the generator to the data.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame
            The algorithm performance data.
        algorithm_features : pd.DataFrame or None, optional
            Additional features related to algorithms.
        **kwargs : Any
            Additional keyword arguments.
        """
        pass

    def generate_features(self, base_features: pd.DataFrame) -> pd.DataFrame:
        """
                Generate additional features based on the provided base features.

                Parameters
                ----------
                base_features : pd.DataFrame
                    The input DataFrame containing the base features.

                Returns
        -------
                pd.DataFrame
                    A DataFrame containing the generated features.
        """
        raise NotImplementedError("Subclasses must implement generate_features.")

    @staticmethod
    def get_configuration_space(**kwargs: Any) -> Any:
        """
        Get the configuration space.
        """
        return None

__init__(**kwargs)

Initialize the AbstractFeatureGenerator.

Parameters

**kwargs : Any Additional keyword arguments.

Source code in asf/selectors/feature_generator.py

def __init__(self, **kwargs: Any) -> None:
    """
    Initialize the AbstractFeatureGenerator.

    Parameters
    ----------
    **kwargs : Any
        Additional keyword arguments.
    """
    pass

fit(features, performance, algorithm_features=None, **kwargs)

Fit the generator to the data.

Parameters

features : pd.DataFrame The input features. performance : pd.DataFrame The algorithm performance data. algorithm_features : pd.DataFrame or None, optional Additional features related to algorithms. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/feature_generator.py

def fit(
    self,
    features: pd.DataFrame,
    performance: pd.DataFrame,
    algorithm_features: pd.DataFrame | None = None,
    **kwargs: Any,
) -> None:
    """
    Fit the generator to the data.

    Parameters
    ----------
    features : pd.DataFrame
        The input features.
    performance : pd.DataFrame
        The algorithm performance data.
    algorithm_features : pd.DataFrame or None, optional
        Additional features related to algorithms.
    **kwargs : Any
        Additional keyword arguments.
    """
    pass

generate_features(base_features)

    Generate additional features based on the provided base features.

    Parameters
    ----------
    base_features : pd.DataFrame
        The input DataFrame containing the base features.

    Returns

---

    pd.DataFrame
        A DataFrame containing the generated features.

Source code in asf/selectors/feature_generator.py

def generate_features(self, base_features: pd.DataFrame) -> pd.DataFrame:
    """
            Generate additional features based on the provided base features.

            Parameters
            ----------
            base_features : pd.DataFrame
                The input DataFrame containing the base features.

            Returns
    -------
            pd.DataFrame
                A DataFrame containing the generated features.
    """
    raise NotImplementedError("Subclasses must implement generate_features.")

get_configuration_space(**kwargs) staticmethod

Get the configuration space.

Source code in asf/selectors/feature_generator.py

@staticmethod
def get_configuration_space(**kwargs: Any) -> Any:
    """
    Get the configuration space.
    """
    return None

AbstractModelBasedSelector

Bases: AbstractSelector

Abstract base class for selectors that utilize a machine learning model.

This class provides functionality to initialize with a model class, save the selector to a file, and load it back.

Attributes

model_class : Callable A callable that represents the model class to be used.

Source code in asf/selectors/abstract_model_based_selector.py

class AbstractModelBasedSelector(AbstractSelector):
    """
    Abstract base class for selectors that utilize a machine learning model.

    This class provides functionality to initialize with a model class,
    save the selector to a file, and load it back.

    Attributes
    ----------
    model_class : Callable
        A callable that represents the model class to be used.
    """

    def __init__(
        self,
        model_class: type[AbstractPredictor] | Callable[..., Any],
        **kwargs: Any,
    ) -> None:
        """
        Initialize the AbstractModelBasedSelector.

        Parameters
        ----------
        model_class : type[AbstractPredictor] or Callable
            The model class or a callable that returns a model instance.
            If a scikit-learn compatible class is provided, it's wrapped with SklearnWrapper.
        **kwargs : Any
            Additional keyword arguments passed to the parent class initializer.
        """
        super().__init__(**kwargs)

        if isinstance(model_class, type) and issubclass(
            model_class, (ClassifierMixin, RegressorMixin)
        ):
            self.model_class: Callable[..., Any] = partial(SklearnWrapper, model_class)
        else:
            self.model_class = model_class

    def save(self, path: str | Path) -> None:
        """
        Save the selector instance to the specified file path.

        Parameters
        ----------
        path : str or Path
            The file path to save the selector.
        """
        joblib.dump(self, path)

    @classmethod
    def load(cls, path: str | Path) -> AbstractModelBasedSelector:
        """
                Load a selector instance from the specified file path.

                Parameters
                ----------
                path : str or Path
                    The file path to load the selector from.

                Returns
        -------
                AbstractModelBasedSelector
                    The loaded selector instance.
        """
        return joblib.load(path)

__init__(model_class, **kwargs)

Initialize the AbstractModelBasedSelector.

Parameters

model_class : type[AbstractPredictor] or Callable The model class or a callable that returns a model instance. If a scikit-learn compatible class is provided, it's wrapped with SklearnWrapper. **kwargs : Any Additional keyword arguments passed to the parent class initializer.

Source code in asf/selectors/abstract_model_based_selector.py

def __init__(
    self,
    model_class: type[AbstractPredictor] | Callable[..., Any],
    **kwargs: Any,
) -> None:
    """
    Initialize the AbstractModelBasedSelector.

    Parameters
    ----------
    model_class : type[AbstractPredictor] or Callable
        The model class or a callable that returns a model instance.
        If a scikit-learn compatible class is provided, it's wrapped with SklearnWrapper.
    **kwargs : Any
        Additional keyword arguments passed to the parent class initializer.
    """
    super().__init__(**kwargs)

    if isinstance(model_class, type) and issubclass(
        model_class, (ClassifierMixin, RegressorMixin)
    ):
        self.model_class: Callable[..., Any] = partial(SklearnWrapper, model_class)
    else:
        self.model_class = model_class

load(path) classmethod

    Load a selector instance from the specified file path.

    Parameters
    ----------
    path : str or Path
        The file path to load the selector from.

    Returns

---

    AbstractModelBasedSelector
        The loaded selector instance.

Source code in asf/selectors/abstract_model_based_selector.py

@classmethod
def load(cls, path: str | Path) -> AbstractModelBasedSelector:
    """
            Load a selector instance from the specified file path.

            Parameters
            ----------
            path : str or Path
                The file path to load the selector from.

            Returns
    -------
            AbstractModelBasedSelector
                The loaded selector instance.
    """
    return joblib.load(path)

save(path)

Save the selector instance to the specified file path.

Parameters

path : str or Path The file path to save the selector.

Source code in asf/selectors/abstract_model_based_selector.py

def save(self, path: str | Path) -> None:
    """
    Save the selector instance to the specified file path.

    Parameters
    ----------
    path : str or Path
        The file path to save the selector.
    """
    joblib.dump(self, path)

AbstractSelector

Bases: ABC

Abstract base class for algorithm selectors.

Provides a framework for fitting, predicting, and managing hierarchical feature generators and configuration spaces.

Attributes

maximize : bool Indicates whether the objective is to maximize or minimize the performance metric. budget : float or None The budget for the selector, if applicable. feature_groups : list[str] or None Groups of features to be considered during selection. hierarchical_generator : AbstractFeatureGenerator or None A generator for hierarchical features, if applicable. algorithm_features : pd.DataFrame or None Additional features related to algorithms, if provided. prediction_mode : str Mode for predictions ('aslib', 'pandas', 'numpy'). algorithms : list[str] List of algorithm names seen during fitting. features : list[str] List of feature names seen during fitting.

Source code in asf/selectors/abstract_selector.py

class AbstractSelector(ABC):
    """
    Abstract base class for algorithm selectors.

    Provides a framework for fitting, predicting, and managing hierarchical feature
    generators and configuration spaces.

    Attributes
    ----------
    maximize : bool
        Indicates whether the objective is to maximize or minimize the performance metric.
    budget : float or None
        The budget for the selector, if applicable.
    feature_groups : list[str] or None
        Groups of features to be considered during selection.
    hierarchical_generator : AbstractFeatureGenerator or None
        A generator for hierarchical features, if applicable.
    algorithm_features : pd.DataFrame or None
        Additional features related to algorithms, if provided.
    prediction_mode : str
        Mode for predictions ('aslib', 'pandas', 'numpy').
    algorithms : list[str]
        List of algorithm names seen during fitting.
    features : list[str]
        List of feature names seen during fitting.
    """

    def __init__(
        self,
        budget: float | None = None,
        maximize: bool = False,
        feature_groups: list[str] | None = None,
        hierarchical_generator: AbstractFeatureGenerator | None = None,
        prediction_mode: str = "aslib",
        **kwargs: Any,
    ) -> None:
        """
        Initialize the AbstractSelector.

        Parameters
        ----------
        budget : float or None, default=None
            The budget for the selector, if applicable.
        maximize : bool, default=False
            Indicates whether to maximize the performance metric.
        feature_groups : list[str] or None, default=None
            Groups of features to be considered during selection.
        hierarchical_generator : AbstractFeatureGenerator or None, default=None
            A generator for hierarchical features, if applicable.
        prediction_mode : str, default="aslib"
            Mode for predictions ('aslib', 'pandas', 'numpy').
        **kwargs : Any
            Additional keyword arguments.
        """
        self.maximize = bool(maximize)
        self.budget = float(budget) if budget is not None else None
        self.feature_groups = feature_groups
        self.hierarchical_generator = hierarchical_generator
        self.algorithm_features: pd.DataFrame | None = None
        self.prediction_mode = str(prediction_mode)
        self.algorithms: list[str] = []
        self.features: list[str] = []

    def fit(
        self,
        features: pd.DataFrame | np.ndarray,
        performance: pd.DataFrame | np.ndarray,
        algorithm_features: pd.DataFrame | None = None,
        **kwargs: Any,
    ) -> None:
        """
        Fit the selector.

        Parameters
        ----------
        features : pd.DataFrame or np.ndarray
            The input features.
        performance : pd.DataFrame or np.ndarray
            The algorithm performance data.
        algorithm_features : pd.DataFrame or None, optional
            Additional features related to algorithms.
        **kwargs : Any
            Additional keyword arguments for fitting.
        """
        if (
            isinstance(features, pd.DataFrame) and isinstance(performance, np.ndarray)
        ) or (
            isinstance(features, np.ndarray) and isinstance(performance, pd.DataFrame)
        ):
            raise ValueError(
                "Mixed input types (DataFrame and numpy array) are not allowed."
            )
        if isinstance(features, np.ndarray):
            features = pd.DataFrame(
                features,
                columns=pd.Index([f"f_{i}" for i in range(features.shape[1])]),
            )
        if isinstance(performance, np.ndarray):
            performance = pd.DataFrame(
                performance,
                columns=pd.Index([f"algo_{i}" for i in range(performance.shape[1])]),
            )

        if not isinstance(features, pd.DataFrame) or not isinstance(
            performance, pd.DataFrame
        ):
            raise ValueError(
                "Features and performance must be pandas DataFrames or numpy arrays."
            )

        if self.hierarchical_generator is not None:
            self.hierarchical_generator.fit(features, performance, algorithm_features)
            features = pd.concat(
                [features, self.hierarchical_generator.generate_features(features)],
                axis=1,
            )

        self.algorithms = [str(a) for a in performance.columns]
        self.features = [str(f) for f in features.columns]
        self.algorithm_features = algorithm_features

        self._fit(features, performance, **kwargs)

    def predict(
        self,
        features: pd.DataFrame | np.ndarray | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[Any, list[tuple[str, float]]] | pd.Series | np.ndarray:
        """
        Predict algorithm selections/rankings.

        Parameters
        ----------
        features : pd.DataFrame or np.ndarray or None
            The input features for prediction.
        performance : pd.DataFrame or None, default=None
            Partial performance data if available (e.g., for oracle selectors).

        Returns
        -------
        dict or pd.Series or np.ndarray
            Predicted selections in the specified prediction_mode.
            For dict return type, each list item is a tuple:
            - (algorithm_name, budget): algorithm selection with time budget
        """
        if features is None:
            df_features: pd.DataFrame | None = None
        elif isinstance(features, np.ndarray):
            cols = (
                self.features
                if self.features
                else [f"f_{i}" for i in range(features.shape[1])]
            )
            df_features = pd.DataFrame(features, columns=pd.Index(cols))
        elif isinstance(features, pd.DataFrame):
            df_features = features
        else:
            raise ValueError(
                "Features must be a numpy array, pandas DataFrame, or None."
            )

        if self.hierarchical_generator is not None and df_features is not None:
            df_features = pd.concat(
                [
                    df_features,
                    self.hierarchical_generator.generate_features(df_features),
                ],
                axis=1,
            )

        scheds = self._predict(df_features, performance=performance)

        def _lookup_schedule(instance: Any) -> list[tuple[str, float]]:
            if instance in scheds:
                return list(scheds[instance])
            instance_str = str(instance)
            if instance_str in scheds:
                return list(scheds[instance_str])
            return []

        if self.prediction_mode == "aslib":
            if df_features is None:
                return scheds

            if self.feature_groups is None:
                return {
                    instance: _lookup_schedule(instance)
                    for instance in df_features.index
                }

            fg_steps = list(self.feature_groups)
            return {
                instance: fg_steps + _lookup_schedule(instance)
                for instance in df_features.index
            }
        elif self.prediction_mode == "pandas":
            if df_features is None:
                raise ValueError("Pandas mode requires features.")
            return pd.Series(
                {
                    instance: (_lookup_schedule(instance) or [(None, 0.0)])[0][0]
                    for instance in df_features.index
                }
            )
        elif self.prediction_mode == "numpy":
            if df_features is None:
                raise ValueError("Numpy mode requires features.")
            labels = [
                (_lookup_schedule(instance) or [(None, 0.0)])[0][0]
                for instance in df_features.index
            ]
            encoder = OneHotEncoder(sparse_output=False, categories=[self.algorithms])
            return encoder.fit_transform(np.array(labels).reshape(-1, 1))
        else:
            raise ValueError(f"Unknown prediction_mode: {self.prediction_mode}")

    def save(self, path: str) -> None:
        """
        Save the selector instance.

        Parameters
        ----------
        path : str
            File path to save to.
        """
        import joblib

        joblib.dump(self, path)

    @classmethod
    def load(cls, path: str) -> AbstractSelector:
        """
        Load a selector instance.

        Parameters
        ----------
        path : str
            File path to load from.

        Returns
        -------
        AbstractSelector
            The loaded selector instance.
        """
        import joblib

        return joblib.load(path)

    @abstractmethod
    def _fit(
        self,
        features: pd.DataFrame,
        performance: pd.DataFrame,
        **kwargs: Any,
    ) -> None:
        """
        Internal fit implementation.
        """
        pass

    @abstractmethod
    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Internal predict implementation.
        """
        pass

__init__(budget=None, maximize=False, feature_groups=None, hierarchical_generator=None, prediction_mode='aslib', **kwargs)

Initialize the AbstractSelector.

Parameters

budget : float or None, default=None The budget for the selector, if applicable. maximize : bool, default=False Indicates whether to maximize the performance metric. feature_groups : list[str] or None, default=None Groups of features to be considered during selection. hierarchical_generator : AbstractFeatureGenerator or None, default=None A generator for hierarchical features, if applicable. prediction_mode : str, default="aslib" Mode for predictions ('aslib', 'pandas', 'numpy'). **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/abstract_selector.py

def __init__(
    self,
    budget: float | None = None,
    maximize: bool = False,
    feature_groups: list[str] | None = None,
    hierarchical_generator: AbstractFeatureGenerator | None = None,
    prediction_mode: str = "aslib",
    **kwargs: Any,
) -> None:
    """
    Initialize the AbstractSelector.

    Parameters
    ----------
    budget : float or None, default=None
        The budget for the selector, if applicable.
    maximize : bool, default=False
        Indicates whether to maximize the performance metric.
    feature_groups : list[str] or None, default=None
        Groups of features to be considered during selection.
    hierarchical_generator : AbstractFeatureGenerator or None, default=None
        A generator for hierarchical features, if applicable.
    prediction_mode : str, default="aslib"
        Mode for predictions ('aslib', 'pandas', 'numpy').
    **kwargs : Any
        Additional keyword arguments.
    """
    self.maximize = bool(maximize)
    self.budget = float(budget) if budget is not None else None
    self.feature_groups = feature_groups
    self.hierarchical_generator = hierarchical_generator
    self.algorithm_features: pd.DataFrame | None = None
    self.prediction_mode = str(prediction_mode)
    self.algorithms: list[str] = []
    self.features: list[str] = []

fit(features, performance, algorithm_features=None, **kwargs)

Fit the selector.

Parameters

features : pd.DataFrame or np.ndarray The input features. performance : pd.DataFrame or np.ndarray The algorithm performance data. algorithm_features : pd.DataFrame or None, optional Additional features related to algorithms. **kwargs : Any Additional keyword arguments for fitting.

Source code in asf/selectors/abstract_selector.py

def fit(
    self,
    features: pd.DataFrame | np.ndarray,
    performance: pd.DataFrame | np.ndarray,
    algorithm_features: pd.DataFrame | None = None,
    **kwargs: Any,
) -> None:
    """
    Fit the selector.

    Parameters
    ----------
    features : pd.DataFrame or np.ndarray
        The input features.
    performance : pd.DataFrame or np.ndarray
        The algorithm performance data.
    algorithm_features : pd.DataFrame or None, optional
        Additional features related to algorithms.
    **kwargs : Any
        Additional keyword arguments for fitting.
    """
    if (
        isinstance(features, pd.DataFrame) and isinstance(performance, np.ndarray)
    ) or (
        isinstance(features, np.ndarray) and isinstance(performance, pd.DataFrame)
    ):
        raise ValueError(
            "Mixed input types (DataFrame and numpy array) are not allowed."
        )
    if isinstance(features, np.ndarray):
        features = pd.DataFrame(
            features,
            columns=pd.Index([f"f_{i}" for i in range(features.shape[1])]),
        )
    if isinstance(performance, np.ndarray):
        performance = pd.DataFrame(
            performance,
            columns=pd.Index([f"algo_{i}" for i in range(performance.shape[1])]),
        )

    if not isinstance(features, pd.DataFrame) or not isinstance(
        performance, pd.DataFrame
    ):
        raise ValueError(
            "Features and performance must be pandas DataFrames or numpy arrays."
        )

    if self.hierarchical_generator is not None:
        self.hierarchical_generator.fit(features, performance, algorithm_features)
        features = pd.concat(
            [features, self.hierarchical_generator.generate_features(features)],
            axis=1,
        )

    self.algorithms = [str(a) for a in performance.columns]
    self.features = [str(f) for f in features.columns]
    self.algorithm_features = algorithm_features

    self._fit(features, performance, **kwargs)

load(path) classmethod

Load a selector instance.

Parameters

path : str File path to load from.

Returns

AbstractSelector The loaded selector instance.

Source code in asf/selectors/abstract_selector.py

@classmethod
def load(cls, path: str) -> AbstractSelector:
    """
    Load a selector instance.

    Parameters
    ----------
    path : str
        File path to load from.

    Returns
    -------
    AbstractSelector
        The loaded selector instance.
    """
    import joblib

    return joblib.load(path)

predict(features, performance=None)

Predict algorithm selections/rankings.

Parameters

features : pd.DataFrame or np.ndarray or None The input features for prediction. performance : pd.DataFrame or None, default=None Partial performance data if available (e.g., for oracle selectors).

Returns

dict or pd.Series or np.ndarray Predicted selections in the specified prediction_mode. For dict return type, each list item is a tuple: - (algorithm_name, budget): algorithm selection with time budget

Source code in asf/selectors/abstract_selector.py

def predict(
    self,
    features: pd.DataFrame | np.ndarray | None,
    performance: pd.DataFrame | None = None,
) -> dict[Any, list[tuple[str, float]]] | pd.Series | np.ndarray:
    """
    Predict algorithm selections/rankings.

    Parameters
    ----------
    features : pd.DataFrame or np.ndarray or None
        The input features for prediction.
    performance : pd.DataFrame or None, default=None
        Partial performance data if available (e.g., for oracle selectors).

    Returns
    -------
    dict or pd.Series or np.ndarray
        Predicted selections in the specified prediction_mode.
        For dict return type, each list item is a tuple:
        - (algorithm_name, budget): algorithm selection with time budget
    """
    if features is None:
        df_features: pd.DataFrame | None = None
    elif isinstance(features, np.ndarray):
        cols = (
            self.features
            if self.features
            else [f"f_{i}" for i in range(features.shape[1])]
        )
        df_features = pd.DataFrame(features, columns=pd.Index(cols))
    elif isinstance(features, pd.DataFrame):
        df_features = features
    else:
        raise ValueError(
            "Features must be a numpy array, pandas DataFrame, or None."
        )

    if self.hierarchical_generator is not None and df_features is not None:
        df_features = pd.concat(
            [
                df_features,
                self.hierarchical_generator.generate_features(df_features),
            ],
            axis=1,
        )

    scheds = self._predict(df_features, performance=performance)

    def _lookup_schedule(instance: Any) -> list[tuple[str, float]]:
        if instance in scheds:
            return list(scheds[instance])
        instance_str = str(instance)
        if instance_str in scheds:
            return list(scheds[instance_str])
        return []

    if self.prediction_mode == "aslib":
        if df_features is None:
            return scheds

        if self.feature_groups is None:
            return {
                instance: _lookup_schedule(instance)
                for instance in df_features.index
            }

        fg_steps = list(self.feature_groups)
        return {
            instance: fg_steps + _lookup_schedule(instance)
            for instance in df_features.index
        }
    elif self.prediction_mode == "pandas":
        if df_features is None:
            raise ValueError("Pandas mode requires features.")
        return pd.Series(
            {
                instance: (_lookup_schedule(instance) or [(None, 0.0)])[0][0]
                for instance in df_features.index
            }
        )
    elif self.prediction_mode == "numpy":
        if df_features is None:
            raise ValueError("Numpy mode requires features.")
        labels = [
            (_lookup_schedule(instance) or [(None, 0.0)])[0][0]
            for instance in df_features.index
        ]
        encoder = OneHotEncoder(sparse_output=False, categories=[self.algorithms])
        return encoder.fit_transform(np.array(labels).reshape(-1, 1))
    else:
        raise ValueError(f"Unknown prediction_mode: {self.prediction_mode}")

save(path)

Save the selector instance.

Parameters

path : str File path to save to.

Source code in asf/selectors/abstract_selector.py

def save(self, path: str) -> None:
    """
    Save the selector instance.

    Parameters
    ----------
    path : str
        File path to save to.
    """
    import joblib

    joblib.dump(self, path)

BaggingSelector

Bases: ConfigurableMixin, AbstractSelector

Bagging ensemble for algorithm selectors.

Trains multiple instances of the same base selector on bootstrap samples of the training data and aggregates predictions via voting.

Attributes

base_selector : AbstractSelector The base selector to bag. n_estimators : int Number of bootstrap samples/selectors to train. sample_fraction : float Fraction of samples to use in each bootstrap sample. random_state : int or None Random state for reproducibility. estimators_ : list[AbstractSelector] Fitted estimators after calling fit().

Source code in asf/selectors/ensembles.py

class BaggingSelector(ConfigurableMixin, AbstractSelector):
    """Bagging ensemble for algorithm selectors.

    Trains multiple instances of the same base selector on bootstrap
    samples of the training data and aggregates predictions via voting.

    Attributes
    ----------
    base_selector : AbstractSelector
        The base selector to bag.
    n_estimators : int
        Number of bootstrap samples/selectors to train.
    sample_fraction : float
        Fraction of samples to use in each bootstrap sample.
    random_state : int or None
        Random state for reproducibility.
    estimators_ : list[AbstractSelector]
        Fitted estimators after calling fit().
    """

    PREFIX = "bagging"
    RETURN_TYPE = "single"

    def __init__(
        self,
        base_selector: AbstractSelector | Callable[[], AbstractSelector] | None = None,
        n_estimators: int = 10,
        sample_fraction: float = 1.0,
        random_state: int | None = None,
        **kwargs: Any,
    ) -> None:
        """Initialize the BaggingSelector.

        Parameters
        ----------
        base_selector : AbstractSelector or Callable or None
            The base selector to bag. Can be an instance or a callable
            that returns one.
        n_estimators : int, default=10
            Number of bootstrap samples/selectors to train.
        sample_fraction : float, default=1.0
            Fraction of samples to use in each bootstrap sample.
        random_state : int or None, default=None
            Random state for reproducibility.
        **kwargs : Any
            Additional keyword arguments passed to AbstractSelector.
        """
        super().__init__(**kwargs)

        if callable(base_selector) and not isinstance(base_selector, AbstractSelector):
            self.base_selector = base_selector()
        else:
            self.base_selector = base_selector

        if self.base_selector is None:
            raise ValueError("base_selector cannot be None")

        self.n_estimators = int(n_estimators)
        self.sample_fraction = float(sample_fraction)
        self.random_state = random_state
        self.estimators_: list[AbstractSelector] = []

    def _fit(
        self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs: Any
    ) -> None:
        """Fit the bagging ensemble.

        Parameters
        ----------
        features : pd.DataFrame
            Training features.
        performance : pd.DataFrame
            Algorithm performance data.
        **kwargs : Any
            Additional keyword arguments.
        """
        rng = np.random.default_rng(self.random_state)
        n_samples = len(features)
        sample_size = max(1, int(n_samples * self.sample_fraction))

        self.estimators_ = []

        for _ in range(self.n_estimators):
            # Bootstrap sample with replacement
            indices = rng.choice(n_samples, size=sample_size, replace=True)

            # Use iloc to sample and copy to ensure independent DataFrames
            X_sample = features.iloc[indices].copy()
            X_sample.index = pd.RangeIndex(len(X_sample))
            y_sample = performance.iloc[indices].copy()
            y_sample.index = pd.RangeIndex(len(y_sample))

            assert self.base_selector is not None
            estimator = _clone_selector(self.base_selector)
            estimator.fit(X_sample, y_sample, **kwargs)
            self.estimators_.append(estimator)

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """Predict using the bagging ensemble.

        Parameters
        ----------
        features : pd.DataFrame or None
            Input features for prediction.
        performance : pd.DataFrame or None, default=None
            Performance data (unused).

        Returns
        -------
        dict
            Mapping from instance names to algorithm schedules.
        """
        if features is None:
            raise ValueError("BaggingSelector requires features for prediction.")

        if not self.estimators_:
            raise RuntimeError("BaggingSelector has not been fitted.")

        # Collect votes from all estimators
        all_votes: dict[str, dict[str, int]] = {str(idx): {} for idx in features.index}

        for estimator in self.estimators_:
            preds = estimator.predict(features)
            if isinstance(preds, dict):
                for instance, schedule in preds.items():
                    if schedule:
                        assert isinstance(schedule, list)
                        first_item = schedule[0]
                        assert isinstance(first_item, tuple) and len(first_item) >= 1
                        algo = str(first_item[0])
                        votes_for_instance = all_votes[str(instance)]
                        votes_for_instance[algo] = votes_for_instance.get(algo, 0) + 1

        # Aggregate votes
        result: dict[str, list[tuple[str, float]]] = {}
        for instance in features.index:
            votes = all_votes[str(instance)]
            if votes:
                best_algo = max(votes.keys(), key=lambda a: votes[a])
                result[str(instance)] = [(best_algo, float(self.budget or 0))]
            else:
                result[str(instance)] = []

        return result

    @staticmethod
    def _define_hyperparameters(
        base_selector_class: list[type] | None = None,
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """Define hyperparameters for BaggingSelector.

        Parameters
        ----------
        base_selector_class : list[type] or None, default=None
            List of selector classes to choose from.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        hyperparameters: list[Any] = [
            Integer("n_estimators", bounds=(2, 50), default=10),
            UniformFloatHyperparameter(
                "sample_fraction", lower=0.5, upper=1.0, default_value=1.0
            ),
        ]

        if base_selector_class:
            hyperparameters.append(
                ClassChoice(
                    "base_selector",
                    choices=cast(list[type | bool], base_selector_class),
                )
            )

        return hyperparameters, [], []

__init__(base_selector=None, n_estimators=10, sample_fraction=1.0, random_state=None, **kwargs)

Initialize the BaggingSelector.

Parameters

base_selector : AbstractSelector or Callable or None The base selector to bag. Can be an instance or a callable that returns one. n_estimators : int, default=10 Number of bootstrap samples/selectors to train. sample_fraction : float, default=1.0 Fraction of samples to use in each bootstrap sample. random_state : int or None, default=None Random state for reproducibility. **kwargs : Any Additional keyword arguments passed to AbstractSelector.

Source code in asf/selectors/ensembles.py

def __init__(
    self,
    base_selector: AbstractSelector | Callable[[], AbstractSelector] | None = None,
    n_estimators: int = 10,
    sample_fraction: float = 1.0,
    random_state: int | None = None,
    **kwargs: Any,
) -> None:
    """Initialize the BaggingSelector.

    Parameters
    ----------
    base_selector : AbstractSelector or Callable or None
        The base selector to bag. Can be an instance or a callable
        that returns one.
    n_estimators : int, default=10
        Number of bootstrap samples/selectors to train.
    sample_fraction : float, default=1.0
        Fraction of samples to use in each bootstrap sample.
    random_state : int or None, default=None
        Random state for reproducibility.
    **kwargs : Any
        Additional keyword arguments passed to AbstractSelector.
    """
    super().__init__(**kwargs)

    if callable(base_selector) and not isinstance(base_selector, AbstractSelector):
        self.base_selector = base_selector()
    else:
        self.base_selector = base_selector

    if self.base_selector is None:
        raise ValueError("base_selector cannot be None")

    self.n_estimators = int(n_estimators)
    self.sample_fraction = float(sample_fraction)
    self.random_state = random_state
    self.estimators_: list[AbstractSelector] = []

CSHCSelector

Bases: ConfigurableMixin, AbstractSelector

Confidence-Switching Hybrid Selector.

A meta-selector that uses a primary selector along with guardian models to predict the success probability of the primary's choice.

References

Malitsky, Y., et al. (2021). "Confidence-based Switching Hybrid Solver Construction." https://link.springer.com/chapter/10.1007/978-3-642-44973-4_17

Attributes

primary_selector : AbstractSelector The primary selector model. backup_selector : AbstractSelector or None The backup selector model. n_folds : int Number of folds for cross-validation to find the optimal threshold. random_state : int Random seed for reproducibility. guardian_kwargs : dict[str, Any] Keyword arguments for the guardian models (RandomForestClassifier). threshold_grid : np.ndarray Grid of thresholds to evaluate. guardians : dict[str, RandomForestClassifier] Trained guardian models for each algorithm. threshold : float The learned optimal threshold for switching.

Source code in asf/selectors/cshc.py

class CSHCSelector(ConfigurableMixin, AbstractSelector):
    """
    Confidence-Switching Hybrid Selector.

    A meta-selector that uses a primary selector along with guardian models to
    predict the success probability of the primary's choice.

    References
    ----------
    Malitsky, Y., et al. (2021).
    "Confidence-based Switching Hybrid Solver Construction."
    https://link.springer.com/chapter/10.1007/978-3-642-44973-4_17

    Attributes
    ----------
    primary_selector : AbstractSelector
        The primary selector model.
    backup_selector : AbstractSelector or None
        The backup selector model.
    n_folds : int
        Number of folds for cross-validation to find the optimal threshold.
    random_state : int
        Random seed for reproducibility.
    guardian_kwargs : dict[str, Any]
        Keyword arguments for the guardian models (RandomForestClassifier).
    threshold_grid : np.ndarray
        Grid of thresholds to evaluate.
    guardians : dict[str, RandomForestClassifier]
        Trained guardian models for each algorithm.
    threshold : float
        The learned optimal threshold for switching.
    """

    PREFIX = "cshc"
    RETURN_TYPE = "single"

    def __init__(
        self,
        primary_selector: AbstractSelector | Callable[[], AbstractSelector],
        backup_selector: AbstractSelector
        | Callable[[], AbstractSelector]
        | None = None,
        n_estimators: int = 100,
        guardian_kwargs: dict[str, Any] | None = None,
        n_folds: int = 5,
        threshold_grid: np.ndarray | None = None,
        random_state: int = 42,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the CSHCSelector.
        """
        super().__init__(**kwargs)

        if callable(primary_selector):
            self.primary_selector = cast(Any, primary_selector)()
        else:
            self.primary_selector = primary_selector

        if callable(backup_selector):
            self.backup_selector = cast(Any, backup_selector)()
        else:
            self.backup_selector = backup_selector

        self.n_folds = int(n_folds)
        self.random_state = int(random_state)
        self.guardian_kwargs = dict(guardian_kwargs or {})
        self.guardian_kwargs.setdefault("n_estimators", int(n_estimators))
        self.guardian_kwargs.setdefault("random_state", int(random_state))
        self.threshold_grid = (
            threshold_grid
            if threshold_grid is not None
            else np.linspace(0.01, 0.99, 99)
        )

        self.guardians: dict[str, RandomForestClassifier] = {}
        self.threshold: float = 0.5

    def _fit(
        self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs: Any
    ) -> None:
        """
        Train one guardian model per algorithm and find the optimal threshold.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame
            The performance data.
        **kwargs : Any
            Additional keyword arguments.
        """
        self.algorithms = [str(a) for a in performance.columns]
        n_instances = len(features)
        kf = KFold(
            n_splits=min(self.n_folds, n_instances),
            shuffle=True,
            random_state=self.random_state,
        )

        oof_probs = []
        oof_true_success = []

        for train_idx, val_idx in kf.split(features):
            X_train, X_val = features.iloc[train_idx], features.iloc[val_idx]
            Y_train, Y_val = performance.iloc[train_idx], performance.iloc[val_idx]

            sel_copy = copy.deepcopy(self.primary_selector)
            sel_copy.fit(X_train, Y_train)
            primary_preds = sel_copy.predict(X_val)

            fold_guardians = {}
            for algo in self.algorithms:
                y_algo_train = (Y_train[algo] <= self.budget).astype(int)
                clf = RandomForestClassifier(**self.guardian_kwargs)
                clf.fit(X_train, y_algo_train)
                fold_guardians[algo] = clf

            assert isinstance(primary_preds, dict)
            for inst_name, pred_list in primary_preds.items():
                if pred_list:
                    assert isinstance(pred_list, list) and len(pred_list) > 0
                    first_item = pred_list[0]
                    assert isinstance(first_item, tuple) and len(first_item) >= 1
                    chosen_algo = first_item[0]
                    if Y_val.index.dtype != object:
                        # Convert back to original index type if necessary
                        index_type = cast(Any, Y_val.index.dtype).type
                        orig_inst_name = index_type(inst_name)
                    else:
                        orig_inst_name = inst_name

                    runtime = Y_val.at[orig_inst_name, chosen_algo]
                    inst_feature_df = X_val.loc[[orig_inst_name]]
                    guardian_for_choice = fold_guardians.get(chosen_algo)

                    if guardian_for_choice:
                        if len(guardian_for_choice.classes_) == 1:
                            class_val = guardian_for_choice.classes_[0]
                            prob = 1.0 if class_val == 1 else 0.0
                        else:
                            prob = guardian_for_choice.predict_proba(inst_feature_df)[
                                0, 1
                            ]

                        oof_probs.append(float(prob))
                        oof_true_success.append(
                            1 if pd.notna(runtime) and runtime <= self.budget else 0
                        )

        oof_probs_arr = np.array(oof_probs)
        oof_true_success_arr = np.array(oof_true_success)

        best_t, best_ratio = 0.5, float("inf")
        for t in self.threshold_grid:
            preds = (oof_probs_arr >= t).astype(int)
            fn = ((oof_true_success_arr == 1) & (preds == 0)).sum()
            tn = ((oof_true_success_arr == 0) & (preds == 0)).sum()
            ratio = fn / tn if tn > 0 else float("inf")
            if ratio < best_ratio:
                best_ratio, best_t = float(ratio), float(t)
        self.threshold = best_t

        for algo in self.algorithms:
            y_algo = (performance[algo] <= self.budget).astype(int)
            self.guardians[str(algo)] = RandomForestClassifier(
                **self.guardian_kwargs
            ).fit(features, y_algo)

        self.primary_selector.fit(features, performance)
        if self.backup_selector:
            self.backup_selector.fit(features, performance)

    def _predict(
        self, features: pd.DataFrame | None, performance: pd.DataFrame | None = None
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict using the guardian/backup logic.

        Parameters
        ----------
        features : pd.DataFrame or None
            The query instance features.
        performance : pd.DataFrame or None, default=None
            The performance data.

        Returns
        -------
        dict
            Mapping from instance names to algorithm schedules.
        """
        if features is None:
            raise ValueError("CSHCSelector requires features for prediction.")
        if not self.guardians:
            raise RuntimeError("The selector has not been fitted yet.")

        primary_preds = cast(dict[Any, Any], self.primary_selector.predict(features))
        final_preds: dict[str, list[tuple[str, float]]] = {}

        for inst_name in features.index:
            preds = primary_preds.get(inst_name, primary_preds.get(str(inst_name), []))
            if not isinstance(preds, list):
                preds = []

            pred_list: list[tuple[str, float]] = []
            for p in preds:
                if isinstance(p, tuple):
                    pred_list.append((str(p[0]), float(p[1])))
                else:
                    pred_list.append((str(p), float(self.budget or 0)))

            inst_feature_df = features.loc[[inst_name]]

            if not pred_list:
                if self.backup_selector:
                    backup_pred = cast(
                        dict[Any, Any], self.backup_selector.predict(inst_feature_df)
                    )
                    preds = backup_pred.get(
                        inst_name, backup_pred.get(str(inst_name), [])
                    )
                    if preds:
                        final_preds[str(inst_name)] = [
                            (str(preds[0][0]), float(self.budget or 0))
                        ]
                    else:
                        final_preds[str(inst_name)] = []
                else:
                    final_preds[str(inst_name)] = []
                continue

            chosen_algo_primary = pred_list[0][0]

            guardian_for_choice = self.guardians.get(str(chosen_algo_primary))
            prob_success = 0.0
            if guardian_for_choice:
                if len(guardian_for_choice.classes_) == 1:
                    prob_success = 1.0 if guardian_for_choice.classes_[0] == 1 else 0.0
                else:
                    prob_success = float(
                        guardian_for_choice.predict_proba(inst_feature_df)[0, 1]
                    )

            if prob_success >= self.threshold:
                final_preds[str(inst_name)] = [
                    (str(chosen_algo_primary), float(self.budget or 0))
                ]
            elif self.backup_selector:
                backup_pred = cast(
                    dict[Any, Any], self.backup_selector.predict(inst_feature_df)
                )
                preds = backup_pred.get(inst_name, backup_pred.get(str(inst_name), []))
                if preds:
                    final_preds[str(inst_name)] = [
                        (str(preds[0][0]), float(self.budget or 0))
                    ]
                else:
                    final_preds[str(inst_name)] = []
            else:
                best_algo = chosen_algo_primary
                max_prob = -1.0
                for algo, guardian in self.guardians.items():
                    if len(guardian.classes_) == 1:
                        prob = 1.0 if guardian.classes_[0] == 1 else 0.0
                    else:
                        prob = float(guardian.predict_proba(inst_feature_df)[0, 1])
                    if prob > max_prob:
                        max_prob = prob
                        best_algo = algo
                final_preds[str(inst_name)] = [
                    (str(best_algo), float(self.budget or 0))
                ]

        return final_preds

    @staticmethod
    def _define_hyperparameters(
        candidate_selectors: list[type] | None = None, **kwargs: Any
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for CSHCSelector.

        Parameters
        ----------
        candidate_selectors : list[type] or None, default=None
            List of selector classes to choose from.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        if candidate_selectors is None:
            return [], [], []

        primary_selector_param = ClassChoice(
            name="primary_selector",
            choices=cast(list[type | bool], candidate_selectors),
            default=candidate_selectors[0],
        )

        use_backup = Categorical(
            name="use_backup_selector",
            items=[True, False],
            default=False,
        )

        backup_selector_param = ClassChoice(
            name="backup_selector",
            choices=cast(list[type | bool], candidate_selectors),
            default=candidate_selectors[0],
        )

        n_estimators_param = Integer(
            name="n_estimators",
            bounds=(10, 200),
            default=100,
        )

        n_folds_param = Integer(
            name="n_folds",
            bounds=(2, 10),
            default=5,
        )

        params = [
            primary_selector_param,
            use_backup,
            cast(Any, backup_selector_param),
            n_estimators_param,
            n_folds_param,
        ]

        conditions = [
            EqualsCondition(cast(Any, backup_selector_param), use_backup, True),
        ]

        return params, conditions, []

__init__(primary_selector, backup_selector=None, n_estimators=100, guardian_kwargs=None, n_folds=5, threshold_grid=None, random_state=42, **kwargs)

Initialize the CSHCSelector.

Source code in asf/selectors/cshc.py

def __init__(
    self,
    primary_selector: AbstractSelector | Callable[[], AbstractSelector],
    backup_selector: AbstractSelector
    | Callable[[], AbstractSelector]
    | None = None,
    n_estimators: int = 100,
    guardian_kwargs: dict[str, Any] | None = None,
    n_folds: int = 5,
    threshold_grid: np.ndarray | None = None,
    random_state: int = 42,
    **kwargs: Any,
) -> None:
    """
    Initialize the CSHCSelector.
    """
    super().__init__(**kwargs)

    if callable(primary_selector):
        self.primary_selector = cast(Any, primary_selector)()
    else:
        self.primary_selector = primary_selector

    if callable(backup_selector):
        self.backup_selector = cast(Any, backup_selector)()
    else:
        self.backup_selector = backup_selector

    self.n_folds = int(n_folds)
    self.random_state = int(random_state)
    self.guardian_kwargs = dict(guardian_kwargs or {})
    self.guardian_kwargs.setdefault("n_estimators", int(n_estimators))
    self.guardian_kwargs.setdefault("random_state", int(random_state))
    self.threshold_grid = (
        threshold_grid
        if threshold_grid is not None
        else np.linspace(0.01, 0.99, 99)
    )

    self.guardians: dict[str, RandomForestClassifier] = {}
    self.threshold: float = 0.5

CollaborativeFilteringSelector

Bases: ConfigurableMixin, AbstractModelBasedSelector

Collaborative filtering selector using SGD matrix factorization (ALORS-style).

References

Misir, M., and Sebag, M. (2016). "ALORS: An algorithm recommender system." https://www.sciencedirect.com/science/article/pii/S0004370216301436

Attributes

n_components : int Number of latent factors. n_iter : int Number of iterations for SGD. lr : float Learning rate for SGD. reg : float Regularization strength. random_state : int Random seed for initialization. U : np.ndarray or None Instance latent factors. V : np.ndarray or None Algorithm latent factors. performance_matrix : pd.DataFrame or None The performance data used for training. model : Any or None The regressor model to predict latent factors from features. mu : float or None Global mean of observed performance entries. b_U : np.ndarray or None Instance biases. b_V : np.ndarray or None Algorithm biases.

Source code in asf/selectors/collaborative_filtering_selector.py

class CollaborativeFilteringSelector(ConfigurableMixin, AbstractModelBasedSelector):
    """
    Collaborative filtering selector using SGD matrix factorization (ALORS-style).

    References
    ----------
    Misir, M., and Sebag, M. (2016).
    "ALORS: An algorithm recommender system."
    https://www.sciencedirect.com/science/article/pii/S0004370216301436

    Attributes
    ----------
    n_components : int
        Number of latent factors.
    n_iter : int
        Number of iterations for SGD.
    lr : float
        Learning rate for SGD.
    reg : float
        Regularization strength.
    random_state : int
        Random seed for initialization.
    U : np.ndarray or None
        Instance latent factors.
    V : np.ndarray or None
        Algorithm latent factors.
    performance_matrix : pd.DataFrame or None
        The performance data used for training.
    model : Any or None
        The regressor model to predict latent factors from features.
    mu : float or None
        Global mean of observed performance entries.
    b_U : np.ndarray or None
        Instance biases.
    b_V : np.ndarray or None
        Algorithm biases.
    """

    PREFIX = "collaborative_filtering"
    RETURN_TYPE = "single"

    def __init__(
        self,
        model_class: type | Callable[..., Any] = RidgeRegressorWrapper,
        n_components: int = 10,
        n_iter: int = 100,
        lr: float = 0.001,
        reg: float = 0.1,
        random_state: int = 42,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the CollaborativeFilteringSelector.

        Parameters
        ----------
        model_class : type or Callable, default=RidgeRegressorWrapper
            The regressor wrapper to predict latent factors from features.
        n_components : int, default=10
            Number of latent factors.
        n_iter : int, default=100
            Number of iterations for SGD.
        lr : float, default=0.001
            Learning rate for SGD.
        reg : float, default=0.1
            Regularization strength.
        random_state : int, default=42
            Random seed for initialization.
        **kwargs : Any
            Additional keyword arguments for the parent classes.
        """
        super().__init__(model_class=model_class, **kwargs)
        self.n_components = int(n_components)
        self.n_iter = int(n_iter)
        self.lr = float(lr)
        self.reg = float(reg)
        self.random_state = int(random_state)
        self.U: np.ndarray | None = None
        self.V: np.ndarray | None = None
        self.performance_matrix: pd.DataFrame | None = None
        self.model: Any | None = None

        self.mu: float | None = None
        self.b_U: np.ndarray | None = None
        self.b_V: np.ndarray | None = None

    def _fit(
        self,
        features: pd.DataFrame,
        performance: pd.DataFrame,
        **kwargs: Any,
    ) -> None:
        """
        Fit the collaborative filtering model.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame
            The algorithm performance data.
        **kwargs : Any
            Additional keyword arguments.
        """
        self.algorithms = [str(a) for a in performance.columns]
        self.performance_matrix = performance.copy()
        rng = np.random.RandomState(self.random_state)

        n_instances, n_algorithms = performance.shape
        self.U = rng.normal(scale=0.1, size=(n_instances, self.n_components)).astype(
            float
        )
        self.V = rng.normal(scale=0.1, size=(n_algorithms, self.n_components)).astype(
            float
        )

        observed = ~performance.isna()
        rows, cols = np.where(observed.values)
        vals = performance.values

        self.mu = float(np.nanmean(vals))
        self.b_U = np.zeros(n_instances, dtype=float)
        self.b_V = np.zeros(n_algorithms, dtype=float)

        for _ in range(self.n_iter):
            for i, j in zip(rows, cols):
                r_ij = float(vals[i, j])
                pred = (
                    float(self.mu or 0)
                    + self.b_U[i]
                    + self.b_V[j]
                    + float(np.dot(self.U[i], self.V[j]))
                )

                err = r_ij - pred
                err = np.clip(err, -10.0, 10.0)

                self.U[i] += self.lr * (err * self.V[j] - self.reg * self.U[i])
                self.V[j] += self.lr * (err * self.U[i] - self.reg * self.V[j])
                self.b_U[i] += self.lr * (err - self.reg * self.b_U[i])
                self.b_V[j] += self.lr * (err - self.reg * self.b_V[j])

        self.model = self.model_class()
        if self.model is None:
            raise RuntimeError("Model could not be initialized.")

        self.model.fit(features.values, self.U)

    def _predict_cold_start(
        self, instance_features: pd.Series, instance_name: str
    ) -> tuple[str, float]:
        """
        Predict for a single instance using features.

        Parameters
        ----------
        instance_features : pd.Series
            Features of the instance.
        instance_name : str
            Name of the instance.

        Returns
        -------
        tuple
            Tuple of (best_algorithm, score).
        """
        if self.model is None or self.V is None or self.b_V is None:
            raise RuntimeError("Model has not been fitted.")

        X = instance_features[self.features].values.reshape(1, -1)
        U_new = self.model.predict(X)

        scores = float(self.mu or 0) + self.b_V + np.dot(U_new, self.V.T).flatten()
        scores = np.asarray(scores, dtype=float).flatten()

        idx = int(np.argmin(scores))
        return str(self.algorithms[idx]), float(scores[idx])

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict best algorithm for instances.

        Parameters
        ----------
        features : pd.DataFrame or None, default=None
            The input features.
        performance : pd.DataFrame or None, default=None
            The performance data.

        Returns
        -------
        dict
            Mapping from instance names to algorithm schedules.
        """
        if (
            self.U is None
            or self.V is None
            or self.performance_matrix is None
            or self.b_U is None
            or self.b_V is None
        ):
            raise ValueError("Model has not been fitted.")

        predictions: dict[str, list[tuple[str, float]]] = {}

        # Case 1: Return best algorithm for training instances
        if features is None and performance is None:
            pred_matrix = (
                float(self.mu or 0)
                + self.b_U[:, None]
                + self.b_V[None, :]
                + (self.U @ self.V.T)
            )
            for idx, instance in enumerate(self.performance_matrix.index):
                scores = np.asarray(pred_matrix[idx], dtype=float).flatten()
                best_idx = int(np.argmin(scores))
                predictions[str(instance)] = [
                    (str(self.algorithms[best_idx]), float(self.budget or 0))
                ]
            return predictions

        # Case 2: Warm-start prediction with some observed performance
        if performance is not None:
            rng = np.random.RandomState(self.random_state)
            for instance in performance.index:
                perf_row = performance.loc[instance]
                if not perf_row.isnull().all():
                    u = rng.normal(scale=0.1, size=(self.n_components,)).astype(float)
                    # SGD refinement for instance factors
                    for _ in range(20):
                        for j, _ in enumerate(self.algorithms):
                            r_ij = perf_row.iloc[j]
                            if not pd.isna(r_ij):
                                pred = (
                                    float(self.mu or 0)
                                    + self.b_V[j]
                                    + float(np.dot(u, self.V[j]))
                                )
                                err = float(r_ij) - pred
                                u += self.lr * (err * self.V[j] - self.reg * u)

                    scores = (
                        float(self.mu or 0) + self.b_V + np.dot(u, self.V.T).flatten()
                    )
                    scores = np.asarray(scores, dtype=float).flatten()
                    best_idx = int(np.argmin(scores))
                    predictions[str(instance)] = [
                        (str(self.algorithms[best_idx]), float(self.budget or 0))
                    ]
                else:
                    if features is None:
                        # Fallback to global average if nothing else available
                        avg_scores = self.performance_matrix.mean(axis=0)
                        best_idx = int(np.argmin(avg_scores.values))
                        predictions[str(instance)] = [
                            (str(self.algorithms[best_idx]), float(self.budget or 0))
                        ]
                    else:
                        best_algo, _ = self._predict_cold_start(
                            features.loc[instance], str(instance)
                        )
                        predictions[str(instance)] = [
                            (best_algo, float(self.budget or 0))
                        ]
            return predictions

        # Case 3: Cold-start prediction using only features
        if features is not None and performance is None:
            for instance in features.index:
                best_algo, _ = self._predict_cold_start(
                    features.loc[instance], str(instance)
                )
                predictions[str(instance)] = [(best_algo, float(self.budget or 0))]
            return predictions

        return predictions

    @staticmethod
    def _define_hyperparameters(
        model_class: list[type] | None = None, **kwargs: Any
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
                Define hyperparameters for CollaborativeFilteringSelector.

                Parameters
                ----------
                model_class : list[type] or None, default=None
                    List of model classes.
                **kwargs : Any
                    Additional keyword arguments.

                Returns
        -------
                tuple
                    Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        if model_class is None:
            model_class = [RidgeRegressorWrapper]

        model_class_param = ClassChoice(
            name="model_class",
            choices=cast(list[type | bool], model_class),
            default=model_class[0],
        )

        n_components_param = Integer(
            name="n_components",
            bounds=(5, 50),
            default=10,
        )

        n_iter_param = Integer(
            name="n_iter",
            bounds=(50, 500),
            default=100,
        )

        lr_param = Float(
            name="lr",
            bounds=(1e-5, 1e-1),
            log=True,
            default=0.001,
        )

        reg_param = Float(
            name="reg",
            bounds=(1e-4, 1.0),
            log=True,
            default=0.1,
        )

        params = [
            model_class_param,
            n_components_param,
            n_iter_param,
            lr_param,
            reg_param,
        ]

        return params, [], []

__init__(model_class=RidgeRegressorWrapper, n_components=10, n_iter=100, lr=0.001, reg=0.1, random_state=42, **kwargs)

Initialize the CollaborativeFilteringSelector.

Parameters

model_class : type or Callable, default=RidgeRegressorWrapper The regressor wrapper to predict latent factors from features. n_components : int, default=10 Number of latent factors. n_iter : int, default=100 Number of iterations for SGD. lr : float, default=0.001 Learning rate for SGD. reg : float, default=0.1 Regularization strength. random_state : int, default=42 Random seed for initialization. **kwargs : Any Additional keyword arguments for the parent classes.

Source code in asf/selectors/collaborative_filtering_selector.py

def __init__(
    self,
    model_class: type | Callable[..., Any] = RidgeRegressorWrapper,
    n_components: int = 10,
    n_iter: int = 100,
    lr: float = 0.001,
    reg: float = 0.1,
    random_state: int = 42,
    **kwargs: Any,
) -> None:
    """
    Initialize the CollaborativeFilteringSelector.

    Parameters
    ----------
    model_class : type or Callable, default=RidgeRegressorWrapper
        The regressor wrapper to predict latent factors from features.
    n_components : int, default=10
        Number of latent factors.
    n_iter : int, default=100
        Number of iterations for SGD.
    lr : float, default=0.001
        Learning rate for SGD.
    reg : float, default=0.1
        Regularization strength.
    random_state : int, default=42
        Random seed for initialization.
    **kwargs : Any
        Additional keyword arguments for the parent classes.
    """
    super().__init__(model_class=model_class, **kwargs)
    self.n_components = int(n_components)
    self.n_iter = int(n_iter)
    self.lr = float(lr)
    self.reg = float(reg)
    self.random_state = int(random_state)
    self.U: np.ndarray | None = None
    self.V: np.ndarray | None = None
    self.performance_matrix: pd.DataFrame | None = None
    self.model: Any | None = None

    self.mu: float | None = None
    self.b_U: np.ndarray | None = None
    self.b_V: np.ndarray | None = None

CosineSelector

Bases: ConfigurableMixin, AbstractSelector

Algorithm selector based on the AS-LLM architecture.

Uses cosine similarity in a learned latent space to match instances to algorithms. This implementation follows the AS-LLM paper (arXiv:2311.13184) architecture: - Instance features → MLP → instance embedding - Algorithm index → Embedding → LSTM → fused with algorithm_features → algorithm embedding - Cosine similarity + MLP for final compatibility prediction

References

Wu, et al. (2023). "Algorithm Selection with Large Language Model." https://arxiv.org/abs/2311.13184

Attributes

normalize_features : bool If True, standardize instance features. embed_size : int Dimensionality of algorithm embeddings. num_hiddens : int Hidden units in LSTM. num_layers : int Number of LSTM layers. alpha : float Weight for learned LSTM features in fusion. beta : float Weight for algorithm_features in fusion. lr : float Learning rate for training. num_epochs : int Number of training epochs. batch_size : int Batch size for training. device : str Device for PyTorch ('cuda' or 'cpu').

Source code in asf/selectors/cosine_selector.py

class CosineSelector(ConfigurableMixin, AbstractSelector):
    """
    Algorithm selector based on the AS-LLM architecture.

    Uses cosine similarity in a learned latent space to match instances to algorithms.
    This implementation follows the AS-LLM paper (arXiv:2311.13184) architecture:
    - Instance features → MLP → instance embedding
    - Algorithm index → Embedding → LSTM → fused with algorithm_features → algorithm embedding
    - Cosine similarity + MLP for final compatibility prediction

    References
    ----------
    Wu, et al. (2023).
    "Algorithm Selection with Large Language Model."
    https://arxiv.org/abs/2311.13184

    Attributes
    ----------
    normalize_features : bool
        If True, standardize instance features.
    embed_size : int
        Dimensionality of algorithm embeddings.
    num_hiddens : int
        Hidden units in LSTM.
    num_layers : int
        Number of LSTM layers.
    alpha : float
        Weight for learned LSTM features in fusion.
    beta : float
        Weight for algorithm_features in fusion.
    lr : float
        Learning rate for training.
    num_epochs : int
        Number of training epochs.
    batch_size : int
        Batch size for training.
    device : str
        Device for PyTorch ('cuda' or 'cpu').
    """

    PREFIX = "cosine"
    RETURN_TYPE = "single"

    # Type hints for attributes set in __init__
    normalize_features: bool
    embed_size: int
    num_hiddens: int
    num_layers: int
    alpha: float
    beta: float
    lr: float
    num_epochs: int
    batch_size: int
    random_state: int
    _device: torch.device
    _model: nn.Module | None
    _scaler: StandardScaler | None
    _imputer: SimpleImputer | None
    _alg_feats: pd.DataFrame | None
    algorithms: list[str]

    def __init__(
        self,
        normalize_features: bool = True,
        embed_size: int = 50,
        num_hiddens: int = 50,
        num_layers: int = 2,
        alpha: float = 0.9,
        beta: float = 0.1,
        lr: float = 0.001,
        num_epochs: int = 100,
        batch_size: int = 128,
        device: str | None = None,
        random_state: int = 42,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the CosineSelector (AS-LLM architecture).

        Parameters
        ----------
        normalize_features : bool, default=True
            If True, standardize instance features.
        embed_size : int, default=50
            Dimensionality of algorithm embeddings.
        num_hiddens : int, default=50
            Hidden units in LSTM.
        num_layers : int, default=2
            Number of LSTM layers.
        alpha : float, default=0.9
            Weight for learned LSTM features in fusion.
        beta : float, default=0.1
            Weight for algorithm_features in fusion.
        lr : float, default=0.001
            Learning rate for training.
        num_epochs : int, default=100
            Number of training epochs.
        batch_size : int, default=128
            Batch size for training.
        device : str, default=None
            Device for PyTorch ('cuda', 'cpu'). If None, auto-detect.
        random_state : int, default=42
            Random seed for reproducibility.
        **kwargs : Any
            Additional keyword arguments.
        """
        if not TORCH_AVAILABLE:
            raise ImportError(
                "CosineSelector requires PyTorch. Install with: pip install torch"
            )

        super().__init__(**kwargs)
        self.normalize_features = bool(normalize_features)
        self.embed_size = int(embed_size)
        self.num_hiddens = int(num_hiddens)
        self.num_layers = int(num_layers)
        self.alpha = float(alpha)
        self.beta = float(beta)
        self.lr = float(lr)
        self.num_epochs = int(num_epochs)
        self.batch_size = int(batch_size)
        self.random_state = int(random_state)

        if device is None:
            self._device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        else:
            self._device = torch.device(device)

        self._model: nn.Module | None = None
        self._scaler: StandardScaler | None = None
        self._imputer: SimpleImputer | None = None
        self._alg_feats: pd.DataFrame | None = None

    def _norm(self, s: str) -> str:
        """Minimal string normalization for matching algorithm identifiers."""
        s = str(s).lower().strip()
        return re.sub(r"[\W_]+", "", s)

    def _generate_training_data(
        self,
        features: np.ndarray,
        performance: np.ndarray,
    ) -> tuple[np.ndarray, np.ndarray]:
        """
        Generate pairwise training data.

        For each instance, create (instance, algorithm) pairs with label 1 if
        the algorithm is best for that instance, 0 otherwise.
        """
        num_instances = features.shape[0]
        num_algorithms = performance.shape[1]

        training_data = []
        training_labels = []

        for i in range(num_instances):
            best_algo = int(np.argmin(performance[i]))
            for j in range(num_algorithms):
                # One-hot encoding for algorithm
                alg_embed = [0] * num_algorithms
                alg_embed[j] = 1
                # Concatenate: instance features + one-hot + algorithm index
                sample = np.append(np.append(features[i], alg_embed), [j])
                training_data.append(sample)
                training_labels.append(1 if j == best_algo else 0)

        return np.array(training_data), np.array(training_labels)

    def _fit(
        self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs: Any
    ) -> None:
        """
        Fit the AS-LLM selector.

        Parameters
        ----------
        features : pd.DataFrame
            Instance feature matrix (rows = instances).
        performance : pd.DataFrame
            Performance matrix (rows = instances, columns = algorithms).
        **kwargs : Any
            Additional keyword arguments.
        """
        torch.manual_seed(self.random_state)
        np.random.seed(self.random_state)

        # Get algorithm features
        alg_df = getattr(self, "algorithm_features", None)
        if alg_df is None or not isinstance(alg_df, pd.DataFrame):
            raise ValueError(
                "Set selector.algorithm_features (pd.DataFrame indexed by algorithm names) before fit()"
            )

        self.algorithms = [str(a) for a in performance.columns]
        alg_df.index = alg_df.index.astype(str)

        # Map algorithm features to performance columns
        norm_to_orig = {self._norm(n): n for n in alg_df.index}
        mapped = []
        missing = []
        for a in self.algorithms:
            na = self._norm(a)
            orig = norm_to_orig.get(na)
            if orig is None:
                missing.append(a)
            else:
                mapped.append(orig)
        if missing:
            avail = list(alg_df.index)[:10]
            raise ValueError(
                f"Algorithm feature rows do not match performance columns. Missing: {missing}. Available sample: {avail}"
            )

        alg_df = alg_df.loc[mapped].copy()
        alg_df.index = [str(a) for a in self.algorithms]
        alg_df = alg_df.select_dtypes(include=[np.number]).astype(float)
        self._alg_feats = alg_df

        # Preprocess features
        X = features.fillna(0.0).to_numpy(dtype=float)
        self._imputer = SimpleImputer()
        X = self._imputer.fit_transform(X)

        if self.normalize_features:
            self._scaler = StandardScaler()
            X = self._scaler.fit_transform(X)

        # Performance matrix
        Y_perf = performance.loc[features.index, self.algorithms].to_numpy(dtype=float)
        col_mean = np.nanmean(Y_perf, axis=0)
        inds = np.where(np.isnan(Y_perf))
        if inds[0].size:
            Y_perf[inds] = np.take(col_mean, inds[1])

        # Generate training data
        training_data, training_labels = self._generate_training_data(X, Y_perf)

        # Convert to tensors
        X_train = torch.tensor(training_data, dtype=torch.float32)
        y_train = torch.tensor(training_labels, dtype=torch.long)
        alg_features_tensor = torch.tensor(alg_df.values, dtype=torch.float32)

        # Create data loader
        train_set = Data.TensorDataset(X_train, y_train)
        train_loader = Data.DataLoader(
            train_set, batch_size=self.batch_size, shuffle=True
        )

        # Create model
        num_user_features = X.shape[1]
        num_algorithms = len(self.algorithms)

        model = _ASLLMRecommendationModel(  # type: ignore[attr-defined]
            num_algorithms=num_algorithms,
            num_user_features=num_user_features,
            algorithm_features=alg_features_tensor,
            embed_size=self.embed_size,
            num_hiddens=self.num_hiddens,
            num_layers=self.num_layers,
            alpha=self.alpha,
            beta=self.beta,
        )
        self._model = cast(Any, model).to(self._device)

        # Training
        optimizer = torch.optim.Adam(self._model.parameters(), lr=self.lr)
        loss_fn = nn.CrossEntropyLoss()

        self._model.train()
        for epoch in range(self.num_epochs):
            for batch_X, batch_y in train_loader:
                batch_X = batch_X.to(self._device)
                batch_y = batch_y.to(self._device)

                optimizer.zero_grad()
                outputs = self._model(batch_X)
                loss = loss_fn(outputs, batch_y)
                loss.backward()
                optimizer.step()

        self._num_user_features = num_user_features

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict the best algorithm for each query instance.
        """
        if self._model is None:
            raise ValueError("fit() must be called before predict()")

        if features is None:
            raise ValueError("CosineSelector requires features for prediction.")

        # Preprocess features
        X = features.fillna(0.0).to_numpy(dtype=float)
        if self._imputer is not None:
            X = self._imputer.transform(X)
        if self.normalize_features and self._scaler is not None:
            X = self._scaler.transform(X)

        self._model.eval()
        num_algorithms = len(self.algorithms)

        out: dict[str, list[tuple[str, float]]] = {}

        with torch.no_grad():
            for i, inst in enumerate(features.index):
                inst_features = X[i]
                best_score = -float("inf")
                best_algo = self.algorithms[0]

                for j in range(num_algorithms):
                    # Create input: instance features + one-hot + algo index
                    alg_embed = [0] * num_algorithms
                    alg_embed[j] = 1
                    sample = np.append(np.append(inst_features, alg_embed), [j])
                    sample_tensor = (
                        torch.tensor(sample, dtype=torch.float32)
                        .unsqueeze(0)
                        .to(self._device)
                    )

                    output = self._model(sample_tensor)
                    # Get probability of being a match (class 1)
                    probs = torch.softmax(output, dim=1)
                    score = probs[0, 1].item()

                    if score > best_score:
                        best_score = score
                        best_algo = self.algorithms[j]

                out[str(inst)] = [(str(best_algo), float(self.budget or 0))]

        return out

    # save and load are inherited from AbstractSelector/ConfigurableMixin

    @staticmethod
    def _define_hyperparameters(
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for CosineSelector.
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        normalize_features_param = Categorical(
            name="normalize_features",
            items=[True, False],
            default=True,
        )

        embed_size_param = Integer(
            name="embed_size",
            bounds=(10, 100),
            default=50,
        )

        num_hiddens_param = Integer(
            name="num_hiddens",
            bounds=(10, 100),
            default=50,
        )

        num_layers_param = Integer(
            name="num_layers",
            bounds=(1, 4),
            default=2,
        )

        alpha_param = Float(
            name="alpha",
            bounds=(0.0, 1.0),
            default=0.9,
        )

        beta_param = Float(
            name="beta",
            bounds=(0.0, 1.0),
            default=0.1,
        )

        lr_param = Float(
            name="lr",
            bounds=(1e-5, 1e-2),
            log=True,
            default=0.001,
        )

        num_epochs_param = Integer(
            name="num_epochs",
            bounds=(10, 200),
            default=100,
        )

        batch_size_param = Integer(
            name="batch_size",
            bounds=(32, 256),
            default=128,
        )

        params = [
            normalize_features_param,
            embed_size_param,
            num_hiddens_param,
            num_layers_param,
            alpha_param,
            beta_param,
            lr_param,
            num_epochs_param,
            batch_size_param,
        ]

        return params, [], []

__init__(normalize_features=True, embed_size=50, num_hiddens=50, num_layers=2, alpha=0.9, beta=0.1, lr=0.001, num_epochs=100, batch_size=128, device=None, random_state=42, **kwargs)

Initialize the CosineSelector (AS-LLM architecture).

Parameters

normalize_features : bool, default=True If True, standardize instance features. embed_size : int, default=50 Dimensionality of algorithm embeddings. num_hiddens : int, default=50 Hidden units in LSTM. num_layers : int, default=2 Number of LSTM layers. alpha : float, default=0.9 Weight for learned LSTM features in fusion. beta : float, default=0.1 Weight for algorithm_features in fusion. lr : float, default=0.001 Learning rate for training. num_epochs : int, default=100 Number of training epochs. batch_size : int, default=128 Batch size for training. device : str, default=None Device for PyTorch ('cuda', 'cpu'). If None, auto-detect. random_state : int, default=42 Random seed for reproducibility. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/cosine_selector.py

def __init__(
    self,
    normalize_features: bool = True,
    embed_size: int = 50,
    num_hiddens: int = 50,
    num_layers: int = 2,
    alpha: float = 0.9,
    beta: float = 0.1,
    lr: float = 0.001,
    num_epochs: int = 100,
    batch_size: int = 128,
    device: str | None = None,
    random_state: int = 42,
    **kwargs: Any,
) -> None:
    """
    Initialize the CosineSelector (AS-LLM architecture).

    Parameters
    ----------
    normalize_features : bool, default=True
        If True, standardize instance features.
    embed_size : int, default=50
        Dimensionality of algorithm embeddings.
    num_hiddens : int, default=50
        Hidden units in LSTM.
    num_layers : int, default=2
        Number of LSTM layers.
    alpha : float, default=0.9
        Weight for learned LSTM features in fusion.
    beta : float, default=0.1
        Weight for algorithm_features in fusion.
    lr : float, default=0.001
        Learning rate for training.
    num_epochs : int, default=100
        Number of training epochs.
    batch_size : int, default=128
        Batch size for training.
    device : str, default=None
        Device for PyTorch ('cuda', 'cpu'). If None, auto-detect.
    random_state : int, default=42
        Random seed for reproducibility.
    **kwargs : Any
        Additional keyword arguments.
    """
    if not TORCH_AVAILABLE:
        raise ImportError(
            "CosineSelector requires PyTorch. Install with: pip install torch"
        )

    super().__init__(**kwargs)
    self.normalize_features = bool(normalize_features)
    self.embed_size = int(embed_size)
    self.num_hiddens = int(num_hiddens)
    self.num_layers = int(num_layers)
    self.alpha = float(alpha)
    self.beta = float(beta)
    self.lr = float(lr)
    self.num_epochs = int(num_epochs)
    self.batch_size = int(batch_size)
    self.random_state = int(random_state)

    if device is None:
        self._device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
    else:
        self._device = torch.device(device)

    self._model: nn.Module | None = None
    self._scaler: StandardScaler | None = None
    self._imputer: SimpleImputer | None = None
    self._alg_feats: pd.DataFrame | None = None

DummyFeatureGenerator

Bases: AbstractFeatureGenerator

Feature generator that does nothing.

Source code in asf/selectors/feature_generator.py

class DummyFeatureGenerator(AbstractFeatureGenerator):
    """
    Feature generator that does nothing.
    """

    def generate_features(self, base_features: pd.DataFrame) -> pd.DataFrame:
        return pd.DataFrame(index=base_features.index)

DyadRanking

Bases: ConfigurableMixin, AbstractModelBasedSelector

Dyad Ranking for Algorithm Selection.

Implements the approach from: Tornede et al. (2019) "Algorithm Selection as Recommendation: From Collaborative Filtering to Dyad Ranking"

Reference link: https://ris.uni-paderborn.de/download/15011/17060/ci_workshop_tornede.pdf

Uses XGBoost instead of PLNet for ranking, but the overall approach is the same.

How it works: 1. Create cross-product: instance_features x algorithm_features 2. Sample pairwise comparisons per instance (not full rankings) 3. Train ranker to predict which dyads rank higher 4. For new instance: score all dyads, pick best algorithm

Source code in asf/selectors/dyad_ranking.py

class DyadRanking(ConfigurableMixin, AbstractModelBasedSelector):
    """
    Dyad Ranking for Algorithm Selection.

    Implements the approach from:
    Tornede et al. (2019) "Algorithm Selection as Recommendation:
    From Collaborative Filtering to Dyad Ranking"

    Reference link:
    https://ris.uni-paderborn.de/download/15011/17060/ci_workshop_tornede.pdf

    Uses XGBoost instead of PLNet for ranking, but the overall approach is the same.


    How it works:
    1. Create cross-product: instance_features x algorithm_features
    2. Sample pairwise comparisons per instance (not full rankings)
    3. Train ranker to predict which dyads rank higher
    4. For new instance: score all dyads, pick best algorithm
    """

    PREFIX = "dyad_ranking"
    RETURN_TYPE = "single"

    def __init__(
        self,
        model_class: type[AbstractPredictor] = XGBoostRankerWrapper,
        algorithm_features: pd.DataFrame | None = None,
        n_pairs_per_instance: int = 10,
        random_state: int = 42,
        **kwargs: Any,
    ) -> None:
        """
        Initialize DyadRanking selector.

        Parameters
        ----------
        model_class : type[AbstractPredictor], default=XGBoostRankerWrapper
            The ranking model class to use.
        algorithm_features : pd.DataFrame or None, default=None
            Algorithm features representing parameters and structural components.
            If None, falls back to one-hot encoding (not recommended per paper).
            Shape: [n_algorithms, n_algo_features]
            Index: algorithm names
        n_pairs_per_instance : int, default=10
            Number of pairwise comparisons to sample per training instance.
        random_state : int, default=42
            Random seed for pairwise sampling reproducibility.
        **kwargs : Any
            Additional keyword arguments.
        """
        AbstractModelBasedSelector.__init__(self, model_class, **kwargs)
        self.classifier: AbstractPredictor | None = None
        self.n_pairs_per_instance = n_pairs_per_instance
        self.random_state = int(random_state)
        # Store user-provided algorithm features separately to avoid being overwritten
        # by parent class fit() method
        self._init_algorithm_features = algorithm_features

    def _fit(
        self,
        features: pd.DataFrame,
        performance: pd.DataFrame,
        **kwargs: Any,
    ) -> None:
        """
        Fit the dyad ranking model.

        Creates dyads by combining instance features with algorithm features,
        then trains a ranker to predict algorithm rankings per instance.

        Parameters
        ----------
        features : pd.DataFrame
            Instance features. Shape: [n_instances, n_instance_features]
        performance : pd.DataFrame
            Algorithm performance per instance. Shape: [n_instances, n_algorithms]
        """
        if (
            self.algorithm_features is None
            and self._init_algorithm_features is not None
        ):
            self.algorithm_features = self._init_algorithm_features

        if self.algorithm_features is None:
            import warnings

            warnings.warn(
                "Using one-hot algorithm features (fallback). Per Tornede et al. (2019), "
                "algorithm features should represent parameters and structural components. "
                "Consider providing meaningful algorithm_features for better performance.",
                UserWarning,
                stacklevel=2,
            )
            self.algorithm_features = self._create_onehot_algorithm_features()
        else:
            # User-provided algorithm features - ensure they match our algorithm set
            if not all(
                algo in self.algorithm_features.index for algo in self.algorithms
            ):
                raise ValueError(
                    "Provided algorithm_features must include all algorithms in the dataset"
                )
            # Reorder to match self.algorithms and make a copy
            self.algorithm_features = self.algorithm_features.loc[
                list(self.algorithms)
            ].copy()

        dyads_df = self._create_pairwise_comparisons(
            features, performance, self.n_pairs_per_instance
        )

        X, y, qid = self._prepare_training_data(dyads_df)

        self.classifier = self.model_class()
        if self.classifier is None:
            raise RuntimeError("Classifier could not be initialized.")

        self.classifier.fit(X, y, qid=qid)

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict the best algorithm for each instance.

        Parameters
        ----------
        features : pd.DataFrame
            Instance features for prediction.

        Returns
        -------
        dict
            Mapping from instance names to algorithm schedules.
            Format: {instance_name: [(algorithm_name, budget)]}
        """
        if features is None:
            raise ValueError("DyadRanking requires features for prediction.")
        if self.classifier is None:
            raise RuntimeError("Classifier has not been fitted.")
        if self.algorithm_features is None:
            raise RuntimeError("Algorithm features missing.")

        # Create dyads for prediction
        dyads_df = self._create_dyads_for_prediction(features)

        feature_cols = list(self.features) + list(self.algorithm_features.columns)
        X = dyads_df[feature_cols]

        predictions = self.classifier.predict(X)
        dyads_df["predicted_score"] = predictions

        results: dict[str, list[tuple[str, float]]] = {}
        for instance_name in features.index:
            instance_dyads = dyads_df[dyads_df["INSTANCE_ID"] == instance_name]

            best_idx = instance_dyads["predicted_score"].argmin()
            best_algo = instance_dyads.iloc[best_idx]["ALGORITHM"]

            results[str(instance_name)] = [(str(best_algo), float(self.budget or 0))]

        return results

    def _create_onehot_algorithm_features(self) -> pd.DataFrame:
        """Create one-hot encoded algorithm features."""
        encoder = OneHotEncoder(sparse_output=False)
        algo_array = np.array(self.algorithms).reshape(-1, 1)
        encoded = encoder.fit_transform(algo_array)

        return pd.DataFrame(
            encoded,
            index=pd.Index(list(self.algorithms)),
            columns=pd.Index([f"algo_{i}" for i in range(len(self.algorithms))]),
        )

    def _create_dyads_for_prediction(self, features: pd.DataFrame) -> pd.DataFrame:
        """Create dyads for prediction (without performance values)."""
        features_reset = features[list(self.features)].reset_index()
        inst_col_name = features.index.name or "index"
        features_reset = features_reset.rename(columns={inst_col_name: "INSTANCE_ID"})

        assert self.algorithm_features is not None
        algo_features = self.algorithm_features.copy()
        algo_features.index.name = "ALGORITHM"
        algo_features_reset = algo_features.reset_index()

        dyads = pd.merge(features_reset, algo_features_reset, how="cross")
        return dyads

    def _create_pairwise_comparisons(
        self,
        features: pd.DataFrame,
        performance: pd.DataFrame,
        n_pairs: int,
    ) -> pd.DataFrame:
        """
        Sample pairwise comparisons per instance.

        For each instance, randomly samples n_pairs algorithm pairs where
        the two algorithms have different performance.

        Parameters
        ----------
        features : pd.DataFrame
            Instance features
        performance : pd.DataFrame
            Algorithm performance per instance
        n_pairs : int
            Number of pairs to sample per instance

        Returns
        -------
        pd.DataFrame
            Dyads with pairwise rankings (rank 1 = better, rank 2 = worse)
        """
        rng = np.random.RandomState(self.random_state)
        all_pairs = []

        for instance_idx in features.index:
            instance_perf = performance.loc[instance_idx, self.algorithms]
            instance_feats = features.loc[instance_idx, self.features]

            valid_pairs = []
            for i, algo1 in enumerate(self.algorithms):
                for algo2 in self.algorithms[i + 1 :]:
                    perf1 = instance_perf[algo1]
                    perf2 = instance_perf[algo2]

                    if pd.isna(perf1) or pd.isna(perf2) or np.isclose(perf1, perf2):
                        continue

                    if self.maximize:
                        better, worse = (
                            (algo1, algo2) if perf1 > perf2 else (algo2, algo1)
                        )
                    else:
                        better, worse = (
                            (algo1, algo2) if perf1 < perf2 else (algo2, algo1)
                        )

                    valid_pairs.append((better, worse))

            if len(valid_pairs) == 0:
                continue

            n_sample = min(n_pairs, len(valid_pairs))
            sampled_pairs = [
                valid_pairs[i]
                for i in rng.choice(len(valid_pairs), n_sample, replace=False)
            ]

            # Create dyads for sampled pairs
            for better_algo, worse_algo in sampled_pairs:
                # Create two dyads: one ranked 1 (better), one ranked 2 (worse)
                assert self.algorithm_features is not None
                for algo, rank in [(better_algo, 1), (worse_algo, 2)]:
                    dyad_features = dict(instance_feats)
                    dyad_features.update(self.algorithm_features.loc[algo].to_dict())
                    dyad_features["INSTANCE_ID"] = instance_idx
                    dyad_features["ALGORITHM"] = algo
                    dyad_features["rank"] = rank
                    dyad_features["pair_id"] = (
                        f"{instance_idx}_{better_algo}_{worse_algo}"
                    )
                    all_pairs.append(dyad_features)

        if len(all_pairs) == 0:
            raise ValueError(
                "No valid pairwise comparisons could be created. "
                "This typically happens when all algorithms have identical performance "
                "(e.g., all timeout on all instances). Cannot train dyad ranking model."
            )

        return pd.DataFrame(all_pairs)

    def _prepare_training_data(
        self,
        dyads_df: pd.DataFrame,
    ) -> tuple[pd.DataFrame, pd.Series, np.ndarray]:
        """
        Prepare training data for ranking model.

        Returns
        -------
        X : pd.DataFrame
            Feature matrix (instance features + algorithm features)
        y : pd.Series
            Target ranks
        qid : np.ndarray
            Group IDs for ranking (pair IDs for pairwise comparisons)
        """
        # Sort by pair_id to ensure qid is in non-decreasing order (required by XGBoost)
        dyads_df = dyads_df.sort_values("pair_id").reset_index(drop=True)

        assert self.algorithm_features is not None
        feature_cols = list(self.features) + list(self.algorithm_features.columns)
        X = dyads_df[feature_cols]
        y = dyads_df["rank"]

        # Group by pair ID (each pairwise comparison is a separate ranking)
        q_encoder = OrdinalEncoder()
        qid = (
            q_encoder.fit_transform(dyads_df["pair_id"].to_numpy().reshape(-1, 1))
            .flatten()
            .astype(int)
        )

        return X, y, qid

    @staticmethod
    def _define_hyperparameters(
        model_class: list[type[AbstractPredictor]] | None = None,
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for SimpleRanking.

        Parameters
        ----------
        model_class : list[type[AbstractPredictor]] or None, default=None
            List of model classes to choose from.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        if model_class is None:
            model_class = [XGBoostRankerWrapper]

        model_class_param = ClassChoice(
            name="model_class",
            choices=cast(list[type | bool], model_class),
            default=model_class[0],
        )

        return [model_class_param], [], []

__init__(model_class=XGBoostRankerWrapper, algorithm_features=None, n_pairs_per_instance=10, random_state=42, **kwargs)

Initialize DyadRanking selector.

Parameters

model_class : type[AbstractPredictor], default=XGBoostRankerWrapper The ranking model class to use. algorithm_features : pd.DataFrame or None, default=None Algorithm features representing parameters and structural components. If None, falls back to one-hot encoding (not recommended per paper). Shape: [n_algorithms, n_algo_features] Index: algorithm names n_pairs_per_instance : int, default=10 Number of pairwise comparisons to sample per training instance. random_state : int, default=42 Random seed for pairwise sampling reproducibility. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/dyad_ranking.py

def __init__(
    self,
    model_class: type[AbstractPredictor] = XGBoostRankerWrapper,
    algorithm_features: pd.DataFrame | None = None,
    n_pairs_per_instance: int = 10,
    random_state: int = 42,
    **kwargs: Any,
) -> None:
    """
    Initialize DyadRanking selector.

    Parameters
    ----------
    model_class : type[AbstractPredictor], default=XGBoostRankerWrapper
        The ranking model class to use.
    algorithm_features : pd.DataFrame or None, default=None
        Algorithm features representing parameters and structural components.
        If None, falls back to one-hot encoding (not recommended per paper).
        Shape: [n_algorithms, n_algo_features]
        Index: algorithm names
    n_pairs_per_instance : int, default=10
        Number of pairwise comparisons to sample per training instance.
    random_state : int, default=42
        Random seed for pairwise sampling reproducibility.
    **kwargs : Any
        Additional keyword arguments.
    """
    AbstractModelBasedSelector.__init__(self, model_class, **kwargs)
    self.classifier: AbstractPredictor | None = None
    self.n_pairs_per_instance = n_pairs_per_instance
    self.random_state = int(random_state)
    # Store user-provided algorithm features separately to avoid being overwritten
    # by parent class fit() method
    self._init_algorithm_features = algorithm_features

HARRIS

Bases: ConfigurableMixin, AbstractSelector

Hybrid Ranking and Regression Forests for Algorithm Selection.

HARRIS builds an ensemble of decision trees trained with a hybrid loss that combines regression (MSE) and ranking (Spearman correlation) objectives. Each tree makes splits based on both objectives, but returns predictions based on regression labels at leaf nodes.

References

Fehring, et al. (2022). "HARRIS-Hybrid Algorithm Selection using Regression and Ranking." https://arxiv.org/pdf/2210.17341

Source code in asf/selectors/hybrid_decision_tree.py

class HARRIS(ConfigurableMixin, AbstractSelector):
    """
    Hybrid Ranking and Regression Forests for Algorithm Selection.

    HARRIS builds an ensemble of decision trees trained with a hybrid loss
    that combines regression (MSE) and ranking (Spearman correlation) objectives.
    Each tree makes splits based on both objectives, but returns predictions
    based on regression labels at leaf nodes.

    References
    ----------
    Fehring, et al. (2022).
    "HARRIS-Hybrid Algorithm Selection using Regression and Ranking."
    https://arxiv.org/pdf/2210.17341
    """

    PREFIX = "harris"
    RETURN_TYPE = "single"

    def __init__(
        self,
        n_estimators: int = 100,
        max_depth: int = 10,
        min_samples_split: int = 2,
        lambda_param: float = 0.5,
        max_features: Optional[str] = "sqrt",
        max_thresholds: int = 32,
        random_state: int = 42,
        **kwargs: Any,
    ):
        """
        Initialize HARRIS selector.

        Args:
            n_estimators: Number of trees in the forest.
            max_depth: Maximum depth of each tree.
            min_samples_split: Minimum samples required to split a node.
            lambda_param: Balance between regression (λ=1) and ranking (λ=0).
                         λ ∈ [0, 1], controls hybrid loss.
            max_features: Number of features to consider for splits:
                         - "sqrt": sqrt(n_features)
                         - "log2": log2(n_features)
                         - int: exact number
                         - None: all features
            max_thresholds: Maximum candidate thresholds per feature (quantile sampled)
            random_state: Random seed.
            **kwargs: Additional arguments for parent class.
        """
        super().__init__(**kwargs)
        self.n_estimators = int(n_estimators)
        self.max_depth = int(max_depth)
        self.min_samples_split = int(min_samples_split)
        self.lambda_param = float(lambda_param)
        self.max_features = max_features
        self.max_thresholds = int(max_thresholds)
        self.random_state = int(random_state)

        self.trees: List[HybridDecisionTree] = []
        self.algorithms: List[str] = []
        self.feature_indices_per_tree: List[np.ndarray] = []

    @staticmethod
    def _define_hyperparameters(
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for HARRIS.

        The defaults mirror the constructor defaults, while the ranges keep the
        search space compact enough for selector-level HPO.
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        hyperparameters = [
            Integer("n_estimators", bounds=(10, 200), default=100, log=True),
            Integer("max_depth", bounds=(2, 20), default=10),
            Integer("min_samples_split", bounds=(2, 20), default=2),
            Float("lambda_param", bounds=(0.0, 1.0), default=0.5),
            Categorical(
                "max_features",
                items=["sqrt", "log2", "all"],
                default="sqrt",
            ),
            Integer("max_thresholds", bounds=(4, 128), default=32, log=True),
        ]
        return hyperparameters, [], []

    def _scale_performance(self, performance: pd.DataFrame) -> np.ndarray:
        """Scale performance to [0, 1] using min-max normalization."""
        y = performance.values

        # Min-max scaling per algorithm
        y_min = np.nanmin(y, axis=0, keepdims=True)
        y_max = np.nanmax(y, axis=0, keepdims=True)

        # Avoid division by zero
        y_range = y_max - y_min
        y_range[y_range == 0] = 1.0

        return (y - y_min) / y_range

    def _get_n_features_to_sample(self, n_total_features: int) -> int:
        """Determine number of features to sample for each tree."""
        if self.max_features == "sqrt":
            return max(1, int(np.sqrt(n_total_features)))
        elif self.max_features == "log2":
            return max(1, int(np.log2(n_total_features)))
        elif isinstance(self.max_features, int):
            return min(self.max_features, n_total_features)
        else:
            return n_total_features

    def _fit(self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs) -> None:
        """
        Train the HARRIS forest.

        Args:
            features: DataFrame of instance features.
            performance: DataFrame of algorithm performance (runtimes).
        """
        self.algorithms = list(performance.columns)
        self.trees = []
        self.feature_indices_per_tree = []

        X = features.values
        y = self._scale_performance(performance)

        n_instances, n_features = X.shape
        n_features_to_sample = self._get_n_features_to_sample(n_features)

        rng = np.random.RandomState(self.random_state)

        for i in range(self.n_estimators):
            # Bootstrap sample
            bootstrap_indices = rng.choice(n_instances, size=n_instances, replace=True)
            X_boot = X[bootstrap_indices]
            y_boot = y[bootstrap_indices]

            # Random feature subset
            feature_indices = rng.choice(
                n_features, size=n_features_to_sample, replace=False
            )
            X_boot_subset = X_boot[:, feature_indices]

            # Train tree
            tree = HybridDecisionTree(
                max_depth=self.max_depth,
                min_samples_split=self.min_samples_split,
                lambda_param=self.lambda_param,
                max_thresholds=self.max_thresholds,
                random_state=rng.randint(0, 1000000),
            )
            tree.fit(X_boot_subset, y_boot)

            self.trees.append(tree)
            self.feature_indices_per_tree.append(feature_indices)

    def _predict(
        self, features: pd.DataFrame | None, performance: pd.DataFrame | None = None
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict the best algorithm for each instance.

        Args:
            features (pd.DataFrame): Feature matrix with instances as rows and features as columns.

        Returns:
            Dict[str, List[Tuple[str, float]]]: A dictionary mapping each instance name to a list containing
                a tuple of the best algorithm (str) and the budget (float).
        """
        budget = getattr(self, "budget", None)
        if budget is None:
            budget = float("inf")

        assert isinstance(features, pd.DataFrame)
        X = features.values
        predictions = {}

        for inst_idx, inst_name in enumerate(features.index):
            x = X[inst_idx : inst_idx + 1, :]

            # Aggregate predictions from all trees
            tree_predictions = []
            for tree, feature_indices in zip(self.trees, self.feature_indices_per_tree):
                x_subset = x[:, feature_indices]
                pred = tree.predict(x_subset)[0, :]
                tree_predictions.append(pred)

            # Average predictions across trees
            avg_prediction = np.mean(tree_predictions, axis=0)

            # Select best algorithm based on maximize parameter
            if self.maximize:
                best_algo_idx = np.argmax(avg_prediction)
            else:
                best_algo_idx = np.argmin(avg_prediction)
            best_algo = self.algorithms[best_algo_idx]

            predictions[str(inst_name)] = [(str(best_algo), float(budget))]

        return predictions

__init__(n_estimators=100, max_depth=10, min_samples_split=2, lambda_param=0.5, max_features='sqrt', max_thresholds=32, random_state=42, **kwargs)

Initialize HARRIS selector.

Parameters:

Name	Type	Description	Default
n_estimators	int	Number of trees in the forest.	100
max_depth	int	Maximum depth of each tree.	10
min_samples_split	int	Minimum samples required to split a node.	2
lambda_param	float	Balance between regression (λ=1) and ranking (λ=0). λ ∈ [0, 1], controls hybrid loss.	0.5
max_features	Optional[str]	Number of features to consider for splits: - "sqrt": sqrt(n_features) - "log2": log2(n_features) - int: exact number - None: all features	'sqrt'
max_thresholds	int	Maximum candidate thresholds per feature (quantile sampled)	32
random_state	int	Random seed.	42
**kwargs	Any	Additional arguments for parent class.	{}

Source code in asf/selectors/hybrid_decision_tree.py

def __init__(
    self,
    n_estimators: int = 100,
    max_depth: int = 10,
    min_samples_split: int = 2,
    lambda_param: float = 0.5,
    max_features: Optional[str] = "sqrt",
    max_thresholds: int = 32,
    random_state: int = 42,
    **kwargs: Any,
):
    """
    Initialize HARRIS selector.

    Args:
        n_estimators: Number of trees in the forest.
        max_depth: Maximum depth of each tree.
        min_samples_split: Minimum samples required to split a node.
        lambda_param: Balance between regression (λ=1) and ranking (λ=0).
                     λ ∈ [0, 1], controls hybrid loss.
        max_features: Number of features to consider for splits:
                     - "sqrt": sqrt(n_features)
                     - "log2": log2(n_features)
                     - int: exact number
                     - None: all features
        max_thresholds: Maximum candidate thresholds per feature (quantile sampled)
        random_state: Random seed.
        **kwargs: Additional arguments for parent class.
    """
    super().__init__(**kwargs)
    self.n_estimators = int(n_estimators)
    self.max_depth = int(max_depth)
    self.min_samples_split = int(min_samples_split)
    self.lambda_param = float(lambda_param)
    self.max_features = max_features
    self.max_thresholds = int(max_thresholds)
    self.random_state = int(random_state)

    self.trees: List[HybridDecisionTree] = []
    self.algorithms: List[str] = []
    self.feature_indices_per_tree: List[np.ndarray] = []

ISA

Bases: ConfigurableMixin, AbstractSelector

ISA (Instance-Specific Aspeed) selector.

References

Lindauer, M., et al. (2016). "Instance-Specific Algorithm Selection Schedules." https://ml.informatik.uni-freiburg.de/wp-content/uploads/papers/16-LION-ASschedules.pdf

Attributes

k : int Number of neighbors for k-NN. use_k_tuning : bool Whether to tune k using cross-validation. n_folds : int Number of folds for cross-validation when tuning k. k_candidates : list[int] Candidate k values to consider when tuning. aspeed_cutoff : int Time limit for the internal aspeed solver. cores : int Number of cores for the internal aspeed solver. random_state : int Random seed for reproducibility. reduced_features : pd.DataFrame or None Training features after set reduction. reduced_performance : pd.DataFrame or None Training performance after set reduction. knn : NearestNeighbors or None k-NN model.

Source code in asf/selectors/isa.py

class ISA(ConfigurableMixin, AbstractSelector):
    """
    ISA (Instance-Specific Aspeed) selector.

    References
    ----------
    Lindauer, M., et al. (2016).
    "Instance-Specific Algorithm Selection Schedules."
    https://ml.informatik.uni-freiburg.de/wp-content/uploads/papers/16-LION-ASschedules.pdf

    Attributes
    ----------
    k : int
        Number of neighbors for k-NN.
    use_k_tuning : bool
        Whether to tune k using cross-validation.
    n_folds : int
        Number of folds for cross-validation when tuning k.
    k_candidates : list[int]
        Candidate k values to consider when tuning.
    aspeed_cutoff : int
        Time limit for the internal aspeed solver.
    cores : int
        Number of cores for the internal aspeed solver.
    random_state : int
        Random seed for reproducibility.
    reduced_features : pd.DataFrame or None
        Training features after set reduction.
    reduced_performance : pd.DataFrame or None
        Training performance after set reduction.
    knn : NearestNeighbors or None
        k-NN model.
    """

    PREFIX = "isa"
    RETURN_TYPE = "schedule"

    def __init__(
        self,
        k: int = 10,
        use_k_tuning: bool = False,
        n_folds: int = 5,
        k_candidates: list[int] | None = None,
        aspeed_cutoff: int = 5,
        cores: int = 1,
        random_state: int = 42,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the ISA selector.

        Parameters
        ----------
        k : int, default=10
            Number of neighbors for k-NN.
        use_k_tuning : bool, default=False
            Whether to tune k using cross-validation.
        n_folds : int, default=5
            Number of folds for cross-validation when tuning k.
        k_candidates : list[int] or None, default=None
            Candidate k values to consider when tuning.
        aspeed_cutoff : int, default=30
            Time limit for the internal aspeed solver.
        cores : int, default=1
            Number of cores for the internal aspeed solver.
        random_state : int, default=42
            Random seed for reproducibility.
        **kwargs : Any
            Additional keyword arguments for the parent class.
        """
        if not CLINGO_AVAIL:
            raise ImportError("clingo is not installed. Please install it to use ISA.")
        super().__init__(**kwargs)
        self.k = int(k)
        self.use_k_tuning = bool(use_k_tuning)
        self.n_folds = int(n_folds)

        if isinstance(k_candidates, str):
            k_map = {
                "small": [3, 5, 10],
                "medium": [3, 5, 10, 15, 20],
                "broad": [3, 5, 10, 15, 20, 30, 50],
            }
            self.k_candidates = k_map.get(k_candidates, [3, 5, 10, 15, 20])
        else:
            self.k_candidates = k_candidates or [3, 5, 10, 15, 20]

        self.aspeed_cutoff = int(aspeed_cutoff)
        self.cores = int(cores)
        self.random_state = int(random_state)

        self.reduced_features: pd.DataFrame | None = None
        self.reduced_performance: pd.DataFrame | None = None
        self.knn: NearestNeighbors | None = None

    def _fit(
        self,
        features: pd.DataFrame,
        performance: pd.DataFrame,
        **kwargs: Any,
    ) -> None:
        """
        Fit the ISA selector.

        Parameters
        ----------
        features : pd.DataFrame
            Training features (instances x features).
        performance : pd.DataFrame
            Training performance (instances x algorithms).
        """
        is_solved = performance < self.budget
        solved_by_all = is_solved.all(axis=1)
        solved_by_none = ~is_solved.any(axis=1)
        trivial_mask = solved_by_all | solved_by_none

        self.reduced_features = features[~trivial_mask].copy()
        self.reduced_performance = performance[~trivial_mask].copy()

        if self.reduced_features.empty:
            return

        if self.use_k_tuning:
            self.k = self._tune_k()

        self.knn = NearestNeighbors(
            n_neighbors=min(self.k, len(self.reduced_features)), metric="euclidean"
        )
        self.knn.fit(self.reduced_features.values)

    def _tune_k(self) -> int:
        """
        Tune the neighborhood size k via cross-validation.

        Returns
        -------
        int
            The best k value found.
        """
        if self.reduced_features is None:
            return self.k
        best_k = self.k
        best_score = float("inf")
        kf = KFold(n_splits=self.n_folds, shuffle=True, random_state=self.random_state)
        instance_indices = np.arange(len(self.reduced_features))

        for candidate_k in self.k_candidates:
            fold_scores = []
            for train_idx, val_idx in kf.split(instance_indices):
                assert self.reduced_features is not None
                assert self.reduced_performance is not None
                train_features = self.reduced_features.iloc[train_idx]
                train_perf = self.reduced_performance.iloc[train_idx]
                val_features = self.reduced_features.iloc[val_idx]
                val_perf = self.reduced_performance.iloc[val_idx]

                if len(train_features) < candidate_k:
                    continue

                knn = NearestNeighbors(n_neighbors=candidate_k, metric="euclidean")
                knn.fit(train_features.values)

                total_runtime = 0.0
                for _, instance_row in val_features.iterrows():
                    x = instance_row.values.reshape(1, -1)
                    _, neighbor_idxs = knn.kneighbors(x)
                    neighbor_perf = train_perf.iloc[neighbor_idxs.flatten()]

                    schedule = self._get_aspeed_schedule(neighbor_perf)

                    instance_actual_perf = val_perf.loc[instance_row.name]
                    solved = False
                    for algo, _ in schedule:
                        runtime = float(instance_actual_perf.get(algo, self.budget))
                        if self.budget is not None and runtime < self.budget:
                            total_runtime += runtime
                            solved = True
                            break
                    if not solved:
                        total_runtime += float(self.budget or 0)

                avg_runtime = total_runtime / len(val_features)
                fold_scores.append(avg_runtime)

            mean_score = np.mean(fold_scores) if fold_scores else float("inf")
            if mean_score < best_score:
                best_score = float(mean_score)
                best_k = candidate_k

        return best_k

    def _get_aspeed_schedule(
        self, performance_subset: pd.DataFrame
    ) -> list[tuple[str, float]]:
        """
        Run aspeed on performance data to get a schedule.

        Parameters
        ----------
        performance_subset : pd.DataFrame
            Performance matrix for the neighborhood.

        Returns
        -------
        list[tuple[str, float]]
            List of (algorithm, time) tuples.
        """
        aspeed_presolver = Aspeed(
            budget=float(self.budget or 0),
            aspeed_cutoff=self.aspeed_cutoff,
            cores=self.cores,
        )

        aspeed_presolver.fit(features=None, performance=performance_subset)
        schedule = cast(list[tuple[str, float]], aspeed_presolver.predict())
        schedule.sort(key=lambda x: x[1])

        total_time = sum(time for _, time in schedule)
        remaining_time = float(self.budget or 0) - total_time

        if remaining_time > 0:
            if schedule:
                max_idx = max(range(len(schedule)), key=lambda i: schedule[i][1])
                algo, time = schedule[max_idx]
                schedule[max_idx] = (str(algo), float(time + remaining_time))
            else:
                # Fallback if aspeed returns empty schedule (unlikely)
                pass

        return [(str(a), float(t)) for a, t in schedule]

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict algorithm schedules for each instance.

        Parameters
        ----------
        features : pd.DataFrame or None
            The input features.
        performance : pd.DataFrame or None, default=None
            Partial performance data.

        Returns
        -------
        dict
            Mapping from instance name to algorithm schedules.
        """
        if features is None:
            raise ValueError("ISA requires features for prediction.")
        if self.knn is None:
            return {str(instance): [] for instance in features.index}

        predictions: dict[str, list[tuple[str, float]]] = {}
        for instance_name in features.index:
            x = features.loc[[instance_name]].values
            _, neighbor_idxs = self.knn.kneighbors(x)
            assert self.reduced_performance is not None
            neighbor_perf = self.reduced_performance.iloc[neighbor_idxs.flatten()]

            schedule = self._get_aspeed_schedule(neighbor_perf)
            predictions[str(instance_name)] = schedule

        return predictions

    @staticmethod
    def _define_hyperparameters(
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for ISA.

        Parameters
        ----------
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        k_param = Integer(
            name="k",
            bounds=(1, 50),
            default=10,
        )

        params = [
            k_param,
        ]

        return params, [], []

__init__(k=10, use_k_tuning=False, n_folds=5, k_candidates=None, aspeed_cutoff=5, cores=1, random_state=42, **kwargs)

Initialize the ISA selector.

Parameters

k : int, default=10 Number of neighbors for k-NN. use_k_tuning : bool, default=False Whether to tune k using cross-validation. n_folds : int, default=5 Number of folds for cross-validation when tuning k. k_candidates : list[int] or None, default=None Candidate k values to consider when tuning. aspeed_cutoff : int, default=30 Time limit for the internal aspeed solver. cores : int, default=1 Number of cores for the internal aspeed solver. random_state : int, default=42 Random seed for reproducibility. **kwargs : Any Additional keyword arguments for the parent class.

Source code in asf/selectors/isa.py

def __init__(
    self,
    k: int = 10,
    use_k_tuning: bool = False,
    n_folds: int = 5,
    k_candidates: list[int] | None = None,
    aspeed_cutoff: int = 5,
    cores: int = 1,
    random_state: int = 42,
    **kwargs: Any,
) -> None:
    """
    Initialize the ISA selector.

    Parameters
    ----------
    k : int, default=10
        Number of neighbors for k-NN.
    use_k_tuning : bool, default=False
        Whether to tune k using cross-validation.
    n_folds : int, default=5
        Number of folds for cross-validation when tuning k.
    k_candidates : list[int] or None, default=None
        Candidate k values to consider when tuning.
    aspeed_cutoff : int, default=30
        Time limit for the internal aspeed solver.
    cores : int, default=1
        Number of cores for the internal aspeed solver.
    random_state : int, default=42
        Random seed for reproducibility.
    **kwargs : Any
        Additional keyword arguments for the parent class.
    """
    if not CLINGO_AVAIL:
        raise ImportError("clingo is not installed. Please install it to use ISA.")
    super().__init__(**kwargs)
    self.k = int(k)
    self.use_k_tuning = bool(use_k_tuning)
    self.n_folds = int(n_folds)

    if isinstance(k_candidates, str):
        k_map = {
            "small": [3, 5, 10],
            "medium": [3, 5, 10, 15, 20],
            "broad": [3, 5, 10, 15, 20, 30, 50],
        }
        self.k_candidates = k_map.get(k_candidates, [3, 5, 10, 15, 20])
    else:
        self.k_candidates = k_candidates or [3, 5, 10, 15, 20]

    self.aspeed_cutoff = int(aspeed_cutoff)
    self.cores = int(cores)
    self.random_state = int(random_state)

    self.reduced_features: pd.DataFrame | None = None
    self.reduced_performance: pd.DataFrame | None = None
    self.knn: NearestNeighbors | None = None

ISAC

Bases: ConfigurableMixin, AbstractSelector

ISAC (Instance-Specific Algorithm Configuration) selector.

Clusters instances in feature space and assigns to each cluster the best algorithm (by mean performance).

References

Kadioglu, S., et al. (2010). "ISAC - Instance-Specific Algorithm Configuration." https://www.researchgate.net/profile/Yuri-Malitsky/publication/220837402_ISAC_-_Instance-Specific_Algorithm_Configuration/links/02e7e52738cb135ccc000000/ISAC-Instance-Specific-Algorithm-Configuration.pdf

Attributes

clusterer : type or Callable or Any The clusterer class, partial, or instance. clusterer_kwargs : dict[str, Any] Arguments for clusterer instantiation. clusterer_instance : Any or None The trained clusterer instance. cluster_to_best_algo : dict[int, str] Mapping from cluster ID to best algorithm name.

Source code in asf/selectors/isac.py

class ISAC(ConfigurableMixin, AbstractSelector):
    """
    ISAC (Instance-Specific Algorithm Configuration) selector.

    Clusters instances in feature space and assigns to each cluster the best
    algorithm (by mean performance).

    References
    ----------
    Kadioglu, S., et al. (2010).
    "ISAC - Instance-Specific Algorithm Configuration."
    https://www.researchgate.net/profile/Yuri-Malitsky/publication/220837402_ISAC_-_Instance-Specific_Algorithm_Configuration/links/02e7e52738cb135ccc000000/ISAC-Instance-Specific-Algorithm-Configuration.pdf

    Attributes
    ----------
    clusterer : type or Callable or Any
        The clusterer class, partial, or instance.
    clusterer_kwargs : dict[str, Any]
        Arguments for clusterer instantiation.
    clusterer_instance : Any or None
        The trained clusterer instance.
    cluster_to_best_algo : dict[int, str]
        Mapping from cluster ID to best algorithm name.
    """

    PREFIX = "isac"
    RETURN_TYPE = "single"

    def __init__(
        self,
        clusterer: type | Callable[..., Any] | Any = GMeansWrapper,
        clusterer_kwargs: dict[str, Any] | None = None,
        random_state: int | None = None,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the ISAC selector.

        Parameters
        ----------
        clusterer : type or Callable or Any, default=GMeansWrapper
            The clusterer class, partial, or instance.
        clusterer_kwargs : dict[str, Any] or None, default=None
            Arguments for clusterer instantiation.
        random_state : int or None, default=None
            Random state for the clusterer.
        **kwargs : Any
            Additional keyword arguments.
        """
        super().__init__(**kwargs)
        self.clusterer = clusterer
        self.clusterer_kwargs = clusterer_kwargs or {}
        self.random_state = random_state
        self.clusterer_instance: Any | None = None
        self.cluster_to_best_algo: dict[int, str] = {}
        self.fallback_algo: str | None = None

    def _fit(
        self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs: Any
    ) -> None:
        """
        Fit the ISAC selector.

        Parameters
        ----------
        features : pd.DataFrame
            Feature matrix (instances x features).
        performance : pd.DataFrame
            Performance matrix (instances x algorithms).
        """
        if isinstance(self.clusterer, type) or isinstance(self.clusterer, partial):
            self.clusterer_instance = self.clusterer(
                random_state=self.random_state, **self.clusterer_kwargs
            )
        elif hasattr(self.clusterer, "fit") and hasattr(self.clusterer, "predict"):
            self.clusterer_instance = self.clusterer
        else:
            raise ValueError(
                "clusterer must be a class, partial, or an instance with fit/predict"
            )

        clusterer = cast(_ClustererProtocol, self.clusterer_instance)
        clusterer.fit(features.values)
        cluster_labels = clusterer.predict(features.values)

        # Compute global fallback algorithm
        perf_by_algo = performance[self.algorithms]
        global_perf_means = perf_by_algo.mean(axis=0)
        if global_perf_means.isna().all():
            self.fallback_algo = str(self.algorithms[0])
        else:
            if self.maximize:
                self.fallback_algo = str(global_perf_means.idxmax())
            else:
                self.fallback_algo = str(global_perf_means.idxmin())

        n_clusters = len(np.unique(cluster_labels))
        for cluster_id in range(n_clusters):
            idxs = np.where(cluster_labels == cluster_id)[0]
            if len(idxs) == 0:
                continue
            cluster_perf = performance.iloc[idxs][self.algorithms]
            algo_means = cluster_perf.mean(axis=0)
            if algo_means.isna().all():
                best_algo = self.fallback_algo
            elif self.maximize:
                best_algo = str(algo_means.idxmax())
            else:
                best_algo = str(algo_means.idxmin())
            self.cluster_to_best_algo[int(cluster_id)] = str(best_algo)

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict the best algorithm for each instance.

        Parameters
        ----------
        features : pd.DataFrame
            Feature matrix for test instances.

        Returns
        -------
        dict
            Mapping from instance name to algorithm schedules.
        """
        if features is None:
            raise ValueError("ISAC require features for prediction.")
        if self.clusterer_instance is None:
            raise RuntimeError("ISAC has not been fitted.")
        clusterer = cast(_ClustererProtocol, self.clusterer_instance)
        cluster_labels = clusterer.predict(features.values)
        predictions: dict[str, list[tuple[str, float]]] = {}
        for idx, instance in enumerate(features.index):
            cluster_id = int(cluster_labels[idx])
            best_algo = self.cluster_to_best_algo.get(cluster_id, self.fallback_algo)
            if best_algo:
                predictions[str(instance)] = [(str(best_algo), float(self.budget or 0))]
            else:
                predictions[str(instance)] = []
        return predictions

    @staticmethod
    def _define_hyperparameters(
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for ISAC.

        Parameters
        ----------
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        clusterer_param = ClassChoice(
            name="clusterer",
            choices=[
                GMeansWrapper,
                KMeansWrapper,
            ],
            default=GMeansWrapper,
        )

        return [clusterer_param], [], []

__init__(clusterer=GMeansWrapper, clusterer_kwargs=None, random_state=None, **kwargs)

Initialize the ISAC selector.

Parameters

clusterer : type or Callable or Any, default=GMeansWrapper The clusterer class, partial, or instance. clusterer_kwargs : dict[str, Any] or None, default=None Arguments for clusterer instantiation. random_state : int or None, default=None Random state for the clusterer. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/isac.py

def __init__(
    self,
    clusterer: type | Callable[..., Any] | Any = GMeansWrapper,
    clusterer_kwargs: dict[str, Any] | None = None,
    random_state: int | None = None,
    **kwargs: Any,
) -> None:
    """
    Initialize the ISAC selector.

    Parameters
    ----------
    clusterer : type or Callable or Any, default=GMeansWrapper
        The clusterer class, partial, or instance.
    clusterer_kwargs : dict[str, Any] or None, default=None
        Arguments for clusterer instantiation.
    random_state : int or None, default=None
        Random state for the clusterer.
    **kwargs : Any
        Additional keyword arguments.
    """
    super().__init__(**kwargs)
    self.clusterer = clusterer
    self.clusterer_kwargs = clusterer_kwargs or {}
    self.random_state = random_state
    self.clusterer_instance: Any | None = None
    self.cluster_to_best_algo: dict[int, str] = {}
    self.fallback_algo: str | None = None

JointRanking

Bases: ConfigurableMixin, AbstractSelector, AbstractFeatureGenerator

Ranking-based algorithm selector.

Combines feature generation and model-based selection to predict algorithm performance.

References

Ozturk, E., et al. (2022). "Joint Ranking for Learning Algorithm Selection." https://arxiv.org/abs/2206.08476

Attributes

model : RankingMLP or Callable or None The model used for ranking.

Source code in asf/selectors/joint_ranking.py

class JointRanking(ConfigurableMixin, AbstractSelector, AbstractFeatureGenerator):
    """
    Ranking-based algorithm selector.

    Combines feature generation and model-based selection to predict algorithm
    performance.

    References
    ----------
    Ozturk, E., et al. (2022).
    "Joint Ranking for Learning Algorithm Selection."
    https://arxiv.org/abs/2206.08476

    Attributes
    ----------
    model : RankingMLP or Callable or None
        The model used for ranking.
    """

    PREFIX = "joint_ranking"
    RETURN_TYPE = "single"

    def save(self, path: str) -> None:
        with open(path, "wb") as f:
            pickle.dump(self, f)

    @classmethod
    def load(cls, path: str) -> "AbstractSelector":
        with open(path, "rb") as f:
            return pickle.load(f)

    def __init__(
        self,
        model: RankingMLP | Callable[..., RankingMLP] | None = None,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the JointRanking selector.

        Parameters
        ----------
        model : RankingMLP or Callable or None, default=None
            The model to be used for ranking algorithms.
        **kwargs : Any
            Additional keyword arguments.
        """
        AbstractSelector.__init__(self, **kwargs)
        AbstractFeatureGenerator.__init__(self)
        self.model = model

    def _fit(
        self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs: Any
    ) -> None:
        """
        Fit the ranking model.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame
            The performance data.
        """
        if self.algorithm_features is None:
            encoder = OneHotEncoder(sparse_output=False)
            self.algorithm_features = pd.DataFrame(
                encoder.fit_transform(np.array(self.algorithms).reshape(-1, 1)),
                index=pd.Index(list(self.algorithms)),
                columns=pd.Index([f"algo_{i}" for i in range(len(self.algorithms))]),
            )

        if self.model is None:
            self.model = RankingMLP(
                input_size=len(self.features) + len(self.algorithm_features.columns)
            )
        elif callable(self.model) and not isinstance(self.model, RankingMLP):
            self.model = self.model(
                input_size=len(self.features) + len(self.algorithm_features.columns)
            )

        if self.model is None:
            raise RuntimeError("Model could not be initialized.")

        self.model.fit(
            X=features[self.features],
            Y=performance,
            algorithm_features=self.algorithm_features,
        )

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict the best algorithm for each instance.

        Parameters
        ----------
        features : pd.DataFrame
            The query instance features.

        Returns
        -------
        dict
            Mapping from instance name to algorithm schedules.
        """
        if features is None:
            raise ValueError("JointRanking require features for prediction.")
        predictions = self.generate_features(features)

        results: dict[str, list[tuple[str, float]]] = {}
        for i, instance_name in enumerate(features.index):
            idx = (
                int(np.argmax(predictions.iloc[i]))
                if self.maximize
                else int(np.argmin(predictions.iloc[i]))
            )
            results[str(instance_name)] = [
                (str(self.algorithms[idx]), float(self.budget or 0))
            ]
        return results

    def generate_features(self, base_features: pd.DataFrame) -> pd.DataFrame:
        """
        Generate predictions for each algorithm.

        Parameters
        ----------
        features : pd.DataFrame
            Input feature matrix.

        Returns
        -------
        pd.DataFrame
            DataFrame containing the predictions for each algorithm.
        """
        if self.model is None:
            raise RuntimeError("Model has not been fitted.")

        predictions = np.zeros((base_features.shape[0], len(self.algorithms)))
        selected_features = base_features[self.features]

        for i, algorithm in enumerate(self.algorithms):
            if self.algorithm_features is None:
                raise RuntimeError("Algorithm features are missing.")

            data = selected_features.assign(**self.algorithm_features.loc[algorithm])
            # Ensure column order matches training
            data = data[self.algorithm_features.columns.to_list() + self.features]
            assert self.model is not None and not isinstance(self.model, type)
            model = cast(Any, self.model)
            prediction = model.predict(data)
            predictions[:, i] = prediction.flatten()

        return pd.DataFrame(predictions, columns=pd.Index(list(self.algorithms)))

    @staticmethod
    def _define_hyperparameters(
        model: list[type] | None = None, **kwargs: Any
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
                Define hyperparameters for JointRanking.

                Parameters
                ----------
                model : list[type] or None, default=None
                    List of model classes to choose from.
                **kwargs : Any
                    Additional keyword arguments.

                Returns
        -------
                tuple
                    Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        if model is None:
            model = [RankingMLP]

        model_param = ClassChoice(
            name="model",
            choices=cast(list[type | bool], model),
            default=model[0],
        )

        return [model_param], [], []

__init__(model=None, **kwargs)

Initialize the JointRanking selector.

Parameters

model : RankingMLP or Callable or None, default=None The model to be used for ranking algorithms. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/joint_ranking.py

def __init__(
    self,
    model: RankingMLP | Callable[..., RankingMLP] | None = None,
    **kwargs: Any,
) -> None:
    """
    Initialize the JointRanking selector.

    Parameters
    ----------
    model : RankingMLP or Callable or None, default=None
        The model to be used for ranking algorithms.
    **kwargs : Any
        Additional keyword arguments.
    """
    AbstractSelector.__init__(self, **kwargs)
    AbstractFeatureGenerator.__init__(self)
    self.model = model

generate_features(base_features)

Generate predictions for each algorithm.

Parameters

features : pd.DataFrame Input feature matrix.

Returns

pd.DataFrame DataFrame containing the predictions for each algorithm.

Source code in asf/selectors/joint_ranking.py

def generate_features(self, base_features: pd.DataFrame) -> pd.DataFrame:
    """
    Generate predictions for each algorithm.

    Parameters
    ----------
    features : pd.DataFrame
        Input feature matrix.

    Returns
    -------
    pd.DataFrame
        DataFrame containing the predictions for each algorithm.
    """
    if self.model is None:
        raise RuntimeError("Model has not been fitted.")

    predictions = np.zeros((base_features.shape[0], len(self.algorithms)))
    selected_features = base_features[self.features]

    for i, algorithm in enumerate(self.algorithms):
        if self.algorithm_features is None:
            raise RuntimeError("Algorithm features are missing.")

        data = selected_features.assign(**self.algorithm_features.loc[algorithm])
        # Ensure column order matches training
        data = data[self.algorithm_features.columns.to_list() + self.features]
        assert self.model is not None and not isinstance(self.model, type)
        model = cast(Any, self.model)
        prediction = model.predict(data)
        predictions[:, i] = prediction.flatten()

    return pd.DataFrame(predictions, columns=pd.Index(list(self.algorithms)))

MetaSelector

Bases: ConfigurableMixin, AbstractSelector

Meta-selector that ensembles multiple base selectors.

Trains multiple base selectors and uses another selector (the meta-selector) to choose among them for each instance based on their out-of-fold performance.

References

Feurer, M., et al. (2021). "Meta-Algorithm Selection." https://arxiv.org/abs/2107.09414

Important: All base selectors and the meta selector must have RETURN_TYPE 'single'

Source code in asf/selectors/meta_selector.py

class MetaSelector(ConfigurableMixin, AbstractSelector):
    """
    Meta-selector that ensembles multiple base selectors.

    Trains multiple base selectors and uses another selector (the meta-selector)
    to choose among them for each instance based on their out-of-fold performance.

    References
    ----------
    Feurer, M., et al. (2021).
    "Meta-Algorithm Selection."
    https://arxiv.org/abs/2107.09414

    Important: All base selectors and the meta selector must have RETURN_TYPE 'single'
    """

    base_selectors: list[AbstractSelector]
    meta_selector: AbstractSelector
    _selector_map: dict[str, AbstractSelector]
    base_selectors_: list[AbstractSelector]

    PREFIX = "meta"
    RETURN_TYPE = "single"

    @staticmethod
    def _clone_selector(selector: AbstractSelector) -> AbstractSelector:
        """
        Create a fresh selector copy while preserving constructor configuration.

        Deep copy is preferred here because several selectors do not expose a
        reliable parameter introspection API, and ``sel.__class__()`` would
        silently reset user-provided hyperparameters to defaults.
        """
        return copy.deepcopy(selector)

    def __init__(
        self,
        base_selectors: Sequence[AbstractSelector]
        | Callable[[], list[AbstractSelector]]
        | None = None,
        meta_selector: AbstractSelector | Callable[[], AbstractSelector] | None = None,
        candidate_selectors: list[type] | None = None,
        par_factor: int = 10,
        n_folds: int = 5,
        random_state: int = 42,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the MetaSelector.

        Parameters
        ----------
        base_selectors : Sequence[AbstractSelector] or Callable or None, default=None
            List of base selectors to ensemble, a callable that returns a list of
            selectors, or None if candidate_selectors is provided.
        meta_selector : AbstractSelector or Callable or None, default=None
            The selector to use for choosing among base selectors, or a callable
            that returns a selector instance.
        candidate_selectors : list[type] or None, default=None
            List of selector classes to instantiate as base selectors if
            base_selectors is None.
        par_factor : int, default=10
            Penalty factor for timeout instances when computing meta-features.
            Penalty = budget * par_factor.
        n_folds : int, default=5
            Number of folds for cross-validation during meta-selector training.
        random_state : int, default=42
            Random seed for the KFold cross-validation splitter.
        **kwargs : Any
            Additional keyword arguments passed to parent class.
        """
        super().__init__(**kwargs)

        if base_selectors is None and candidate_selectors is not None:
            base_selectors = []
            for sel_cls in candidate_selectors:
                try:
                    base_selectors.append(sel_cls())
                except Exception:
                    pass

        if callable(base_selectors):
            self.base_selectors = cast(Any, base_selectors)()
        else:
            self.base_selectors = (
                list(base_selectors) if base_selectors is not None else []
            )

        if callable(meta_selector):
            self.meta_selector = cast(Any, meta_selector)()
        else:
            self.meta_selector = cast(AbstractSelector, meta_selector)

        if not self.base_selectors:
            raise ValueError("`base_selectors` list cannot be empty.")
        if self.meta_selector is None:
            raise ValueError("`meta_selector` cannot be None.")

        for sel in self.base_selectors:
            if getattr(sel, "RETURN_TYPE", None) != "single":
                raise ValueError(
                    f"Base selector {sel.__class__.__name__} must have RETURN_TYPE 'single'."
                )
        if getattr(self.meta_selector, "RETURN_TYPE", None) != "single":
            raise ValueError("Meta selector must have RETURN_TYPE 'single'.")

        self.par_factor = int(par_factor)
        self.n_folds = int(n_folds)
        self.random_state = int(random_state)

        self.selector_names = [
            f"{s.__class__.__name__}_{i}" for i, s in enumerate(self.base_selectors)
        ]
        self._selector_map: dict[str, AbstractSelector] = {
            name: sel for name, sel in zip(self.selector_names, self.base_selectors)
        }
        self.base_selectors_: list[AbstractSelector] | None = None

    def _fit(
        self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs: Any
    ) -> None:
        """
        Fit the MetaSelector.

        Parameters
        ----------
        features : pd.DataFrame
            Training features (instances x features).
        performance : pd.DataFrame
            Training performance (instances x algorithms).
        """
        penalty = float(self.budget or 0) * float(self.par_factor)
        n_instances = len(features)
        meta_performance = pd.DataFrame(
            index=features.index,
            columns=pd.Index(list(self.selector_names)),
            dtype=float,
        )
        meta_performance[:] = np.nan

        n_splits = min(self.n_folds, max(2, n_instances))
        kf = KFold(n_splits=n_splits, shuffle=True, random_state=self.random_state)

        for train_idx, val_idx in kf.split(np.arange(n_instances)):
            train_ix = features.index[train_idx]
            val_ix = features.index[val_idx]

            X_train = features.loc[train_ix]
            Y_train = performance.loc[train_ix]
            X_val = features.loc[val_ix]

            for sel_idx, sel in enumerate(self.base_selectors):
                sel_copy = self._clone_selector(sel)
                sel_copy.fit(X_train, Y_train)
                preds = cast(dict[str, Any], sel_copy.predict(X_val))

                col = self.selector_names[sel_idx]
                for inst in val_ix:
                    pred = preds.get(str(inst), [])
                    if not pred:
                        meta_performance.at[inst, col] = penalty
                        continue
                    algo_name, _ = pred[0]
                    rt = performance.at[inst, algo_name]
                    if pd.isna(rt) or float(rt) >= float(self.budget or 0):
                        meta_performance.at[inst, col] = penalty
                    else:
                        meta_performance.at[inst, col] = float(rt)

        meta_performance.fillna(penalty, inplace=True)

        self.base_selectors_ = []
        for sel in self.base_selectors:
            sel_full = self._clone_selector(sel)
            sel_full.fit(features, performance)
            self.base_selectors_.append(sel_full)

        self._selector_map = {
            name: sel for name, sel in zip(self.selector_names, self.base_selectors_)
        }

        self.meta_selector.fit(features, meta_performance)

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict with the fitted MetaSelector.

        Parameters
        ----------
        features : pd.DataFrame
            Feature matrix for the test instances.

        Returns
        -------
        dict
            Mapping from instance names to algorithm schedules.
        """
        meta_predictions = cast(dict[Any, Any], self.meta_selector.predict(features))

        if features is None:
            raise ValueError("MetaSelector require features for prediction.")
        final_predictions: dict[str, list[tuple[str, float]]] = {}
        for instance_name, chosen_selector_list in meta_predictions.items():
            if not chosen_selector_list:
                final_predictions[str(instance_name)] = []
                continue

            chosen_selector_name = str(chosen_selector_list[0][0])
            chosen_selector = self._selector_map.get(chosen_selector_name)
            if chosen_selector is None:
                final_predictions[str(instance_name)] = []
                continue

            orig_instance_name = instance_name
            instance_features = features.loc[[orig_instance_name]]
            final_prediction = cast(
                dict[Any, Any], chosen_selector.predict(instance_features)
            )
            preds = final_prediction.get(
                instance_name, final_prediction.get(str(instance_name), [])
            )

            formatted = []
            for entry in preds:
                algo, _ = entry
                formatted.append((str(algo), float(self.budget or 0)))

            final_predictions[str(instance_name)] = formatted
        return final_predictions

    @staticmethod
    def _define_hyperparameters(
        candidate_selectors: list[type] | None = None, **kwargs: Any
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for MetaSelector.

        Parameters
        ----------
        candidate_selectors : list[type] or None, default=None
            List of selector classes to choose from.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        if candidate_selectors is None:
            return [], [], []

        meta_selector_param = ClassChoice(
            name="meta_selector",
            choices=cast(list[type | bool], candidate_selectors),
            default=candidate_selectors[0],
        )

        par_factor_param = Integer(
            name="par_factor",
            bounds=(1, 100),
            default=10,
        )

        n_folds_param = Integer(
            name="n_folds",
            bounds=(2, 10),
            default=5,
        )

        params = [
            meta_selector_param,
            par_factor_param,
            n_folds_param,
        ]

        return params, [], []

__init__(base_selectors=None, meta_selector=None, candidate_selectors=None, par_factor=10, n_folds=5, random_state=42, **kwargs)

Initialize the MetaSelector.

Parameters

base_selectors : Sequence[AbstractSelector] or Callable or None, default=None List of base selectors to ensemble, a callable that returns a list of selectors, or None if candidate_selectors is provided. meta_selector : AbstractSelector or Callable or None, default=None The selector to use for choosing among base selectors, or a callable that returns a selector instance. candidate_selectors : list[type] or None, default=None List of selector classes to instantiate as base selectors if base_selectors is None. par_factor : int, default=10 Penalty factor for timeout instances when computing meta-features. Penalty = budget * par_factor. n_folds : int, default=5 Number of folds for cross-validation during meta-selector training. random_state : int, default=42 Random seed for the KFold cross-validation splitter. **kwargs : Any Additional keyword arguments passed to parent class.

Source code in asf/selectors/meta_selector.py

def __init__(
    self,
    base_selectors: Sequence[AbstractSelector]
    | Callable[[], list[AbstractSelector]]
    | None = None,
    meta_selector: AbstractSelector | Callable[[], AbstractSelector] | None = None,
    candidate_selectors: list[type] | None = None,
    par_factor: int = 10,
    n_folds: int = 5,
    random_state: int = 42,
    **kwargs: Any,
) -> None:
    """
    Initialize the MetaSelector.

    Parameters
    ----------
    base_selectors : Sequence[AbstractSelector] or Callable or None, default=None
        List of base selectors to ensemble, a callable that returns a list of
        selectors, or None if candidate_selectors is provided.
    meta_selector : AbstractSelector or Callable or None, default=None
        The selector to use for choosing among base selectors, or a callable
        that returns a selector instance.
    candidate_selectors : list[type] or None, default=None
        List of selector classes to instantiate as base selectors if
        base_selectors is None.
    par_factor : int, default=10
        Penalty factor for timeout instances when computing meta-features.
        Penalty = budget * par_factor.
    n_folds : int, default=5
        Number of folds for cross-validation during meta-selector training.
    random_state : int, default=42
        Random seed for the KFold cross-validation splitter.
    **kwargs : Any
        Additional keyword arguments passed to parent class.
    """
    super().__init__(**kwargs)

    if base_selectors is None and candidate_selectors is not None:
        base_selectors = []
        for sel_cls in candidate_selectors:
            try:
                base_selectors.append(sel_cls())
            except Exception:
                pass

    if callable(base_selectors):
        self.base_selectors = cast(Any, base_selectors)()
    else:
        self.base_selectors = (
            list(base_selectors) if base_selectors is not None else []
        )

    if callable(meta_selector):
        self.meta_selector = cast(Any, meta_selector)()
    else:
        self.meta_selector = cast(AbstractSelector, meta_selector)

    if not self.base_selectors:
        raise ValueError("`base_selectors` list cannot be empty.")
    if self.meta_selector is None:
        raise ValueError("`meta_selector` cannot be None.")

    for sel in self.base_selectors:
        if getattr(sel, "RETURN_TYPE", None) != "single":
            raise ValueError(
                f"Base selector {sel.__class__.__name__} must have RETURN_TYPE 'single'."
            )
    if getattr(self.meta_selector, "RETURN_TYPE", None) != "single":
        raise ValueError("Meta selector must have RETURN_TYPE 'single'.")

    self.par_factor = int(par_factor)
    self.n_folds = int(n_folds)
    self.random_state = int(random_state)

    self.selector_names = [
        f"{s.__class__.__name__}_{i}" for i, s in enumerate(self.base_selectors)
    ]
    self._selector_map: dict[str, AbstractSelector] = {
        name: sel for name, sel in zip(self.selector_names, self.base_selectors)
    }
    self.base_selectors_: list[AbstractSelector] | None = None

MultiClassClassifier

Bases: ConfigurableMixin, AbstractModelBasedSelector

Multi-class classification algorithm selector.

References

Xu, L., et al. (2013). "SATzilla2012: Improved Algorithm Selection Based on Cost-sensitive Classification Models." https://arxiv.org/abs/1306.1031

Attributes

classifier : AbstractPredictor or None The trained classification model.

Source code in asf/selectors/multi_class.py

class MultiClassClassifier(ConfigurableMixin, AbstractModelBasedSelector):
    """
    Multi-class classification algorithm selector.

    References
    ----------
    Xu, L., et al. (2013).
    "SATzilla2012: Improved Algorithm Selection Based on Cost-sensitive Classification Models."
    https://arxiv.org/abs/1306.1031

    Attributes
    ----------
    classifier : AbstractPredictor or None
        The trained classification model.
    """

    PREFIX = "multi_class_classifier"
    RETURN_TYPE = "single"

    def __init__(
        self,
        model_class: type[AbstractPredictor] = RandomForestClassifierWrapper,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the MultiClassClassifier.

        Parameters
        ----------
        model_class : type[AbstractPredictor], default=RandomForestClassifierWrapper
            The class of the model to be used for classification.
        **kwargs : Any
            Additional keyword arguments.
        """
        AbstractModelBasedSelector.__init__(self, model_class, **kwargs)
        self.classifier: AbstractPredictor | None = None
        self._observed_class_ids: np.ndarray | None = None

    def _fit(
        self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs: Any
    ) -> None:
        """
        Fit the classification model.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame
            The algorithm performance data.
        """
        if self.algorithm_features is not None:
            raise ValueError("MultiClassClassifier does not use algorithm features.")

        self.classifier = self.model_class()
        if self.classifier is None:
            raise RuntimeError("Classifier could not be initialized.")

        # Best algorithm (lowest value) per instance.
        target_global = np.argmin(performance.values, axis=1).astype(int)
        observed_class_ids = np.unique(target_global)
        class_to_local = {
            int(class_id): local_idx
            for local_idx, class_id in enumerate(observed_class_ids.tolist())
        }
        target_local = np.asarray(
            [class_to_local[int(class_id)] for class_id in target_global],
            dtype=int,
        )

        self._observed_class_ids = observed_class_ids
        self.classifier.fit(features, target_local)

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict the best algorithm for each instance.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.

        Returns
        -------
        dict
            Mapping from instance names to algorithm schedules.
        """
        if self.classifier is None:
            raise RuntimeError("Classifier has not been fitted.")

        if features is None:
            raise ValueError("MultiClassClassifier require features for prediction.")
        predictions = self.classifier.predict(features)
        if self._observed_class_ids is None:
            raise RuntimeError(
                "Classifier class mapping was not initialized during fit."
            )

        results: dict[str, list[tuple[str, float]]] = {}
        for i, instance_name in enumerate(features.index):
            local_idx = int(predictions[i])
            idx = int(self._observed_class_ids[local_idx])
            results[str(instance_name)] = [
                (str(self.algorithms[idx]), float(self.budget or 0))
            ]
        return results

    @staticmethod
    def _define_hyperparameters(
        model_class: list[type[AbstractPredictor]] | None = None,
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for MultiClassClassifier.

        Parameters
        ----------
        model_class : list[type[AbstractPredictor]] or None, default=None
            List of model classes to include in the configuration space.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        if model_class is None:
            model_class = [RandomForestClassifierWrapper, XGBoostClassifierWrapper]

        hyperparameters = [
            ClassChoice(
                "model_class",
                choices=cast(list[type | bool], model_class),
                default=model_class[0],
            ),
        ]
        return hyperparameters, [], []

__init__(model_class=RandomForestClassifierWrapper, **kwargs)

Initialize the MultiClassClassifier.

Parameters

model_class : type[AbstractPredictor], default=RandomForestClassifierWrapper The class of the model to be used for classification. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/multi_class.py

def __init__(
    self,
    model_class: type[AbstractPredictor] = RandomForestClassifierWrapper,
    **kwargs: Any,
) -> None:
    """
    Initialize the MultiClassClassifier.

    Parameters
    ----------
    model_class : type[AbstractPredictor], default=RandomForestClassifierWrapper
        The class of the model to be used for classification.
    **kwargs : Any
        Additional keyword arguments.
    """
    AbstractModelBasedSelector.__init__(self, model_class, **kwargs)
    self.classifier: AbstractPredictor | None = None
    self._observed_class_ids: np.ndarray | None = None

OSLLinearSelector

Bases: ConfigurableMixin, AbstractSelector

Selector using Optimistic Superset Loss (OSL) to predict runtimes.

References

Hanselle, J., et al. (2021). "Superset Learning for Algorithm Selection with Right Censored Data." https://epub.ub.uni-muenchen.de/91670/1/_PAKDD__Superset_Learning_for_Algorithm_Selection_with_Right_Censored_Data.pdf

Attributes

reg : float L2 regularization strength. optimizer_method : str Method for scipy.optimize.minimize. maxiter : int Maximum number of optimizer iterations. tol : float or None Tolerance for the optimizer. thetas : dict[str, np.ndarray] Learned parameters for each algorithm.

Source code in asf/selectors/osl_linear.py

class OSLLinearSelector(ConfigurableMixin, AbstractSelector):
    """
    Selector using Optimistic Superset Loss (OSL) to predict runtimes.

    References
    ----------
    Hanselle, J., et al. (2021).
    "Superset Learning for Algorithm Selection with Right Censored Data."
    https://epub.ub.uni-muenchen.de/91670/1/_PAKDD__Superset_Learning_for_Algorithm_Selection_with_Right_Censored_Data.pdf

    Attributes
    ----------
    reg : float
        L2 regularization strength.
    optimizer_method : str
        Method for scipy.optimize.minimize.
    maxiter : int
        Maximum number of optimizer iterations.
    tol : float or None
        Tolerance for the optimizer.
    thetas : dict[str, np.ndarray]
        Learned parameters for each algorithm.
    """

    PREFIX = "osl_linear"
    RETURN_TYPE = "single"

    def __init__(
        self,
        budget: float,
        reg: float = 0.0,
        optimizer_method: str = "L-BFGS-B",
        maxiter: int = 1000,
        tol: float | None = None,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the OSLLinearSelector.

        Parameters
        ----------
        budget : float
            Global cutoff time.
        reg : float, default=0.0
            L2 regularization strength.
        optimizer_method : str, default="L-BFGS-B"
            Optimization algorithm name for scipy.optimize.minimize.
        maxiter : int, default=1000
            Maximum number of optimizer iterations.
        tol : float or None, default=None
            Tolerance for the optimizer.
        **kwargs : Any
            Additional keyword arguments.
        """
        super().__init__(**kwargs)
        self.budget = float(budget)
        self.reg = float(reg)
        self.optimizer_method = optimizer_method
        self.maxiter = int(maxiter)
        self.tol = None if tol is None else float(tol)
        self.thetas: dict[str, np.ndarray] = {}

    def _osl_obj_grad(
        self,
        theta: np.ndarray,
        X: np.ndarray,
        y: np.ndarray,
        censored_mask: np.ndarray,
        C: float,
    ) -> tuple[float, np.ndarray]:
        """
                Compute loss and gradient for parameters theta.

                Parameters
                ----------
                theta : np.ndarray
                    Parameter vector.
                X : np.ndarray
                    Design matrix.
                y : np.ndarray
                    Observed runtimes.
                censored_mask : np.ndarray
                    Boolean mask of censored observations.
                C : float
                    Cutoff time.

                Returns
        -------
                tuple
                    Tuple of (loss, gradient).
        """
        preds = X.dot(theta)
        precise_mask = ~censored_mask & ~np.isnan(y)
        cens_pred_mask = censored_mask & (preds < C)

        loss_precise = (
            float(((y[precise_mask] - preds[precise_mask]) ** 2).sum())
            if precise_mask.any()
            else 0.0
        )
        loss_cens = (
            float(((C - preds[cens_pred_mask]) ** 2).sum())
            if cens_pred_mask.any()
            else 0.0
        )
        loss = loss_precise + loss_cens

        if self.reg:
            loss += 0.5 * self.reg * float(np.sum(theta**2))

        grad = np.zeros_like(theta)
        if precise_mask.any():
            resid = y[precise_mask] - preds[precise_mask]
            grad_prec = -2.0 * (X[precise_mask].T.dot(resid))
            grad += grad_prec
        if cens_pred_mask.any():
            diff = C - preds[cens_pred_mask]
            grad_cens = -2.0 * (X[cens_pred_mask].T.dot(diff))
            grad += grad_cens

        if self.reg:
            grad += self.reg * theta

        return float(loss), grad

    def _fit(
        self,
        features: pd.DataFrame,
        performance: pd.DataFrame,
        **kwargs: Any,
    ) -> None:
        """
        Fit linear models for each algorithm by minimizing OSL.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame
            The algorithm performance data.
        **kwargs : Any
            Additional keyword arguments.
        """
        X_base = np.asarray(features, dtype=float)
        n, d = X_base.shape
        X = np.hstack([X_base, np.ones((n, 1), dtype=float)])

        self.algorithms = [str(a) for a in performance.columns]
        thetas: dict[str, np.ndarray] = {}

        censored = (performance >= self.budget) | performance.isna()

        for algo in self.algorithms:
            y_col = performance[algo].to_numpy(dtype=float)
            cens_mask = censored[algo].to_numpy(dtype=bool)

            try:
                unc_idx = (~cens_mask) & (~np.isnan(y_col))
                if unc_idx.sum() >= d + 1:
                    theta0, *_ = np.linalg.lstsq(X[unc_idx], y_col[unc_idx], rcond=None)
                    theta0 = np.asarray(theta0, dtype=float)
                else:
                    theta0 = np.zeros(d + 1, dtype=float)
            except Exception:
                theta0 = np.zeros(d + 1, dtype=float)

            def fun_and_grad(th: np.ndarray) -> tuple[float, np.ndarray]:
                val, grad = self._osl_obj_grad(
                    th, X, y_col, cens_mask, float(self.budget or 0)
                )
                return val, grad

            res = minimize(
                fun=lambda th: fun_and_grad(th)[0],
                x0=theta0,
                jac=lambda th: fun_and_grad(th)[1],
                method=self.optimizer_method,
                tol=self.tol,
                options={"maxiter": self.maxiter, "disp": False},
            )
            theta_opt = np.asarray(res.x, dtype=float)
            thetas[str(algo)] = theta_opt

        self.thetas = thetas

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict best algorithm per instance.

        Parameters
        ----------
        features : pd.DataFrame or None
            The input features.
        performance : pd.DataFrame or None, default=None
            Partial performance data.

        Returns
        -------
        dict
            Mapping from instance name to algorithm schedules.
        """
        if features is None:
            raise ValueError("OSLLinearSelector requires features for prediction.")
        X_base = np.asarray(features, dtype=float)
        n = X_base.shape[0]
        X = np.hstack([X_base, np.ones((n, 1), dtype=float)])

        preds_per_algo = {}
        for algo, theta in self.thetas.items():
            preds = X.dot(theta)
            preds_per_algo[algo] = np.asarray(preds, dtype=float)

        out: dict[str, list[tuple[str, float]]] = {}
        algs = [str(a) for a in self.algorithms]
        for i, idx in enumerate(features.index):
            best_algo = None
            best_val = float("inf")
            for algo in algs:
                val = float(preds_per_algo[algo][i])
                if val < best_val:
                    best_val = val
                    best_algo = algo
            out[str(idx)] = [(str(best_algo), float(self.budget or 0))]
        return out

    @staticmethod
    def _define_hyperparameters(
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for OSLLinearSelector.

        Parameters
        ----------
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        reg_param = Float(
            name="reg",
            bounds=(0.0, 10.0),
            default=0.0,
        )

        optimizer_method_param = Categorical(
            name="optimizer_method",
            items=["L-BFGS-B", "CG", "BFGS", "TNC", "SLSQP"],
            default="L-BFGS-B",
        )

        maxiter_param = Integer(
            name="maxiter",
            bounds=(100, 5000),
            default=1000,
        )

        tol_param = Float(
            name="tol",
            bounds=(1e-6, 1e-2),
            log=True,
            default=1e-5,
        )

        params = [
            reg_param,
            optimizer_method_param,
            maxiter_param,
            tol_param,
        ]

        return params, [], []

__init__(budget, reg=0.0, optimizer_method='L-BFGS-B', maxiter=1000, tol=None, **kwargs)

Initialize the OSLLinearSelector.

Parameters

budget : float Global cutoff time. reg : float, default=0.0 L2 regularization strength. optimizer_method : str, default="L-BFGS-B" Optimization algorithm name for scipy.optimize.minimize. maxiter : int, default=1000 Maximum number of optimizer iterations. tol : float or None, default=None Tolerance for the optimizer. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/osl_linear.py

def __init__(
    self,
    budget: float,
    reg: float = 0.0,
    optimizer_method: str = "L-BFGS-B",
    maxiter: int = 1000,
    tol: float | None = None,
    **kwargs: Any,
) -> None:
    """
    Initialize the OSLLinearSelector.

    Parameters
    ----------
    budget : float
        Global cutoff time.
    reg : float, default=0.0
        L2 regularization strength.
    optimizer_method : str, default="L-BFGS-B"
        Optimization algorithm name for scipy.optimize.minimize.
    maxiter : int, default=1000
        Maximum number of optimizer iterations.
    tol : float or None, default=None
        Tolerance for the optimizer.
    **kwargs : Any
        Additional keyword arguments.
    """
    super().__init__(**kwargs)
    self.budget = float(budget)
    self.reg = float(reg)
    self.optimizer_method = optimizer_method
    self.maxiter = int(maxiter)
    self.tol = None if tol is None else float(tol)
    self.thetas: dict[str, np.ndarray] = {}

PairwiseClassifier

Bases: ConfigurableMixin, AbstractModelBasedSelector, AbstractFeatureGenerator

Selector using pairwise comparison of algorithms.

References

Xu, L., et al. (2012). "SATzilla Solver Description." https://ml.informatik.uni-freiburg.de/wp-content/uploads/papers/12-SATzilla-solver-description.pdf

Attributes

classifiers : list[AbstractPredictor] Trained classifiers for pairwise comparisons. use_weights : bool Whether to use weights based on performance differences.

Source code in asf/selectors/pairwise_classifier.py

class PairwiseClassifier(
    ConfigurableMixin, AbstractModelBasedSelector, AbstractFeatureGenerator
):
    """
    Selector using pairwise comparison of algorithms.

    References
    ----------
    Xu, L., et al. (2012).
    "SATzilla Solver Description."
    https://ml.informatik.uni-freiburg.de/wp-content/uploads/papers/12-SATzilla-solver-description.pdf

    Attributes
    ----------
    classifiers : list[AbstractPredictor]
        Trained classifiers for pairwise comparisons.
    use_weights : bool
        Whether to use weights based on performance differences.
    """

    PREFIX = "pairwise_classifier"
    RETURN_TYPE = "single"

    def __init__(
        self,
        model_class: type[AbstractPredictor] = RandomForestClassifierWrapper,
        use_weights: bool = True,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the PairwiseClassifier.

        Parameters
        ----------
        model_class : type[AbstractPredictor], default=RandomForestClassifierWrapper
            The classifier model class used for pairwise comparisons.
        use_weights : bool, default=True
            Whether to use weights based on performance differences.
        **kwargs : Any
            Additional keyword arguments.
        """
        AbstractModelBasedSelector.__init__(self, model_class, **kwargs)
        AbstractFeatureGenerator.__init__(self)
        self.classifiers: list[AbstractPredictor | _ConstantBinaryPredictor] = []
        self.use_weights = bool(use_weights)

    def _fit(
        self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs: Any
    ) -> None:
        """
        Fit the pairwise classifiers.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame
            The algorithm performance data.
        """
        if self.algorithm_features is not None:
            raise ValueError("PairwiseClassifier does not use algorithm features.")

        self.classifiers = []
        for i, algorithm in enumerate(self.algorithms):
            for other_algorithm in self.algorithms[i + 1 :]:
                val1 = performance[algorithm].to_numpy(dtype=float)
                val2 = performance[other_algorithm].to_numpy(dtype=float)

                if self.maximize:
                    diffs = (val1 > val2).astype(int)
                else:
                    diffs = (val1 < val2).astype(int)

                unique_diffs = np.unique(diffs)
                if unique_diffs.size == 1:
                    # Some folds/task groups can contain only one class for a pair.
                    # Keep behavior deterministic instead of training a model that may fail.
                    self.classifiers.append(
                        _ConstantBinaryPredictor(int(unique_diffs[0]))
                    )
                    continue

                cur_model = self.model_class()
                if cur_model is None:
                    raise RuntimeError("Classifier could not be initialized.")

                cur_model.fit(
                    features,
                    diffs,
                    sample_weight=None if not self.use_weights else np.abs(val1 - val2),
                )
                self.classifiers.append(cur_model)

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict the best algorithm for each instance.

        Parameters
        ----------
        features : pd.DataFrame
            The query instance features.

        Returns
        -------
        dict
            Mapping from instance name to algorithm schedules.
        """
        if features is None:
            raise ValueError("PairwiseClassifier require features for prediction.")
        votes = self.generate_features(features)
        result: dict[str, list[tuple[str, float]]] = {}
        for instance in features.index:
            best_algo = votes.loc[instance].idxmax()
            result[str(instance)] = [(str(best_algo), float(self.budget or 0))]
        return result

    def generate_features(self, base_features: pd.DataFrame) -> pd.DataFrame:
        """
                Generate vote counts for each algorithm.

                Parameters
                ----------
                base_features : pd.DataFrame
                    The input features.

                Returns
        -------
                pd.DataFrame
                    DataFrame of vote counts for each algorithm.
        """
        # Ensure input is a DataFrame
        if not isinstance(base_features, pd.DataFrame):
            cols = (
                self.features
                if self.features
                else [f"f_{i}" for i in range(base_features.shape[1])]
            )
            features_df = pd.DataFrame(base_features, columns=list(cols))
        else:
            features_df = base_features

        votes = pd.DataFrame(
            0,
            index=features_df.index,
            columns=pd.Index(list(self.algorithms)),
        )
        cnt = 0
        for i, algo1 in enumerate(self.algorithms):
            for _j, algo2 in enumerate(self.algorithms[i + 1 :]):
                pred = self.classifiers[cnt].predict(features_df)
                # 1 means algo1 is better, 0 means algo2 is better
                votes.loc[features_df.index[pred == 1], algo1] += 1
                votes.loc[features_df.index[pred == 0], algo2] += 1
                cnt += 1
        return votes

    @staticmethod
    def _define_hyperparameters(
        model_class: list[type[AbstractPredictor]] | None = None,
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for PairwiseClassifier.

        Parameters
        ----------
        model_class : list[type[AbstractPredictor]] or None, default=None
            List of model classes.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        if model_class is None:
            model_class = [RandomForestClassifierWrapper, XGBoostClassifierWrapper]

        hyperparameters = [
            ClassChoice("model_class", choices=cast(list[type | bool], model_class)),
            Categorical("use_weights", items=[True, False], default=True),
        ]
        return hyperparameters, [], []

__init__(model_class=RandomForestClassifierWrapper, use_weights=True, **kwargs)

Initialize the PairwiseClassifier.

Parameters

model_class : type[AbstractPredictor], default=RandomForestClassifierWrapper The classifier model class used for pairwise comparisons. use_weights : bool, default=True Whether to use weights based on performance differences. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/pairwise_classifier.py

def __init__(
    self,
    model_class: type[AbstractPredictor] = RandomForestClassifierWrapper,
    use_weights: bool = True,
    **kwargs: Any,
) -> None:
    """
    Initialize the PairwiseClassifier.

    Parameters
    ----------
    model_class : type[AbstractPredictor], default=RandomForestClassifierWrapper
        The classifier model class used for pairwise comparisons.
    use_weights : bool, default=True
        Whether to use weights based on performance differences.
    **kwargs : Any
        Additional keyword arguments.
    """
    AbstractModelBasedSelector.__init__(self, model_class, **kwargs)
    AbstractFeatureGenerator.__init__(self)
    self.classifiers: list[AbstractPredictor | _ConstantBinaryPredictor] = []
    self.use_weights = bool(use_weights)

generate_features(base_features)

    Generate vote counts for each algorithm.

    Parameters
    ----------
    base_features : pd.DataFrame
        The input features.

    Returns

---

    pd.DataFrame
        DataFrame of vote counts for each algorithm.

Source code in asf/selectors/pairwise_classifier.py

def generate_features(self, base_features: pd.DataFrame) -> pd.DataFrame:
    """
            Generate vote counts for each algorithm.

            Parameters
            ----------
            base_features : pd.DataFrame
                The input features.

            Returns
    -------
            pd.DataFrame
                DataFrame of vote counts for each algorithm.
    """
    # Ensure input is a DataFrame
    if not isinstance(base_features, pd.DataFrame):
        cols = (
            self.features
            if self.features
            else [f"f_{i}" for i in range(base_features.shape[1])]
        )
        features_df = pd.DataFrame(base_features, columns=list(cols))
    else:
        features_df = base_features

    votes = pd.DataFrame(
        0,
        index=features_df.index,
        columns=pd.Index(list(self.algorithms)),
    )
    cnt = 0
    for i, algo1 in enumerate(self.algorithms):
        for _j, algo2 in enumerate(self.algorithms[i + 1 :]):
            pred = self.classifiers[cnt].predict(features_df)
            # 1 means algo1 is better, 0 means algo2 is better
            votes.loc[features_df.index[pred == 1], algo1] += 1
            votes.loc[features_df.index[pred == 0], algo2] += 1
            cnt += 1
    return votes

PairwiseRegressor

Bases: ConfigurableMixin, AbstractModelBasedSelector, AbstractFeatureGenerator

Selector using pairwise regression of algorithms.

Attributes

regressors : list[AbstractPredictor] Trained regressors for pairwise comparisons.

Source code in asf/selectors/pairwise_regressor.py

class PairwiseRegressor(
    ConfigurableMixin, AbstractModelBasedSelector, AbstractFeatureGenerator
):
    """
    Selector using pairwise regression of algorithms.

    Attributes
    ----------
    regressors : list[AbstractPredictor]
        Trained regressors for pairwise comparisons.
    """

    PREFIX = "pairwise_regressor"
    RETURN_TYPE = "single"

    def __init__(
        self,
        model_class: type[AbstractPredictor] = RandomForestRegressorWrapper,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the PairwiseRegressor.

        Parameters
        ----------
        model_class : type[AbstractPredictor], default=RandomForestRegressorWrapper
            The regression model class used for pairwise comparisons.
        **kwargs : Any
            Additional keyword arguments.
        """
        AbstractModelBasedSelector.__init__(self, model_class, **kwargs)
        AbstractFeatureGenerator.__init__(self)
        self.regressors: list[AbstractPredictor] = []

    def _fit(
        self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs: Any
    ) -> None:
        """
        Fit the pairwise regressors.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame
            The algorithm performance data.
        """
        if self.algorithm_features is not None:
            raise ValueError("PairwiseRegressor does not use algorithm features.")

        self.regressors = []
        for i, algorithm in enumerate(self.algorithms):
            for other_algorithm in self.algorithms[i + 1 :]:
                val1 = performance[algorithm].to_numpy(dtype=float)
                val2 = performance[other_algorithm].to_numpy(dtype=float)

                diffs = val1 - val2
                cur_model = self.model_class()
                if cur_model is None:
                    raise RuntimeError("Regressor could not be initialized.")

                cur_model.fit(features, diffs)
                self.regressors.append(cur_model)

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
                Predict the best algorithm for each instance.

                Parameters
                ----------
                features : pd.DataFrame
                    The query instance features.

                Returns
        -------
                dict
                    Mapping from instance name to algorithm schedules.
        """
        if features is None:
            raise ValueError("PairwiseRegressor require features for prediction.")
        scores = self.generate_features(features)
        result: dict[str, list[tuple[str, float]]] = {}
        for instance in features.index:
            # If maximizing, we want the highest combined score (algo1 - algo2 > 0)
            # If minimizing, we want the lowest combined score
            if self.maximize:
                best_algo = scores.loc[instance].idxmax()
            else:
                best_algo = scores.loc[instance].idxmin()
            result[str(instance)] = [(str(best_algo), float(self.budget or 0))]
        return result

    def generate_features(self, base_features: pd.DataFrame) -> pd.DataFrame:
        """
                Generate pairwise comparisons for each algorithm.

                Parameters
                ----------
                features : pd.DataFrame
                    The input features.

                Returns
        -------
                pd.DataFrame
                    DataFrame of aggregated regression values for each algorithm.
        """
        # Ensure input is a DataFrame
        if not isinstance(base_features, pd.DataFrame):
            cols = (
                self.features
                if self.features
                else [f"f_{i}" for i in range(base_features.shape[1])]
            )
            features_df = pd.DataFrame(base_features, columns=list(cols))
        else:
            features_df = base_features

        scores = pd.DataFrame(
            0.0,
            index=features_df.index,
            columns=pd.Index(list(self.algorithms)),
        )
        cnt = 0
        for i, algo1 in enumerate(self.algorithms):
            for _j, algo2 in enumerate(self.algorithms[i + 1 :]):
                pred = self.regressors[cnt].predict(features_df)
                # pred is algo1_perf - algo2_perf
                scores.loc[features_df.index, algo1] += pred
                scores.loc[features_df.index, algo2] -= pred
                cnt += 1
        return scores

    @staticmethod
    def _define_hyperparameters(
        model_class: list[type[AbstractPredictor]] | None = None,
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for PairwiseRegressor.

        Parameters
        ----------
        model_class : list[type[AbstractPredictor]] or None, default=None
            List of model classes.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        if model_class is None:
            model_class = [RandomForestRegressorWrapper, XGBoostRegressorWrapper]

        hyperparameters = [
            ClassChoice("model_class", choices=cast(list[type | bool], model_class)),
        ]
        return hyperparameters, [], []

__init__(model_class=RandomForestRegressorWrapper, **kwargs)

Initialize the PairwiseRegressor.

Parameters

model_class : type[AbstractPredictor], default=RandomForestRegressorWrapper The regression model class used for pairwise comparisons. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/pairwise_regressor.py

def __init__(
    self,
    model_class: type[AbstractPredictor] = RandomForestRegressorWrapper,
    **kwargs: Any,
) -> None:
    """
    Initialize the PairwiseRegressor.

    Parameters
    ----------
    model_class : type[AbstractPredictor], default=RandomForestRegressorWrapper
        The regression model class used for pairwise comparisons.
    **kwargs : Any
        Additional keyword arguments.
    """
    AbstractModelBasedSelector.__init__(self, model_class, **kwargs)
    AbstractFeatureGenerator.__init__(self)
    self.regressors: list[AbstractPredictor] = []

generate_features(base_features)

    Generate pairwise comparisons for each algorithm.

    Parameters
    ----------
    features : pd.DataFrame
        The input features.

    Returns

---

    pd.DataFrame
        DataFrame of aggregated regression values for each algorithm.

Source code in asf/selectors/pairwise_regressor.py

def generate_features(self, base_features: pd.DataFrame) -> pd.DataFrame:
    """
            Generate pairwise comparisons for each algorithm.

            Parameters
            ----------
            features : pd.DataFrame
                The input features.

            Returns
    -------
            pd.DataFrame
                DataFrame of aggregated regression values for each algorithm.
    """
    # Ensure input is a DataFrame
    if not isinstance(base_features, pd.DataFrame):
        cols = (
            self.features
            if self.features
            else [f"f_{i}" for i in range(base_features.shape[1])]
        )
        features_df = pd.DataFrame(base_features, columns=list(cols))
    else:
        features_df = base_features

    scores = pd.DataFrame(
        0.0,
        index=features_df.index,
        columns=pd.Index(list(self.algorithms)),
    )
    cnt = 0
    for i, algo1 in enumerate(self.algorithms):
        for _j, algo2 in enumerate(self.algorithms[i + 1 :]):
            pred = self.regressors[cnt].predict(features_df)
            # pred is algo1_perf - algo2_perf
            scores.loc[features_df.index, algo1] += pred
            scores.loc[features_df.index, algo2] -= pred
            cnt += 1
    return scores

PerformanceModel

Bases: ConfigurableMixin, AbstractModelBasedSelector, AbstractFeatureGenerator

PerformanceModel predicts algorithm performance based on instance features.

It can handle both single-target (one model per algorithm) and multi-target regression models.

References

Hutter, F., et al. (2011). "Sequential Model-Based Optimization for General Algorithm Configuration." https://arxiv.org/abs/1111.2249

Attributes

model_class : type The class of the regression model to be used. use_multi_target : bool Whether to use multi-target regression. normalize : AbstractNormalization Method to normalize the performance data. regressors : list or object Trained regression models.

Source code in asf/selectors/performance_model.py

class PerformanceModel(
    ConfigurableMixin, AbstractModelBasedSelector, AbstractFeatureGenerator
):
    """
    PerformanceModel predicts algorithm performance based on instance features.

    It can handle both single-target (one model per algorithm) and multi-target
    regression models.

    References
    ----------
    Hutter, F., et al. (2011).
    "Sequential Model-Based Optimization for General Algorithm Configuration."
    https://arxiv.org/abs/1111.2249

    Attributes
    ----------
    model_class : type
        The class of the regression model to be used.
    use_multi_target : bool
        Whether to use multi-target regression.
    normalize : AbstractNormalization
        Method to normalize the performance data.
    regressors : list or object
        Trained regression models.
    """

    PREFIX = "performance_model"
    RETURN_TYPE = "single"

    def __init__(
        self,
        model_class: type[AbstractPredictor] = RandomForestRegressorWrapper,
        use_multi_target: bool = False,
        normalize: AbstractNormalization | None = None,
        init_params: dict[str, Any] | None = None,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the PerformanceModel.

        Parameters
        ----------
        model_class : type[AbstractPredictor], default=RandomForestRegressorWrapper
            The class of the regression model to be used.
        use_multi_target : bool, default=False
            Indicates whether to use multi-target regression.
        normalize : AbstractNormalization or None, default=None
            Method to normalize performance data. If None, defaults to LogNormalization().
        init_params : dict[str, Any] or None, default=None
            Initialization parameters from configuration.
        **kwargs : Any
            Additional arguments for the parent classes.
        """
        params = init_params if isinstance(init_params, dict) else {}
        params.update(kwargs)

        model_class = params.pop("model_class", model_class)
        use_multi_target = params.pop("use_multi_target", use_multi_target)
        normalize = params.pop("normalize", normalize)

        AbstractModelBasedSelector.__init__(self, model_class, **params)
        AbstractFeatureGenerator.__init__(self)

        self.regressors: list[AbstractPredictor] | AbstractPredictor | None = None
        self.use_multi_target = bool(use_multi_target)

        # Instantiate normalize if it is a partial or class
        if callable(normalize):
            self.normalize = normalize()
        elif normalize is not None:
            self.normalize = normalize
        else:
            self.normalize = LogNormalization()

    def _fit(
        self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs: Any
    ) -> None:
        """
        Fit the regression models.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame
            The performance data.
        """
        if self.normalize is not None:
            normalized = self.normalize.fit_transform(performance)
            # Preserve performance matrix shape if the transformer flattens values.
            if isinstance(normalized, np.ndarray):
                norm_array = np.asarray(normalized)
                if norm_array.shape != performance.shape:
                    if norm_array.size != performance.size:
                        raise ValueError(
                            "Normalization changed performance size unexpectedly: "
                            f"got {norm_array.shape}, expected compatible with {performance.shape}."
                        )
                    norm_array = norm_array.reshape(performance.shape)
                performance = pd.DataFrame(
                    norm_array,
                    index=performance.index,
                    columns=performance.columns,
                )
            else:
                performance = normalized

        regressor_init_args: dict[str, Any] = {}
        # Safely check for input_size if it's a type (standard wrapper classes usually have it)
        try:
            sig = inspect.signature(self.model_class)
            if "input_size" in sig.parameters:
                regressor_init_args["input_size"] = features.shape[1]
        except (ValueError, TypeError):
            pass

        if self.use_multi_target:
            if self.algorithm_features is not None:
                raise ValueError(
                    "PerformanceModel does not use algorithm features for multi-target regression."
                )
            self.regressors = self.model_class(**regressor_init_args)
            self.regressors.fit(features, performance)
        else:
            if self.algorithm_features is None:
                self.regressors = []
                for i, _ in enumerate(self.algorithms):
                    algo_times = performance.iloc[:, i]
                    cur_model = self.model_class(**regressor_init_args)
                    cur_model.fit(features, algo_times)
                    self.regressors.append(cur_model)
            else:
                train_data_list = []
                for i, algorithm in enumerate(self.algorithms):
                    # Align algorithm features with instance features
                    data = pd.merge(
                        features,
                        self.algorithm_features.loc[[algorithm]]
                        .reindex([algorithm] * len(features))
                        .set_index(features.index),
                        left_index=True,
                        right_index=True,
                    )
                    data = pd.merge(
                        data,
                        performance.iloc[:, [i]].rename(
                            columns={performance.columns[i]: "performance"}
                        ),
                        left_index=True,
                        right_index=True,
                    )
                    train_data_list.append(data)
                train_data = pd.concat(train_data_list)
                self.regressors = self.model_class(**regressor_init_args)
                self.regressors.fit(train_data.iloc[:, :-1], train_data.iloc[:, -1])

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict the best algorithm for each instance.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.

        Returns
        -------
        dict
            Mapping from instance name to algorithm schedules.
        """
        if features is None:
            raise ValueError("PerformanceModel require features for prediction.")
        predictions = self.generate_features(features)

        results: dict[str, list[tuple[str, float]]] = {}
        for i, instance_name in enumerate(features.index):
            idx = (
                int(np.argmax(predictions.iloc[i]))
                if self.maximize
                else int(np.argmin(predictions.iloc[i]))
            )
            results[str(instance_name)] = [
                (str(self.algorithms[idx]), float(self.budget or 0))
            ]
        return results

    def generate_features(self, base_features: pd.DataFrame) -> pd.DataFrame:
        """
        Generate predictions for each algorithm.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.

        Returns
        -------
        np.ndarray
            Predicted performance for each algorithm (n_instances x n_algorithms).
        """
        if self.regressors is None:
            raise RuntimeError("Model has not been fitted.")

        predictions = np.zeros((base_features.shape[0], len(self.algorithms)))

        if self.use_multi_target:
            if not isinstance(self.regressors, AbstractPredictor):
                raise RuntimeError("Multi-target regressor missing.")
            predictions = self.regressors.predict(base_features)
            if isinstance(predictions, pd.DataFrame):
                predictions = predictions.values
        else:
            if self.algorithm_features is None:
                if not isinstance(self.regressors, list):
                    raise RuntimeError("Individual regressors missing.")
                for i, _ in enumerate(self.algorithms):
                    regressor = cast(AbstractPredictor, self.regressors[i])
                    predictions[:, i] = np.asarray(
                        regressor.predict(base_features)
                    ).flatten()
            else:
                if not isinstance(self.regressors, AbstractPredictor):
                    raise RuntimeError("Joint regressor missing.")
                regressor: AbstractPredictor = self.regressors
                for i, algorithm in enumerate(self.algorithms):
                    data = pd.merge(
                        base_features,
                        self.algorithm_features.loc[[algorithm]]
                        .reindex([algorithm] * len(base_features))
                        .set_index(base_features.index),
                        left_index=True,
                        right_index=True,
                    )
                    predictions[:, i] = regressor.predict(data)

        return pd.DataFrame(
            predictions,
            index=base_features.index,
            columns=pd.Index(list(self.algorithms)),
        )

    @staticmethod
    def _define_hyperparameters(
        model_class: list[type[AbstractPredictor]] | None = None,
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for PerformanceModel.

        Parameters
        ----------
        model_class : list[type[AbstractPredictor]] or None, default=None
            List of model classes to include in the configuration space.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        if model_class is None:
            model_class = [RandomForestRegressorWrapper, XGBoostRegressorWrapper]

        # Get all normalization options from asf.preprocessing.performance_scaling
        from asf.preprocessing.performance_scaling import (
            BoxCoxNormalization,
            DummyNormalization,
            InvSigmoidNormalization,
            LogNormalization,
            MinMaxNormalization,
            NegExpNormalization,
            SqrtNormalization,
            ZScoreNormalization,
        )

        norm_choices = [
            LogNormalization,
            MinMaxNormalization,
            ZScoreNormalization,
            SqrtNormalization,
            InvSigmoidNormalization,
            NegExpNormalization,
            DummyNormalization,
            BoxCoxNormalization,
        ]

        hyperparameters = [
            ClassChoice(
                "model_class",
                choices=cast(list[type | bool], model_class),
                default=model_class[0],
            ),
            ClassChoice(
                "normalize",
                choices=cast(list[type | bool], norm_choices),
                default=LogNormalization,
            ),
        ]
        return hyperparameters, [], []

__init__(model_class=RandomForestRegressorWrapper, use_multi_target=False, normalize=None, init_params=None, **kwargs)

Initialize the PerformanceModel.

Parameters

model_class : type[AbstractPredictor], default=RandomForestRegressorWrapper The class of the regression model to be used. use_multi_target : bool, default=False Indicates whether to use multi-target regression. normalize : AbstractNormalization or None, default=None Method to normalize performance data. If None, defaults to LogNormalization(). init_params : dict[str, Any] or None, default=None Initialization parameters from configuration. **kwargs : Any Additional arguments for the parent classes.

Source code in asf/selectors/performance_model.py

def __init__(
    self,
    model_class: type[AbstractPredictor] = RandomForestRegressorWrapper,
    use_multi_target: bool = False,
    normalize: AbstractNormalization | None = None,
    init_params: dict[str, Any] | None = None,
    **kwargs: Any,
) -> None:
    """
    Initialize the PerformanceModel.

    Parameters
    ----------
    model_class : type[AbstractPredictor], default=RandomForestRegressorWrapper
        The class of the regression model to be used.
    use_multi_target : bool, default=False
        Indicates whether to use multi-target regression.
    normalize : AbstractNormalization or None, default=None
        Method to normalize performance data. If None, defaults to LogNormalization().
    init_params : dict[str, Any] or None, default=None
        Initialization parameters from configuration.
    **kwargs : Any
        Additional arguments for the parent classes.
    """
    params = init_params if isinstance(init_params, dict) else {}
    params.update(kwargs)

    model_class = params.pop("model_class", model_class)
    use_multi_target = params.pop("use_multi_target", use_multi_target)
    normalize = params.pop("normalize", normalize)

    AbstractModelBasedSelector.__init__(self, model_class, **params)
    AbstractFeatureGenerator.__init__(self)

    self.regressors: list[AbstractPredictor] | AbstractPredictor | None = None
    self.use_multi_target = bool(use_multi_target)

    # Instantiate normalize if it is a partial or class
    if callable(normalize):
        self.normalize = normalize()
    elif normalize is not None:
        self.normalize = normalize
    else:
        self.normalize = LogNormalization()

generate_features(base_features)

Generate predictions for each algorithm.

Parameters

features : pd.DataFrame The input features.

Returns

np.ndarray Predicted performance for each algorithm (n_instances x n_algorithms).

Source code in asf/selectors/performance_model.py

def generate_features(self, base_features: pd.DataFrame) -> pd.DataFrame:
    """
    Generate predictions for each algorithm.

    Parameters
    ----------
    features : pd.DataFrame
        The input features.

    Returns
    -------
    np.ndarray
        Predicted performance for each algorithm (n_instances x n_algorithms).
    """
    if self.regressors is None:
        raise RuntimeError("Model has not been fitted.")

    predictions = np.zeros((base_features.shape[0], len(self.algorithms)))

    if self.use_multi_target:
        if not isinstance(self.regressors, AbstractPredictor):
            raise RuntimeError("Multi-target regressor missing.")
        predictions = self.regressors.predict(base_features)
        if isinstance(predictions, pd.DataFrame):
            predictions = predictions.values
    else:
        if self.algorithm_features is None:
            if not isinstance(self.regressors, list):
                raise RuntimeError("Individual regressors missing.")
            for i, _ in enumerate(self.algorithms):
                regressor = cast(AbstractPredictor, self.regressors[i])
                predictions[:, i] = np.asarray(
                    regressor.predict(base_features)
                ).flatten()
        else:
            if not isinstance(self.regressors, AbstractPredictor):
                raise RuntimeError("Joint regressor missing.")
            regressor: AbstractPredictor = self.regressors
            for i, algorithm in enumerate(self.algorithms):
                data = pd.merge(
                    base_features,
                    self.algorithm_features.loc[[algorithm]]
                    .reindex([algorithm] * len(base_features))
                    .set_index(base_features.index),
                    left_index=True,
                    right_index=True,
                )
                predictions[:, i] = regressor.predict(data)

    return pd.DataFrame(
        predictions,
        index=base_features.index,
        columns=pd.Index(list(self.algorithms)),
    )

RPCSelector

Bases: AbstractSelector

Ranking by Pairwise Comparison (RPC) for Algorithm Selection.

RPC decomposes the k-label ranking problem into m = k(k-1)/2 binary classification tasks (one for each pair of algorithms).

References

Hüllermeier, E., et al. (2008). "Label Ranking by Learning Pairwise Preferences." https://en.cs.uni-paderborn.de/fileadmin-eim/informatik/fg/is/Publications/mpub109.pdf

Source code in asf/selectors/rpc_selector.py

class RPCSelector(AbstractSelector):
    """
    Ranking by Pairwise Comparison (RPC) for Algorithm Selection.

    RPC decomposes the k-label ranking problem into m = k(k-1)/2 binary
    classification tasks (one for each pair of algorithms).

    References
    ----------
    Hüllermeier, E., et al. (2008).
    "Label Ranking by Learning Pairwise Preferences."
    https://en.cs.uni-paderborn.de/fileadmin-eim/informatik/fg/is/Publications/mpub109.pdf
    """

    PREFIX = "rpc"
    RETURN_TYPE = "single"

    def __init__(
        self,
        classifier_class: Type = RandomForestClassifierWrapper,
        n_estimators: int = 100,
        classifier_kwargs: dict[str, Any] | None = None,
        random_state: int = 42,
        top_n: int = 1,
        **kwargs: Any,
    ):
        """
        Initialize RPC selector.

        Args:
            classifier_class: The classifier class to use for pairwise tasks.
            n_estimators: Number of estimators for ensemble classifiers.
            classifier_kwargs: Additional kwargs to pass to each classifier.
            random_state: Random seed for reproducibility.
            top_n: Number of top algorithms to return. If > 1, selector returns
                    parallel portfolios (list of algorithm names). If 1, returns
                    a single (algorithm, budget) tuple.
            **kwargs: Additional arguments for parent class.
        """
        super().__init__(**kwargs)
        self.classifier_class = classifier_class
        self.n_estimators = int(n_estimators)
        self.classifier_kwargs = dict(classifier_kwargs or {})

        # Only set defaults if the classifier accepts these parameters
        sig_params = set(inspect.signature(classifier_class).parameters.keys())
        if (
            "n_estimators" in sig_params
            and "n_estimators" not in self.classifier_kwargs
        ):
            self.classifier_kwargs["n_estimators"] = int(n_estimators)

        if (
            "random_state" in sig_params
            and "random_state" not in self.classifier_kwargs
        ):
            self.classifier_kwargs["random_state"] = int(random_state)

        self.random_state = int(random_state)
        self.top_n = max(1, int(top_n))

        # Set return type dynamically based on top_n
        self.RETURN_TYPE = "parallel" if self.top_n > 1 else "single"

        self.classifiers: Dict[Tuple[str, str], Any] = {}
        self.algorithms: List[str] = []
        self.pairs: List[Tuple[str, str]] = []

    def _create_pairs(self, algorithms: List[str]) -> List[Tuple[str, str]]:
        """Create all unique pairs of algorithms (i, j) where i < j."""
        pairs = []
        for i in range(len(algorithms)):
            for j in range(i + 1, len(algorithms)):
                pairs.append((algorithms[i], algorithms[j]))
        return pairs

    def _create_binary_labels(
        self, performance: pd.DataFrame, algo_i: str, algo_j: str
    ) -> np.ndarray:
        """
        Create binary labels for pairwise comparison.

        Args:
            performance: Performance matrix (n_instances, n_algorithms)
            algo_i: First algorithm
            algo_j: Second algorithm

        Returns:
            Binary labels: 1 if algo_i < algo_j (better), 0 otherwise
        """
        perf_i = performance[algo_i].values
        perf_j = performance[algo_j].values
        # 1 if algo_i is better (lower runtime), 0 otherwise
        return (perf_i < perf_j).astype(int)

    def _fit(self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs) -> None:
        """
        Train binary classifiers for all algorithm pairs.

        Args:
            features: Instance features (n_instances, n_features)
            performance: Algorithm performance (n_instances, n_algorithms)
        """
        self.algorithms = list(performance.columns)
        self.pairs = self._create_pairs(self.algorithms)

        X = features.values

        for algo_i, algo_j in self.pairs:
            y_ij = self._create_binary_labels(performance, algo_i, algo_j)

            clf = self.classifier_class(**self.classifier_kwargs)
            clf.fit(X, y_ij)

            self.classifiers[(algo_i, algo_j)] = clf

    def _compute_copeland_scores(self, features: pd.DataFrame) -> np.ndarray:
        """
        Compute Copeland scores for each instance.

        Args:
            features: Instance features (n_instances, n_features)

        Returns:
            Array of shape (n_instances, n_algorithms) with Copeland scores
        """
        n_instances = len(features)
        n_algorithms = len(self.algorithms)

        scores = np.zeros((n_instances, n_algorithms))
        X = features.values

        for algo_i, algo_j in self.pairs:
            i_idx = self.algorithms.index(algo_i)
            j_idx = self.algorithms.index(algo_j)

            clf = self.classifiers[(algo_i, algo_j)]
            predictions = clf.predict(X)  # 1 if algo_i better, 0 if algo_j better

            scores[predictions == 1, i_idx] += 1
            scores[predictions == 0, j_idx] += 1

        return scores

    def _predict(
        self, features: pd.DataFrame | None, performance: pd.DataFrame | None = None
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict algorithm(s) for each instance using Copeland scores.

        Args:
            features: Instance features

        Returns:
            If top_n == 1: dict[str, list[tuple[str, float]]] with single (algo, budget).
            If top_n > 1: dict[str, list[tuple[str, float]]] with equal time slices.
        """
        if not self.classifiers:
            raise RuntimeError("The selector has not been fitted yet.")

        budget = getattr(self, "budget", None)
        if budget is None:
            budget = float("inf")

        assert isinstance(features, pd.DataFrame)
        scores = self._compute_copeland_scores(features)
        predictions = {}

        for inst_idx, inst_name in enumerate(features.index):
            inst_scores = scores[inst_idx, :]
            order = np.argsort(-inst_scores)
            top_indices = order[: self.top_n]
            top_algos = [self.algorithms[i] for i in top_indices]

            if self.top_n == 1:
                predictions[inst_name] = [(top_algos[0], float(budget))]
            else:
                time_slice = float(budget) / len(top_algos)
                predictions[inst_name] = [(algo, time_slice) for algo in top_algos]

        return predictions

__init__(classifier_class=RandomForestClassifierWrapper, n_estimators=100, classifier_kwargs=None, random_state=42, top_n=1, **kwargs)

Initialize RPC selector.

Parameters:

Name	Type	Description	Default
classifier_class	Type	The classifier class to use for pairwise tasks.	RandomForestClassifierWrapper
n_estimators	int	Number of estimators for ensemble classifiers.	100
classifier_kwargs	dict[str, Any] | None	Additional kwargs to pass to each classifier.	None
random_state	int	Random seed for reproducibility.	42
top_n	int	Number of top algorithms to return. If > 1, selector returns parallel portfolios (list of algorithm names). If 1, returns a single (algorithm, budget) tuple.	1
**kwargs	Any	Additional arguments for parent class.	{}

Source code in asf/selectors/rpc_selector.py

def __init__(
    self,
    classifier_class: Type = RandomForestClassifierWrapper,
    n_estimators: int = 100,
    classifier_kwargs: dict[str, Any] | None = None,
    random_state: int = 42,
    top_n: int = 1,
    **kwargs: Any,
):
    """
    Initialize RPC selector.

    Args:
        classifier_class: The classifier class to use for pairwise tasks.
        n_estimators: Number of estimators for ensemble classifiers.
        classifier_kwargs: Additional kwargs to pass to each classifier.
        random_state: Random seed for reproducibility.
        top_n: Number of top algorithms to return. If > 1, selector returns
                parallel portfolios (list of algorithm names). If 1, returns
                a single (algorithm, budget) tuple.
        **kwargs: Additional arguments for parent class.
    """
    super().__init__(**kwargs)
    self.classifier_class = classifier_class
    self.n_estimators = int(n_estimators)
    self.classifier_kwargs = dict(classifier_kwargs or {})

    # Only set defaults if the classifier accepts these parameters
    sig_params = set(inspect.signature(classifier_class).parameters.keys())
    if (
        "n_estimators" in sig_params
        and "n_estimators" not in self.classifier_kwargs
    ):
        self.classifier_kwargs["n_estimators"] = int(n_estimators)

    if (
        "random_state" in sig_params
        and "random_state" not in self.classifier_kwargs
    ):
        self.classifier_kwargs["random_state"] = int(random_state)

    self.random_state = int(random_state)
    self.top_n = max(1, int(top_n))

    # Set return type dynamically based on top_n
    self.RETURN_TYPE = "parallel" if self.top_n > 1 else "single"

    self.classifiers: Dict[Tuple[str, str], Any] = {}
    self.algorithms: List[str] = []
    self.pairs: List[Tuple[str, str]] = []

SATzilla

Bases: ConfigurableMixin, AbstractEPMBasedSelector, AbstractModelBasedSelector

SATzilla-like selector using iterative imputation for censored runtimes.

Uses per-algorithm ridge models on expanded features with optional instance label conditioning.

References

Xu, L., et al. (2008). "SATzilla: Portfolio-based Algorithm Selection for SAT." https://www.jair.org/index.php/jair/article/view/10556

Attributes

epms : dict[str, dict[str, EPM]] Mapping from algorithm name to another mapping of label to EPM. label_classifier_model : type or Callable The classifier class used for the label classifier (only used when labels are provided). epm_regressor_model : type or Callable The regressor class used internally by EPM for per-algorithm performance prediction.

Source code in asf/selectors/satzilla.py

class SATzilla(ConfigurableMixin, AbstractEPMBasedSelector, AbstractModelBasedSelector):
    """
    SATzilla-like selector using iterative imputation for censored runtimes.

    Uses per-algorithm ridge models on expanded features with optional instance label conditioning.

    References
    ----------
    Xu, L., et al. (2008).
    "SATzilla: Portfolio-based Algorithm Selection for SAT."
    https://www.jair.org/index.php/jair/article/view/10556

    Attributes
    ----------
    epms : dict[str, dict[str, EPM]]
        Mapping from algorithm name to another mapping of label to EPM.
    label_classifier_model : type or Callable
        The classifier class used for the label classifier (only used when labels are provided).
    epm_regressor_model : type or Callable
        The regressor class used internally by EPM for per-algorithm performance prediction.
    """

    PREFIX = "satzilla"
    RETURN_TYPE = "single"

    def __init__(
        self,
        label_classifier_model: type[Any] = RandomForestClassifierWrapper,
        epm_regressor_model: type[Any] = RidgeRegressorWrapper,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the SATzilla selector.

        Parameters
        ----------
        label_classifier_model : type, default=RandomForestClassifierWrapper
            The classifier class for predicting instance labels (e.g., SAT/UNSAT).
        epm_regressor_model : type, default=RidgeRegressorWrapper
            The regressor class used internally by EPM for per-algorithm runtime prediction.
        **kwargs : Any
            Additional keyword arguments passed to parent classes.
        """
        super().__init__(model_class=label_classifier_model, **kwargs)
        self.epm_regressor_model = epm_regressor_model
        self.epm_kwargs["predictor_class"] = epm_regressor_model
        self.epms: dict[str, dict[str, EPM]] = {}
        self.label_classifier: Any = None
        self.labels: list[str] = []

    def _fit(
        self,
        features: pd.DataFrame,
        performance: pd.DataFrame,
        labels: pd.DataFrame | pd.Series | list[str] | np.ndarray | None = None,
        **kwargs: Any,
    ) -> None:
        """
        Fit per-algorithm models.

        Parameters
        ----------
        features : pd.DataFrame
            Training features (instances x features).
        performance : pd.DataFrame
            Training performance matrix (instances x algorithms).
        labels : pd.DataFrame, pd.Series, list, or np.ndarray, optional
            Optional labels for training conditioned EPMs.
        """
        if labels is None:
            labels_series = pd.Series(["default"] * len(features), index=features.index)
            self.label_classifier = None
            self.labels = ["default"]
        else:
            if isinstance(labels, pd.DataFrame):
                labels_series = labels.squeeze(axis=1)
            elif isinstance(labels, pd.Series):
                labels_series = labels
            else:
                labels_series = pd.Series(labels, index=features.index)

            assert isinstance(labels_series, pd.Series)
            if not labels_series.index.equals(features.index):
                labels_series = labels_series.reindex(features.index)

            self.label_classifier = self.model_class()
            self.label_classifier.fit(features.values, labels_series.values)

            # Extract unique labels
            if hasattr(self.label_classifier, "model_class") and hasattr(
                self.label_classifier.model_class, "classes_"
            ):
                self.labels = [
                    str(c) for c in self.label_classifier.model_class.classes_
                ]
            else:
                self.labels = [str(c) for c in np.unique(labels_series.values)]

        for algo in self.algorithms:
            self.epms[str(algo)] = {}
            for label in self.labels:
                idx = labels_series.astype(str) == str(label)
                if idx.sum() == 0:
                    continue
                self.epms[str(algo)][str(label)] = EPM(**self.epm_kwargs)
                X_sub = features.loc[idx]
                y_sub = performance.loc[idx, algo]
                self.epms[str(algo)][str(label)].fit(X_sub, y_sub)

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict the best algorithm for each instance.

        Parameters
        ----------
        features : pd.DataFrame or None
            The input features.
        performance : pd.DataFrame or None, default=None
            Partial performance data.

        Returns
        -------
        dict
            Mapping from instance name to algorithm schedules.
        """
        if features is None:
            raise ValueError("SATzilla requires features for prediction.")
        n_instances = features.shape[0]
        n_algorithms = len(self.algorithms)
        preds = np.zeros((n_instances, n_algorithms), dtype=float)

        if self.label_classifier is None:
            label_probs = np.ones((n_instances, 1), dtype=float)
        elif hasattr(self.label_classifier, "model_class") and hasattr(
            self.label_classifier.model_class, "predict_proba"
        ):
            label_probs = self.label_classifier.model_class.predict_proba(
                features.values
            )
        else:
            hard_preds = np.asarray(self.label_classifier.predict(features.values))
            classes = np.asarray(self.labels)
            label_probs = (hard_preds[:, None] == classes[None, :]).astype(float)

        for j, algo in enumerate(self.algorithms):
            for k, label in enumerate(self.labels):
                if str(algo) not in self.epms or str(label) not in self.epms[str(algo)]:
                    continue
                pred_time = np.asarray(
                    self.epms[str(algo)][str(label)].predict(features)
                )
                preds[:, j] += label_probs[:, k] * pred_time

        best_idx = np.argmin(preds, axis=1)
        results: dict[str, list[tuple[str, float]]] = {}
        for i, inst in enumerate(features.index):
            j = int(best_idx[i])
            algo = str(self.algorithms[j])
            results[str(inst)] = [(algo, float(self.budget or 0))]
        return results

    @staticmethod
    def _define_hyperparameters(
        label_classifier_model: list[type] | None = None,
        epm_regressor_model: list[type] | None = None,
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for SATzilla.

        Parameters
        ----------
        label_classifier_model : list[type] or None, default=None
            List of classifier classes to choose from for the label classifier.
        epm_regressor_model : list[type] or None, default=None
            List of regressor classes to choose from for the EPM's internal regressor.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        if label_classifier_model is None:
            label_classifier_model = [RandomForestClassifierWrapper]

        if epm_regressor_model is None:
            epm_regressor_model = [RidgeRegressorWrapper]

        label_classifier_model_param = ClassChoice(
            name="label_classifier_model",
            choices=cast(list[type | bool], label_classifier_model),
            default=label_classifier_model[0],
        )

        epm_regressor_model_param = ClassChoice(
            name="epm_regressor_model",
            choices=cast(list[type | bool], epm_regressor_model),
            default=epm_regressor_model[0],
        )

        params = [
            label_classifier_model_param,
            epm_regressor_model_param,
        ]

        return params, [], []

__init__(label_classifier_model=RandomForestClassifierWrapper, epm_regressor_model=RidgeRegressorWrapper, **kwargs)

Initialize the SATzilla selector.

Parameters

label_classifier_model : type, default=RandomForestClassifierWrapper The classifier class for predicting instance labels (e.g., SAT/UNSAT). epm_regressor_model : type, default=RidgeRegressorWrapper The regressor class used internally by EPM for per-algorithm runtime prediction. **kwargs : Any Additional keyword arguments passed to parent classes.

Source code in asf/selectors/satzilla.py

def __init__(
    self,
    label_classifier_model: type[Any] = RandomForestClassifierWrapper,
    epm_regressor_model: type[Any] = RidgeRegressorWrapper,
    **kwargs: Any,
) -> None:
    """
    Initialize the SATzilla selector.

    Parameters
    ----------
    label_classifier_model : type, default=RandomForestClassifierWrapper
        The classifier class for predicting instance labels (e.g., SAT/UNSAT).
    epm_regressor_model : type, default=RidgeRegressorWrapper
        The regressor class used internally by EPM for per-algorithm runtime prediction.
    **kwargs : Any
        Additional keyword arguments passed to parent classes.
    """
    super().__init__(model_class=label_classifier_model, **kwargs)
    self.epm_regressor_model = epm_regressor_model
    self.epm_kwargs["predictor_class"] = epm_regressor_model
    self.epms: dict[str, dict[str, EPM]] = {}
    self.label_classifier: Any = None
    self.labels: list[str] = []

SNNAP

Bases: ConfigurableMixin, AbstractSelector

SNNAP (Solver-based Nearest Neighbor for Algorithm Portfolio) selector.

Uses per-algorithm performance prediction models and Jaccard distance on predicted top algorithms to find similar instances, then selects from neighbors' best algorithms.

References

Collautti, M., et al. (2013). "A Solver-Based Nearest Neighbor Method for Portfolio Selection." https://link.springer.com/chapter/10.1007/978-3-642-40994-3_28

Attributes

k : int Number of similar instances (neighbors) to use. top_n : int Number of top algorithms to consider for Jaccard distance calculation. algorithm_models : dict[str, AbstractPredictor] or None Per-algorithm regression models for predicting scaled runtime. scaled_performance_df : pd.DataFrame or None Z-score normalized performance matrix (per instance). original_performance_df : pd.DataFrame or None Original unnormalized performance data. training_top_n_sets : list[set[str]] or None Pre-computed top-n algorithm sets for each training instance. features_df : pd.DataFrame or None Training features.

Source code in asf/selectors/snnap.py

class SNNAP(ConfigurableMixin, AbstractSelector):
    """
    SNNAP (Solver-based Nearest Neighbor for Algorithm Portfolio) selector.

    Uses per-algorithm performance prediction models and Jaccard distance on predicted
    top algorithms to find similar instances, then selects from neighbors' best algorithms.

    References
    ----------
    Collautti, M., et al. (2013).
    "A Solver-Based Nearest Neighbor Method for Portfolio Selection."
    https://link.springer.com/chapter/10.1007/978-3-642-40994-3_28

    Attributes
    ----------
    k : int
        Number of similar instances (neighbors) to use.
    top_n : int
        Number of top algorithms to consider for Jaccard distance calculation.
    algorithm_models : dict[str, AbstractPredictor] or None
        Per-algorithm regression models for predicting scaled runtime.
    scaled_performance_df : pd.DataFrame or None
        Z-score normalized performance matrix (per instance).
    original_performance_df : pd.DataFrame or None
        Original unnormalized performance data.
    training_top_n_sets : list[set[str]] or None
        Pre-computed top-n algorithm sets for each training instance.
    features_df : pd.DataFrame or None
        Training features.
    """

    PREFIX = "snnap"
    RETURN_TYPE = "single"

    def __init__(
        self,
        k: int = 5,
        top_n: int = 3,
        random_state: int | None = 42,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the SNNAP selector.

        Parameters
        ----------
        k : int, default=5
            Number of nearest neighbors (similar instances) to consider.
        top_n : int, default=3
            Number of top algorithms to use for Jaccard distance calculation.
        random_state : int or None, default=42
            Random seed for the RandomForest models.
        **kwargs : Any
            Additional keyword arguments.
        """
        super().__init__(**kwargs)
        self.k = int(k)
        self.top_n = int(top_n)
        self.random_state = random_state

        self.features_df: pd.DataFrame | None = None
        self.original_performance_df: pd.DataFrame | None = None
        self.scaled_performance_df: pd.DataFrame | None = None
        self.algorithm_models: dict[str, Any] | None = None
        self.training_top_n_sets: list[set[str]] | None = None

    def _fit(
        self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs: Any
    ) -> None:
        """
        Fit per-algorithm regression models on z-score normalized performance.

        Parameters
        ----------
        features : pd.DataFrame
            The training features (instances x features).
        performance : pd.DataFrame
            The training performance data (instances x algorithms).
        """
        self.features_df = features.copy()
        self.original_performance_df = performance[self.algorithms].copy()

        # Step 1: Z-score normalize performance per instance (row-wise)
        self.scaled_performance_df = self.original_performance_df.copy().astype(float)
        for idx in self.scaled_performance_df.index:
            row = self.scaled_performance_df.loc[idx]
            mean_val = row.mean()
            std_val = row.std()
            if std_val > 0:
                self.scaled_performance_df.loc[idx] = (row - mean_val) / std_val
            else:
                # If all values are the same, set to 0
                self.scaled_performance_df.loc[idx] = 0

        # Step 2: Train one regressor per algorithm
        self.algorithm_models = {}
        for algo in self.algorithms:
            y = self.scaled_performance_df[algo]
            model = RandomForestRegressorWrapper(random_state=self.random_state)
            model.fit(features.values, y.values)
            self.algorithm_models[str(algo)] = model

        # Step 3: Pre-compute top-n sets for each training instance
        self.training_top_n_sets = []
        for train_idx in range(len(self.scaled_performance_df)):
            train_scaled_perf = self.scaled_performance_df.iloc[train_idx]
            # For minimize: lowest scaled runtime is best. For maximize: highest is best.
            ascending = not self.maximize
            train_sorted = train_scaled_perf.sort_values(ascending=ascending)
            train_top_n = set(str(algo) for algo in train_sorted.index[: self.top_n])
            self.training_top_n_sets.append(train_top_n)

    @staticmethod
    def _jaccard_distance(set1: set[str], set2: set[str]) -> float:
        """
        Compute Jaccard distance between two sets.

        Parameters
        ----------
        set1 : set[str]
            First set of algorithm names.
        set2 : set[str]
            Second set of algorithm names.

        Returns
        -------
        float
            Jaccard distance: 1 - |intersection| / |union|
        """
        if not set1 or not set2:
            return 1.0
        intersection = len(set1 & set2)
        union = len(set1 | set2)
        if union == 0:
            return 1.0
        return 1.0 - (intersection / union)

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict the best algorithm using Jaccard distance on top-n algorithms.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame, optional
            Unused, kept for interface compatibility.

        Returns
        -------
        dict
            Mapping from instance name to algorithm schedules.
        """
        if features is None:
            raise ValueError("SNNAP requires features for prediction.")
        if (
            self.algorithm_models is None
            or self.features_df is None
            or self.original_performance_df is None
            or self.scaled_performance_df is None
            or self.training_top_n_sets is None
        ):
            raise RuntimeError("SNNAP must be fitted before prediction.")

        predictions: dict[str, list[tuple[str, float]]] = {}

        for instance_name in features.index:
            # Step 1: Use per-algorithm models to predict scaled runtimes for query instance
            query_features = features.loc[[instance_name]]
            predicted_scaled_runtimes = {}
            for algo in self.algorithms:
                model = self.algorithm_models[str(algo)]
                pred = model.predict(query_features.values)[0]
                predicted_scaled_runtimes[algo] = pred

            # Step 2: Identify query's predicted top-n algorithms (respecting maximize flag)
            sorted_algos = sorted(
                predicted_scaled_runtimes.items(),
                key=lambda x: x[1],
                reverse=self.maximize,  # If maximize, sort descending; if minimize, ascending
            )
            query_top_n = set(str(algo) for algo, _ in sorted_algos[: self.top_n])

            # Step 3: Compute Jaccard distance for each training instance
            jaccard_distances = []
            for train_idx in range(len(self.features_df)):
                # Use pre-computed top-n set
                train_top_n = self.training_top_n_sets[train_idx]

                # Compute Jaccard distance
                jdist = self._jaccard_distance(query_top_n, train_top_n)
                jaccard_distances.append((train_idx, jdist))

            # Step 4: Select k neighbors with lowest Jaccard distance
            jaccard_distances.sort(key=lambda x: x[1])
            k_actual = min(self.k, len(self.features_df))
            neighbor_idxs = [idx for idx, _ in jaccard_distances[:k_actual]]

            # Step 5: Compute mean runtime for ALL algorithms across neighbors, then select best
            # This differs from selecting best-per-neighbor and averaging only winners
            runtimes_by_algo: dict[str, list[float]] = {}
            for ni in neighbor_idxs:
                neighbor_perf = self.original_performance_df.iloc[ni]
                # For each algorithm, collect its actual runtime in this neighbor
                for algo in self.algorithms:
                    if algo in neighbor_perf.index and pd.notna(neighbor_perf[algo]):
                        if algo not in runtimes_by_algo:
                            runtimes_by_algo[algo] = []
                        runtimes_by_algo[algo].append(float(neighbor_perf[algo]))

            if not runtimes_by_algo:
                predictions[str(instance_name)] = []
                continue

            # Compute mean runtime for all algorithms
            mean_runtimes = {
                algo: float(np.mean(times)) for algo, times in runtimes_by_algo.items()
            }
            # Select algorithm with best mean runtime
            if self.maximize:
                chosen = max(mean_runtimes.items(), key=lambda x: x[1])[0]
            else:
                chosen = min(mean_runtimes.items(), key=lambda x: x[1])[0]

            predictions[str(instance_name)] = [(str(chosen), float(self.budget or 0))]

        return predictions

    @staticmethod
    def _define_hyperparameters(
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for SNNAP.

        Parameters
        ----------
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        k_param = Integer(
            name="k",
            bounds=(1, 50),
            default=5,
        )

        top_n_param = Integer(
            name="top_n",
            bounds=(1, 20),
            default=3,
        )

        random_state_param = Integer(
            name="random_state",
            bounds=(0, 2**31 - 1),
            default=42,
        )

        return [k_param, top_n_param, random_state_param], [], []

__init__(k=5, top_n=3, random_state=42, **kwargs)

Initialize the SNNAP selector.

Parameters

k : int, default=5 Number of nearest neighbors (similar instances) to consider. top_n : int, default=3 Number of top algorithms to use for Jaccard distance calculation. random_state : int or None, default=42 Random seed for the RandomForest models. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/snnap.py

def __init__(
    self,
    k: int = 5,
    top_n: int = 3,
    random_state: int | None = 42,
    **kwargs: Any,
) -> None:
    """
    Initialize the SNNAP selector.

    Parameters
    ----------
    k : int, default=5
        Number of nearest neighbors (similar instances) to consider.
    top_n : int, default=3
        Number of top algorithms to use for Jaccard distance calculation.
    random_state : int or None, default=42
        Random seed for the RandomForest models.
    **kwargs : Any
        Additional keyword arguments.
    """
    super().__init__(**kwargs)
    self.k = int(k)
    self.top_n = int(top_n)
    self.random_state = random_state

    self.features_df: pd.DataFrame | None = None
    self.original_performance_df: pd.DataFrame | None = None
    self.scaled_performance_df: pd.DataFrame | None = None
    self.algorithm_models: dict[str, Any] | None = None
    self.training_top_n_sets: list[set[str]] | None = None

SUNNY

Bases: ConfigurableMixin, AbstractSelector

SUNNY/SUNNY-AS2 algorithm selector.

This selector uses k-nearest neighbors (k-NN) in feature space to construct a schedule. When SUNNY-AS2 is enabled, k is optimized.

References

Liu, Z., et al. (2022). "SUNNY-AS2: Enhancing SUNNY for Algorithm Selection." https://dl.acm.org/doi/10.1613/jair.1.13116

Attributes

k : int Number of neighbors for k-NN. use_v2 : bool Whether to tune k using cross-validation. n_folds : int Number of folds for cross-validation when tuning. k_candidates : list[int] Candidate k values for tuning. random_state : int Random seed for reproducibility. use_tsunny : bool Whether to tune the maximum number of algorithms. algorithm_limit : int or None Manual cap on the number of algorithms in each schedule. tuned_algorithm_limit : int or None Tuned cap on the number of algorithms. features_df : pd.DataFrame or None Training features. performance_df : pd.DataFrame or None Training performance. knn : NearestNeighbors or None Trained k-NN model.

Source code in asf/selectors/sunny.py

class SUNNY(ConfigurableMixin, AbstractSelector):
    """
    SUNNY/SUNNY-AS2 algorithm selector.

    This selector uses k-nearest neighbors (k-NN) in feature space to construct
    a schedule. When SUNNY-AS2 is enabled, k is optimized.

    References
    ----------
    Liu, Z., et al. (2022).
    "SUNNY-AS2: Enhancing SUNNY for Algorithm Selection."
    https://dl.acm.org/doi/10.1613/jair.1.13116

    Attributes
    ----------
    k : int
        Number of neighbors for k-NN.
    use_v2 : bool
        Whether to tune k using cross-validation.
    n_folds : int
        Number of folds for cross-validation when tuning.
    k_candidates : list[int]
        Candidate k values for tuning.
    random_state : int
        Random seed for reproducibility.
    use_tsunny : bool
        Whether to tune the maximum number of algorithms.
    algorithm_limit : int or None
        Manual cap on the number of algorithms in each schedule.
    tuned_algorithm_limit : int or None
        Tuned cap on the number of algorithms.
    features_df : pd.DataFrame or None
        Training features.
    performance_df : pd.DataFrame or None
        Training performance.
    knn : NearestNeighbors or None
        Trained k-NN model.
    """

    PREFIX = "sunny"
    RETURN_TYPE = "schedule"

    def __init__(
        self,
        k: int = 10,
        use_v2: bool = False,
        n_folds: int = 5,
        k_candidates: list[int] | None = None,
        random_state: int = 42,
        use_tsunny: bool = False,
        algorithm_limit: int | None = None,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the SUNNY selector.

        Parameters
        ----------
        k : int, default=10
            Number of neighbors for k-NN.
        use_v2 : bool, default=False
            Whether to tune k using cross-validation (SUNNY-AS2).
        n_folds : int, default=5
            Number of folds for cross-validation when tuning.
        k_candidates : list[int] or None, default=None
            Candidate k values to consider when tuning.
        random_state : int, default=42
            Random seed for reproducibility.
        use_tsunny : bool, default=False
            Whether to tune the max number of algorithms via cross-validation.
        algorithm_limit : int or None, default=None
            If set, cap the number of algorithms in each schedule.
        **kwargs : Any
            Additional keyword arguments.
        """
        super().__init__(**kwargs)
        self.k = int(k)
        self.use_v2 = bool(use_v2)
        self.n_folds = int(n_folds)

        if isinstance(k_candidates, str):
            k_map = {
                "small": [3, 5, 7],
                "medium": [3, 5, 7, 10, 20],
                "broad": [3, 5, 7, 10, 20, 50],
            }
            self.k_candidates = k_map.get(k_candidates, [3, 5, 7, 10, 20, 50])
        else:
            self.k_candidates = k_candidates or [3, 5, 7, 10, 20, 50]

        self.random_state = int(random_state)
        self.use_tsunny = bool(use_tsunny)
        self.algorithm_limit = algorithm_limit
        self.tuned_algorithm_limit: int | None = None

        self.features_df: pd.DataFrame | None = None
        self.performance_df: pd.DataFrame | None = None
        self.knn: NearestNeighbors | None = None

    def _fit(
        self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs: Any
    ) -> None:
        """
        Fit the SUNNY selector.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame
            The algorithm performance data.
        """
        self.features_df = features.copy()
        perf = performance.copy()
        budget = float(self.budget or 1e10)
        perf[perf > budget] = np.nan
        self.performance_df = perf

        if self.use_v2:
            self.k = self._tune_k()

        if self.use_tsunny and self.algorithm_limit is None:
            self.tuned_algorithm_limit = self._tune_algorithm_limit()

        self.knn = NearestNeighbors(
            n_neighbors=min(self.k, len(self.features_df)), metric="euclidean"
        )
        self.knn.fit(self.features_df.values)

    def _tune_k(self) -> int:
        """
        Tune the neighborhood size k via cross-validation.

        Returns
        -------
        int
            The best k value found.
        """
        if self.features_df is None or self.performance_df is None:
            return self.k

        best_k = self.k
        best_score = float("inf")
        n_splits = min(self.n_folds, max(2, len(self.features_df)))

        if n_splits < 2:
            return best_k

        kf = KFold(n_splits=n_splits, shuffle=True, random_state=self.random_state)
        indices = np.arange(len(self.features_df))

        for candidate_k in self.k_candidates:
            fold_scores = []
            for train_idx, val_idx in kf.split(indices):
                train_feat = self.features_df.iloc[train_idx]
                train_perf = self.performance_df.iloc[train_idx]
                val_feat = self.features_df.iloc[val_idx]
                val_perf = self.performance_df.iloc[val_idx]

                if len(train_feat) == 0:
                    continue

                knn = NearestNeighbors(
                    n_neighbors=min(candidate_k, len(train_feat)),
                    metric="euclidean",
                )
                knn.fit(train_feat.values)

                total_cost = 0.0
                for instance in val_feat.index:
                    x = val_feat.loc[[instance]].values
                    _, n_idx = knn.kneighbors(
                        x, n_neighbors=min(candidate_k, len(train_feat))
                    )
                    n_perf = train_perf.iloc[n_idx.flatten()]
                    schedule = self._construct_sunny_schedule(n_perf)

                    inst_perf = val_perf.loc[instance]
                    solved = False
                    for algo, _ in schedule:
                        runtime = inst_perf[algo]
                        if not pd.isna(runtime) and runtime <= float(
                            self.budget or 1e10
                        ):
                            total_cost += float(runtime)
                            solved = True
                            break
                    if not solved:
                        total_cost += float(self.budget or 1e10)

                fold_scores.append(total_cost / len(val_feat))

            mean_score = float(np.mean(fold_scores)) if fold_scores else float("inf")
            if mean_score < best_score:
                best_score = mean_score
                best_k = int(candidate_k)

        return best_k

    def _tune_algorithm_limit(self) -> int:
        """
        Tune the maximum number of algorithms via cross-validation.

        Returns
        -------
        int
            Best limit found.
        """
        if self.features_df is None or self.performance_df is None:
            return len(self.algorithms)

        n_solvers = len(self.performance_df.columns)
        if n_solvers <= 1:
            return n_solvers

        n_splits = min(self.n_folds, max(2, len(self.features_df)))
        if n_splits < 2:
            return n_solvers

        best_lam = n_solvers
        best_score = float("inf")
        kf = KFold(n_splits=n_splits, shuffle=True, random_state=self.random_state)
        indices = np.arange(len(self.features_df))

        for lam in range(1, n_solvers + 1):
            fold_scores = []
            for train_idx, val_idx in kf.split(indices):
                train_feat = self.features_df.iloc[train_idx]
                train_perf = self.performance_df.iloc[train_idx]
                val_feat = self.features_df.iloc[val_idx]
                val_perf = self.performance_df.iloc[val_idx]

                knn = NearestNeighbors(
                    n_neighbors=min(self.k, len(train_feat)), metric="euclidean"
                )
                knn.fit(train_feat.values)

                total_cost = 0.0
                for instance in val_feat.index:
                    x = val_feat.loc[[instance]].values
                    _, n_idx = knn.kneighbors(
                        x, n_neighbors=min(self.k, len(train_feat))
                    )
                    n_perf = train_perf.iloc[n_idx.flatten()]
                    schedule = self._construct_sunny_schedule(n_perf, lam_limit=lam)

                    inst_perf = val_perf.loc[instance]
                    solved = False
                    for algo, _ in schedule:
                        runtime = inst_perf[algo]
                        if not pd.isna(runtime) and runtime <= float(
                            self.budget or 1e10
                        ):
                            total_cost += float(runtime)
                            solved = True
                            break
                    if not solved:
                        total_cost += float(self.budget or 1e10)
                fold_scores.append(total_cost / len(val_feat))

            mean_score = float(np.mean(fold_scores)) if fold_scores else float("inf")
            if mean_score < best_score:
                best_score = mean_score
                best_lam = lam

        return best_lam

    def _mine_solvers(
        self,
        neighbor_perf: pd.DataFrame,
        cutoff: int,
        already_selected: list[str] | None = None,
        already_covered: set[str] | None = None,
    ) -> list[str]:
        """
        Recursive greedy set cover to identify a portfolio.
        """
        if already_selected is None:
            already_selected = []
        if already_covered is None:
            already_covered = set()

        remaining_instances = set(neighbor_perf.index) - already_covered
        if len(already_selected) >= cutoff or not remaining_instances:
            return already_selected

        best_solver: str | None = None
        best_cover: set[str] = set()
        best_runtime = float("inf")

        for algo in self.algorithms:
            if algo in already_selected:
                continue
            covers = (
                set(neighbor_perf.index[neighbor_perf[algo].notna()])
                & remaining_instances
            )

            if not best_solver or len(covers) > len(best_cover):
                best_solver = algo
                best_cover = covers
                best_runtime = (
                    float(neighbor_perf.loc[list(covers), algo].sum())
                    if covers
                    else float("inf")
                )
            elif len(covers) == len(best_cover) and len(covers) > 0:
                runtime = float(neighbor_perf.loc[list(covers), algo].sum())
                if runtime < best_runtime:
                    best_solver = algo
                    best_cover = covers
                    best_runtime = runtime

        if not best_solver or not best_cover:
            return already_selected

        already_selected.append(str(best_solver))
        already_covered |= best_cover
        return self._mine_solvers(
            neighbor_perf, cutoff, already_selected, already_covered
        )

    def _construct_sunny_schedule(
        self, neighbor_perf: pd.DataFrame, lam_limit: int | None = None
    ) -> list[tuple[str, float]]:
        """
        Construct a SUNNY schedule.
        """
        lam = (
            lam_limit
            or self.algorithm_limit
            or self.tuned_algorithm_limit
            or len(self.algorithms)
        )
        lam = max(1, min(int(lam), len(self.algorithms)))

        cutoff = min(self.k, lam, len(self.algorithms))
        best_pfolio = self._mine_solvers(neighbor_perf, cutoff)

        valid_perf = neighbor_perf.notna()
        slots = {algo: int(valid_perf[algo].sum()) for algo in best_pfolio}
        n_unsolved = len(
            set(neighbor_perf.index)
            - set().union(
                *(set(neighbor_perf.index[valid_perf[a]]) for a in best_pfolio)
            )
        )

        total_slots = sum(slots.values()) + n_unsolved
        if total_slots == 0:
            slots = {algo: 1 for algo in best_pfolio}
            total_slots = len(best_pfolio)

        budget = float(self.budget or 1e10)
        schedule: list[tuple[str, float]] = []
        for algo in best_pfolio:
            allocated = budget * (slots[algo] / total_slots)
            schedule.append((str(algo), float(allocated)))

        # Sort by mean running time in neighborhood
        avg_times = neighbor_perf[[a for a, _ in schedule]].mean(axis=0).to_dict()
        schedule.sort(key=lambda x: avg_times.get(x[0], float("inf")))

        # Allocate remaining budget
        used = sum(t for _, t in schedule)
        remaining = max(0.0, budget - used)
        if n_unsolved > 0 and remaining > 0:
            # Add to the globally best solver in the neighborhood
            best_global = str(valid_perf.sum(axis=0).idxmax())
            for i, (a, t) in enumerate(schedule):
                if a == best_global:
                    schedule[i] = (a, t + remaining)
                    break
            else:
                if len(schedule) < lam:
                    schedule.append((best_global, remaining))
                    # Resort
                    avg_t = (
                        float(neighbor_perf[best_global].mean())
                        if best_global in neighbor_perf.columns
                        else float("inf")
                    )
                    avg_times[best_global] = avg_t
                    schedule.sort(key=lambda x: avg_times.get(x[0], float("inf")))
                else:
                    a_last, t_last = schedule[-1]
                    schedule[-1] = (a_last, t_last + remaining)

        return schedule

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict schedules for each instance.
        """
        if self.knn is None or self.performance_df is None:
            raise RuntimeError("SUNNY must be fitted.")

        if features is None:
            raise ValueError("Sunny require features for prediction.")
        predictions: dict[str, list[tuple[str, float]]] = {}
        for instance in features.index:
            x = features.loc[[instance]].values
            _, n_idx = self.knn.kneighbors(x, n_neighbors=self.k)
            n_perf = self.performance_df.iloc[n_idx.flatten()]
            predictions[str(instance)] = self._construct_sunny_schedule(n_perf)
        return predictions

    @staticmethod
    def _define_hyperparameters(
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for SUNNY.
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        use_v2_param = Categorical(name="use_v2", items=[True, False], default=False)
        k_param = Integer(name="k", bounds=(1, 50), default=10)
        n_folds_param = Integer(name="n_folds", bounds=(3, 10), default=5)
        k_candidates_param = Categorical(
            name="k_candidates", items=["small", "medium", "broad"], default="medium"
        )

        params = [use_v2_param, k_param, n_folds_param, k_candidates_param]
        return params, [], []

__init__(k=10, use_v2=False, n_folds=5, k_candidates=None, random_state=42, use_tsunny=False, algorithm_limit=None, **kwargs)

Initialize the SUNNY selector.

Parameters

k : int, default=10 Number of neighbors for k-NN. use_v2 : bool, default=False Whether to tune k using cross-validation (SUNNY-AS2). n_folds : int, default=5 Number of folds for cross-validation when tuning. k_candidates : list[int] or None, default=None Candidate k values to consider when tuning. random_state : int, default=42 Random seed for reproducibility. use_tsunny : bool, default=False Whether to tune the max number of algorithms via cross-validation. algorithm_limit : int or None, default=None If set, cap the number of algorithms in each schedule. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/sunny.py

def __init__(
    self,
    k: int = 10,
    use_v2: bool = False,
    n_folds: int = 5,
    k_candidates: list[int] | None = None,
    random_state: int = 42,
    use_tsunny: bool = False,
    algorithm_limit: int | None = None,
    **kwargs: Any,
) -> None:
    """
    Initialize the SUNNY selector.

    Parameters
    ----------
    k : int, default=10
        Number of neighbors for k-NN.
    use_v2 : bool, default=False
        Whether to tune k using cross-validation (SUNNY-AS2).
    n_folds : int, default=5
        Number of folds for cross-validation when tuning.
    k_candidates : list[int] or None, default=None
        Candidate k values to consider when tuning.
    random_state : int, default=42
        Random seed for reproducibility.
    use_tsunny : bool, default=False
        Whether to tune the max number of algorithms via cross-validation.
    algorithm_limit : int or None, default=None
        If set, cap the number of algorithms in each schedule.
    **kwargs : Any
        Additional keyword arguments.
    """
    super().__init__(**kwargs)
    self.k = int(k)
    self.use_v2 = bool(use_v2)
    self.n_folds = int(n_folds)

    if isinstance(k_candidates, str):
        k_map = {
            "small": [3, 5, 7],
            "medium": [3, 5, 7, 10, 20],
            "broad": [3, 5, 7, 10, 20, 50],
        }
        self.k_candidates = k_map.get(k_candidates, [3, 5, 7, 10, 20, 50])
    else:
        self.k_candidates = k_candidates or [3, 5, 7, 10, 20, 50]

    self.random_state = int(random_state)
    self.use_tsunny = bool(use_tsunny)
    self.algorithm_limit = algorithm_limit
    self.tuned_algorithm_limit: int | None = None

    self.features_df: pd.DataFrame | None = None
    self.performance_df: pd.DataFrame | None = None
    self.knn: NearestNeighbors | None = None

SelectorPipeline

Bases: ConfigurableMixin

Sequence of preprocessing, feature selection, and algorithm selection steps.

Attributes

selector : AbstractSelector The main selector model to be used. pre_solving : AbstractPresolver or None A presolver for selecting initial algorithms. feature_selector : Any or None A component for feature selection. algorithm_pre_selector : Any or None A component for algorithm pre-selection. feature_groups : Any or None Feature groups to be used by the selector. max_feature_time : float or None Budget (seconds) to allocate per feature group in predictions. preprocessor : Pipeline The preprocessing pipeline (including SimpleImputer).

Source code in asf/selectors/selector_pipeline.py

class SelectorPipeline(ConfigurableMixin):
    """
    Sequence of preprocessing, feature selection, and algorithm selection steps.

    Attributes
    ----------
    selector : AbstractSelector
        The main selector model to be used.
    pre_solving : AbstractPresolver or None
        A presolver for selecting initial algorithms.
    feature_selector : Any or None
        A component for feature selection.
    algorithm_pre_selector : Any or None
        A component for algorithm pre-selection.
    feature_groups : Any or None
        Feature groups to be used by the selector.
    max_feature_time : float or None
        Budget (seconds) to allocate per feature group in predictions.
    preprocessor : Pipeline
        The preprocessing pipeline (including SimpleImputer).
    """

    PREFIX = "pipeline"

    def __init__(
        self,
        selector: AbstractSelector,
        preprocessor: Any | list[Any] | None = None,
        pre_solving: AbstractPresolver | None = None,
        feature_selector: Any | None = None,
        algorithm_pre_selector: Any | None = None,
        feature_groups: dict[str, Any] | list[str] | None = None,
        max_feature_time: float | None = None,
    ) -> None:
        """
        Initialize the SelectorPipeline.

        Parameters
        ----------
        selector : AbstractSelector
            The main selector model to be used.
        preprocessor : Any or list or None, default=None
            Preprocessing steps. SimpleImputer(strategy="mean") is always added first.
        pre_solving : AbstractPresolver or None, default=None
            Presolver for initial algorithm selection.
        feature_selector : Any or None, default=None
            Component for feature selection.
        algorithm_pre_selector : Any or None, default=None
            Component for algorithm pre-selection.
        feature_groups : dict or list or None, default=None
            Feature groups configuration.
        max_feature_time : float or None, default=None
            Budget (seconds) per feature group.
        """
        self.selector = selector
        self.pre_solving = pre_solving
        self.feature_selector = feature_selector
        self.algorithm_pre_selector = algorithm_pre_selector
        self.feature_groups = feature_groups
        self.max_feature_time = (
            float(max_feature_time) if max_feature_time is not None else None
        )

        if preprocessor is None:
            preproc_list = []
        elif not isinstance(preprocessor, list):
            preproc_list = [preprocessor]
        else:
            preproc_list = preprocessor

        # Always include SimpleImputer as the first step
        steps = [("SimpleImputer", SimpleImputer(strategy="mean"))]
        for p in preproc_list:
            steps.append((type(p).__name__, p))

        self.preprocessor = Pipeline(steps)
        self.preprocessor.set_output(transform="pandas")

        self._logger = logging.getLogger(__name__)

    def _filter_features(self, X: pd.DataFrame) -> pd.DataFrame:
        """
        Filter features based on selected feature groups.

        Parameters
        ----------
        X : pd.DataFrame
            The input features.

        Returns
        -------
        pd.DataFrame
            Filtered features.
        """
        if self.feature_groups and isinstance(self.feature_groups, dict):
            selected_features = []
            for fg_info in self.feature_groups.values():
                if isinstance(fg_info, dict) and "provides" in fg_info:
                    selected_features.extend(fg_info["provides"])

            available_features = [f for f in selected_features if f in X.columns]
            if available_features:
                return X[available_features]
        return X

    def fit(
        self,
        features: pd.DataFrame,
        performance: pd.DataFrame,
        algorithm_features: pd.DataFrame | None = None,
        **kwargs: Any,
    ) -> None:
        """
        Fit the pipeline.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame
            The performance data.
        algorithm_features : pd.DataFrame or None, default=None
            Optional algorithm features.
        **kwargs : Any
            Additional keyword arguments.
        """
        start = time.time()
        self._logger.debug("Starting fit process")

        X = self.preprocessor.fit_transform(features, performance)
        self._logger.debug(
            f"Preprocessing completed in {time.time() - start:.2f} seconds"
        )
        start = time.time()

        y = performance

        if self.algorithm_pre_selector:
            if hasattr(self.algorithm_pre_selector, "fit_transform"):
                y = self.algorithm_pre_selector.fit_transform(y)
            else:
                self.algorithm_pre_selector.fit(y)
                if hasattr(self.algorithm_pre_selector, "transform"):
                    y = self.algorithm_pre_selector.transform(y)

        self._logger.debug(
            f"Algorithm pre-selection completed in {time.time() - start:.2f} seconds"
        )
        start = time.time()

        if self.pre_solving:
            self.pre_solving.fit(features, performance)

        self._logger.debug(
            f"Pre-solving completed in {time.time() - start:.2f} seconds"
        )
        start = time.time()

        if self.feature_selector:
            if hasattr(self.feature_selector, "fit_transform"):
                X, y = self.feature_selector.fit_transform(X, y)
            else:
                self.feature_selector.fit(X, y)
                X = self.feature_selector.transform(X)

        self._logger.debug(
            f"Feature selection completed in {time.time() - start:.2f} seconds"
        )
        start = time.time()

        X = self._filter_features(X)
        self.selector.fit(X, y, algorithm_features=algorithm_features, **kwargs)

        self._logger.debug(
            f"Selector fitting completed in {time.time() - start:.2f} seconds"
        )

    def predict(
        self,
        features: pd.DataFrame,
        performance: pd.DataFrame | None = None,
        **kwargs: Any,
    ) -> dict[Any, list[tuple[str, float] | tuple[str, float, float]]]:
        """
        Make predictions.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame or None, default=None
            Performance data for oracle selectors.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        dict
            Predictions mapping instance IDs to schedules.
        """
        X = self.preprocessor.transform(features)

        scheds: list[Any] = []
        if self.pre_solving:
            # Presolver prediction returns a single schedule applied to all test instances
            scheds = list(self.pre_solving.predict())

        if self.feature_selector:
            X = self.feature_selector.transform(X)

        X = self._filter_features(X)

        predictions = self.selector.predict(X, performance=performance)

        feature_steps: list[Any] = []
        if self.feature_groups is not None:
            if isinstance(self.feature_groups, dict):
                feature_steps = list(self.feature_groups.keys())
            elif isinstance(self.feature_groups, list):
                feature_steps = self.feature_groups

        if self.max_feature_time is not None and feature_steps:
            feature_steps = [
                (str(fg), float(self.max_feature_time)) for fg in feature_steps
            ]

        final_preds: dict[Any, list[tuple[str, float] | tuple[str, float, float]]] = {}
        assert isinstance(predictions, dict)
        for instance_id in X.index:
            if instance_id in predictions:
                prediction = list(predictions[instance_id])
            elif str(instance_id) in predictions:
                prediction = list(predictions[str(instance_id)])
            else:
                prediction = []
            final_preds[instance_id] = scheds + feature_steps + prediction

        return final_preds

    def save(self, path: str | Path) -> None:
        """
        Save the pipeline to a file.

        Parameters
        ----------
        path : str or Path
            File path to save the pipeline.
        """
        import joblib

        joblib.dump(self, path)

    @staticmethod
    def load(path: str | Path) -> SelectorPipeline:
        """
        Load a pipeline from a file.

        Parameters
        ----------
        path : str or Path
            File path to load from.

        Returns
        -------
        SelectorPipeline
            The loaded pipeline.
        """
        import joblib

        return joblib.load(path)

    def get_config(self) -> dict[str, Any]:
        """
        Return configuration details.

        Returns
        -------
        dict
            Configuration metadata.
        """

        def get_model_name(obj: Any) -> str | None:
            if obj is None:
                return None
            if hasattr(obj, "model_class"):
                mc = obj.model_class
                if hasattr(mc, "func"):
                    return str(mc.func.__name__)
                return str(getattr(mc, "__name__", type(mc).__name__))
            return type(obj).__name__

        return {
            "selector": type(self.selector).__name__,
            "selector_model": get_model_name(self.selector),
            "pre_solving": type(self.pre_solving).__name__
            if self.pre_solving
            else None,
            "selector_budget": getattr(self.selector, "budget", None),
            "presolving_budget": getattr(self.pre_solving, "budget", None),
            "preprocessor_steps": [
                type(step[1]).__name__ for step in self.preprocessor.steps
            ],
            "feature_selector": type(self.feature_selector).__name__
            if self.feature_selector
            else None,
            "algorithm_pre_selector": type(self.algorithm_pre_selector).__name__
            if self.algorithm_pre_selector
            else None,
        }

    @staticmethod
    def _define_hyperparameters(
        selector_class: list[type] | None = None,
        preprocessing_class: list[type] | None = None,
        pre_solving_class: list[type] | None = None,
        feature_groups: dict[str, Any] | None = None,
        algorithm_pre_selector: type | tuple[type, dict[str, Any]] | None = None,
        max_feature_time: float | None | bool = False,
        budget: float | None = None,
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for SelectorPipeline.

        Parameters
        ----------
        selector_class : list[type] or None, default=None
            List of selector classes.
        preprocessing_class : list[type] or None, default=None
            List of preprocessor classes.
        pre_solving_class : list[type] or None, default=None
            List of presolver classes.
        feature_groups : dict or None, default=None
            Feature groups definition.
        algorithm_pre_selector : type or tuple or None, default=None
            Algorithm pre-selector class.
        max_feature_time : float or None or bool, default=False
            Maximum feature computation time.
        budget : float or None, default=None
            Total budget.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        hyperparameters = []
        conditions = []
        forbiddens = []

        if selector_class:
            if (
                isinstance(selector_class, list)
                and selector_class
                and isinstance(selector_class[0], tuple)
            ):
                selector_choices = [
                    c[0] if isinstance(c, tuple) else c for c in selector_class
                ]
            else:
                selector_choices = (
                    selector_class
                    if isinstance(selector_class, list)
                    else [selector_class]
                )
            hyperparameters.append(
                ClassChoice(
                    "selector", choices=cast(list[type | bool], selector_choices)
                )
            )

        if pre_solving_class:
            ps_choices = (
                pre_solving_class
                if isinstance(pre_solving_class, list)
                else [pre_solving_class]
            )
            use_presolver = Categorical(
                "use_presolver", items=[True, False], default=False
            )
            hyperparameters.append(use_presolver)
            presolver_choice = ClassChoice(
                "presolver", choices=cast(list[type | bool], ps_choices)
            )
            hyperparameters.append(cast(Any, presolver_choice))
            conditions.append(
                EqualsCondition(cast(Any, presolver_choice), use_presolver, True)
            )

        if preprocessing_class:
            for preproc_cls in preprocessing_class:
                hyperparameters.append(
                    ClassChoice(
                        f"preprocessor:{preproc_cls.__name__}",
                        choices=[preproc_cls, False],
                        default=False,
                    )
                )

        fg_params = {}
        if feature_groups and len(feature_groups) > 1:
            for fg_name in feature_groups:
                fg_param = Categorical(
                    f"feature_group:{fg_name}", [True, False], default=True
                )
                hyperparameters.append(fg_param)
                fg_params[fg_name] = fg_param

            for fg_name, fg_info in feature_groups.items():
                for req in fg_info.get("requires", []):
                    if req in fg_params:
                        forbiddens.append(
                            ForbiddenAndConjunction(
                                ForbiddenEqualsClause(fg_params[fg_name], True),
                                ForbiddenEqualsClause(fg_params[req], False),
                            )
                        )
            forbiddens.append(
                ForbiddenAndConjunction(
                    *[ForbiddenEqualsClause(p, False) for p in fg_params.values()]
                )
            )

        if algorithm_pre_selector:
            aps_cls = (
                algorithm_pre_selector[0]
                if isinstance(algorithm_pre_selector, tuple)
                else algorithm_pre_selector
            )
            hyperparameters.append(
                ClassChoice("algorithm_pre_selector", choices=[aps_cls])
            )

        if max_feature_time is None:
            upper = float(budget or 3600.0)
            hyperparameters.append(
                UniformFloatHyperparameter(
                    "max_feature_time",
                    lower=0.0,
                    upper=upper,
                    default_value=min(60.0, upper),
                )
            )

        return hyperparameters, conditions, forbiddens

    @classmethod
    def _get_from_clean_configuration(
        cls,
        clean_config: dict[str, Any],
        configuration: Configuration | dict[str, Any] | None = None,
        pre_prefix: str = "",
        feature_groups: dict[str, Any] | None = None,
        max_feature_time: float | None = None,
        **kwargs: Any,
    ) -> partial[SelectorPipeline]:
        """
        Create a SelectorPipeline from a clean configuration.

        Parameters
        ----------
        clean_config : dict
            Clean configuration mapping.
        configuration : Configuration or dict or None, default=None
            Original configuration.
        pre_prefix : str, default=""
            Prefix for nested lookups.
        feature_groups : dict or None, default=None
            Feature groups definition.
        max_feature_time : float or None, default=None
            Maximum feature computation time.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        partial
            Partial function for SelectorPipeline.
        """
        init_kwargs: dict[str, Any] = {}
        prefix = f"{pre_prefix}:{cls.PREFIX}:" if pre_prefix else f"{cls.PREFIX}:"

        if "selector" in clean_config:
            val = clean_config["selector"]
            init_kwargs["selector"] = val() if callable(val) else val

        if clean_config.get("use_presolver") and "presolver" in clean_config:
            val = clean_config["presolver"]
            init_kwargs["pre_solving"] = val() if callable(val) else val

        preprocs = []
        for k, v in clean_config.items():
            if k.startswith("preprocessor:") and v and v != "False":
                preprocs.append(v() if callable(v) else v)
        if preprocs:
            init_kwargs["preprocessor"] = preprocs

        if feature_groups and configuration is not None:
            from asf.preprocessing.feature_group_selector import FeatureGroupSelector

            init_kwargs["feature_groups"] = (
                FeatureGroupSelector.get_selected_groups_from_config(
                    feature_groups, configuration, prefix=f"{prefix}feature_group:"
                )
            )

        if "algorithm_pre_selector" in clean_config:
            val = clean_config["algorithm_pre_selector"]
            init_kwargs["algorithm_pre_selector"] = val() if callable(val) else val

        init_kwargs["max_feature_time"] = clean_config.get(
            "max_feature_time", max_feature_time
        )

        return partial(cls, **init_kwargs)

    @classmethod
    def get_from_configuration(
        cls,
        configuration: Configuration | dict[str, Any],
        pre_prefix: str = "",
        feature_groups: dict[str, Any] | None = None,
        budget: float | None = None,
        max_feature_time: float | None = None,
        **kwargs: Any,
    ) -> partial[SelectorPipeline]:
        """
        Create a SelectorPipeline from a configuration.

        Parameters
        ----------
        configuration : Configuration or dict
            Configuration object.
        pre_prefix : str, default=""
            Prefix for nested lookups.
        feature_groups : dict or None, default=None
            Feature groups definition.
        budget : float or None, default=None
            Total budget.
        max_feature_time : float or None, default=None
            Maximum feature computation time.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        partial
            Partial function for SelectorPipeline.
        """
        if not CONFIGSPACE_AVAILABLE:
            raise RuntimeError("ConfigSpace is not available.")

        prefix = f"{pre_prefix}:{cls.PREFIX}:" if pre_prefix else f"{cls.PREFIX}:"
        clean_config: dict[str, Any] = {}
        cs = getattr(configuration, "config_space", None)

        def resolve(hp_name: str, val: Any) -> Any:
            if isinstance(val, (type, partial)) or callable(val):
                return val
            if cs and val:
                hp = cs.get(hp_name)
                if hp:
                    return cls._resolve_class_from_hp(hp, str(val))
            return val if isinstance(val, type) else None

        # 1. Presolver (do this first to get the budget)
        use_ps = configuration.get(f"{prefix}use_presolver")
        clean_config["use_presolver"] = use_ps
        ps_val = configuration.get(f"{prefix}presolver")
        ps_cls = resolve(f"{prefix}presolver", ps_val)

        presolver_budget = 0.0
        if use_ps and ps_cls:
            if hasattr(ps_cls, "get_from_configuration"):
                clean_config["presolver"] = ps_cls.get_from_configuration(
                    configuration,
                    pre_prefix=f"{prefix}presolver",
                    total_budget=budget,
                    **kwargs,
                )
                # Try to extract the budget from the configuration to subtract it from the selector budget
                # The parameter name is f"{prefix}presolver:{ps_cls.PREFIX}:presolver_budget"
                ps_prefix = f"{prefix}presolver:{ps_cls.PREFIX}:"
                presolver_budget = configuration.get(
                    f"{ps_prefix}presolver_budget", 0.0
                )
            else:
                clean_config["presolver"] = ps_cls()

        # 2. Selector
        s_val = configuration.get(f"{prefix}selector")
        s_cls = resolve(f"{prefix}selector", s_val)

        selector_budget = budget
        if budget is not None:
            selector_budget = max(0.0, budget - presolver_budget)

        if s_cls:
            if hasattr(s_cls, "get_from_configuration"):
                clean_config["selector"] = s_cls.get_from_configuration(
                    configuration,
                    pre_prefix=f"{prefix}selector",
                    budget=selector_budget,
                    **kwargs,
                )
            else:
                clean_config["selector"] = (
                    s_cls(budget=selector_budget) if selector_budget else s_cls()
                )

        # 3. Preprocessors
        if cs:
            for hp in list(cs.values()):
                if hp.name.startswith(f"{prefix}preprocessor:"):
                    val = configuration.get(hp.name)
                    if val and val != "False":
                        res = cls._resolve_class_from_hp(hp, str(val))
                        if res:
                            key = hp.name[len(prefix) :]
                            if hasattr(res, "get_from_configuration"):
                                clean_config[key] = res.get_from_configuration(  # type: ignore[misc]
                                    configuration, pre_prefix=hp.name, **kwargs
                                )
                            else:
                                res_any: Any = res
                                clean_config[key] = (
                                    res_any() if callable(res_any) else res_any
                                )
                    else:
                        clean_config[hp.name[len(prefix) :]] = False

        # 4. Algorithm Pre-selector
        aps_val = configuration.get(f"{prefix}algorithm_pre_selector")
        aps_cls = resolve(f"{prefix}algorithm_pre_selector", aps_val)
        if aps_cls:
            if hasattr(aps_cls, "get_from_configuration"):
                clean_config["algorithm_pre_selector"] = aps_cls.get_from_configuration(
                    configuration,
                    pre_prefix=f"{prefix}algorithm_pre_selector",
                    **kwargs,
                )
            else:
                clean_config["algorithm_pre_selector"] = aps_cls()

        # 5. Max feature time
        mft = configuration.get(f"{prefix}max_feature_time")
        if mft is not None:
            clean_config["max_feature_time"] = mft

        return cls._get_from_clean_configuration(
            clean_config=clean_config,
            configuration=configuration,
            pre_prefix=pre_prefix,
            feature_groups=feature_groups,
            max_feature_time=max_feature_time,
            **kwargs,
        )

__init__(selector, preprocessor=None, pre_solving=None, feature_selector=None, algorithm_pre_selector=None, feature_groups=None, max_feature_time=None)

Initialize the SelectorPipeline.

Parameters

selector : AbstractSelector The main selector model to be used. preprocessor : Any or list or None, default=None Preprocessing steps. SimpleImputer(strategy="mean") is always added first. pre_solving : AbstractPresolver or None, default=None Presolver for initial algorithm selection. feature_selector : Any or None, default=None Component for feature selection. algorithm_pre_selector : Any or None, default=None Component for algorithm pre-selection. feature_groups : dict or list or None, default=None Feature groups configuration. max_feature_time : float or None, default=None Budget (seconds) per feature group.

Source code in asf/selectors/selector_pipeline.py

def __init__(
    self,
    selector: AbstractSelector,
    preprocessor: Any | list[Any] | None = None,
    pre_solving: AbstractPresolver | None = None,
    feature_selector: Any | None = None,
    algorithm_pre_selector: Any | None = None,
    feature_groups: dict[str, Any] | list[str] | None = None,
    max_feature_time: float | None = None,
) -> None:
    """
    Initialize the SelectorPipeline.

    Parameters
    ----------
    selector : AbstractSelector
        The main selector model to be used.
    preprocessor : Any or list or None, default=None
        Preprocessing steps. SimpleImputer(strategy="mean") is always added first.
    pre_solving : AbstractPresolver or None, default=None
        Presolver for initial algorithm selection.
    feature_selector : Any or None, default=None
        Component for feature selection.
    algorithm_pre_selector : Any or None, default=None
        Component for algorithm pre-selection.
    feature_groups : dict or list or None, default=None
        Feature groups configuration.
    max_feature_time : float or None, default=None
        Budget (seconds) per feature group.
    """
    self.selector = selector
    self.pre_solving = pre_solving
    self.feature_selector = feature_selector
    self.algorithm_pre_selector = algorithm_pre_selector
    self.feature_groups = feature_groups
    self.max_feature_time = (
        float(max_feature_time) if max_feature_time is not None else None
    )

    if preprocessor is None:
        preproc_list = []
    elif not isinstance(preprocessor, list):
        preproc_list = [preprocessor]
    else:
        preproc_list = preprocessor

    # Always include SimpleImputer as the first step
    steps = [("SimpleImputer", SimpleImputer(strategy="mean"))]
    for p in preproc_list:
        steps.append((type(p).__name__, p))

    self.preprocessor = Pipeline(steps)
    self.preprocessor.set_output(transform="pandas")

    self._logger = logging.getLogger(__name__)

fit(features, performance, algorithm_features=None, **kwargs)

Fit the pipeline.

Parameters

features : pd.DataFrame The input features. performance : pd.DataFrame The performance data. algorithm_features : pd.DataFrame or None, default=None Optional algorithm features. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/selector_pipeline.py

def fit(
    self,
    features: pd.DataFrame,
    performance: pd.DataFrame,
    algorithm_features: pd.DataFrame | None = None,
    **kwargs: Any,
) -> None:
    """
    Fit the pipeline.

    Parameters
    ----------
    features : pd.DataFrame
        The input features.
    performance : pd.DataFrame
        The performance data.
    algorithm_features : pd.DataFrame or None, default=None
        Optional algorithm features.
    **kwargs : Any
        Additional keyword arguments.
    """
    start = time.time()
    self._logger.debug("Starting fit process")

    X = self.preprocessor.fit_transform(features, performance)
    self._logger.debug(
        f"Preprocessing completed in {time.time() - start:.2f} seconds"
    )
    start = time.time()

    y = performance

    if self.algorithm_pre_selector:
        if hasattr(self.algorithm_pre_selector, "fit_transform"):
            y = self.algorithm_pre_selector.fit_transform(y)
        else:
            self.algorithm_pre_selector.fit(y)
            if hasattr(self.algorithm_pre_selector, "transform"):
                y = self.algorithm_pre_selector.transform(y)

    self._logger.debug(
        f"Algorithm pre-selection completed in {time.time() - start:.2f} seconds"
    )
    start = time.time()

    if self.pre_solving:
        self.pre_solving.fit(features, performance)

    self._logger.debug(
        f"Pre-solving completed in {time.time() - start:.2f} seconds"
    )
    start = time.time()

    if self.feature_selector:
        if hasattr(self.feature_selector, "fit_transform"):
            X, y = self.feature_selector.fit_transform(X, y)
        else:
            self.feature_selector.fit(X, y)
            X = self.feature_selector.transform(X)

    self._logger.debug(
        f"Feature selection completed in {time.time() - start:.2f} seconds"
    )
    start = time.time()

    X = self._filter_features(X)
    self.selector.fit(X, y, algorithm_features=algorithm_features, **kwargs)

    self._logger.debug(
        f"Selector fitting completed in {time.time() - start:.2f} seconds"
    )

get_config()

Return configuration details.

Returns

dict Configuration metadata.

Source code in asf/selectors/selector_pipeline.py

def get_config(self) -> dict[str, Any]:
    """
    Return configuration details.

    Returns
    -------
    dict
        Configuration metadata.
    """

    def get_model_name(obj: Any) -> str | None:
        if obj is None:
            return None
        if hasattr(obj, "model_class"):
            mc = obj.model_class
            if hasattr(mc, "func"):
                return str(mc.func.__name__)
            return str(getattr(mc, "__name__", type(mc).__name__))
        return type(obj).__name__

    return {
        "selector": type(self.selector).__name__,
        "selector_model": get_model_name(self.selector),
        "pre_solving": type(self.pre_solving).__name__
        if self.pre_solving
        else None,
        "selector_budget": getattr(self.selector, "budget", None),
        "presolving_budget": getattr(self.pre_solving, "budget", None),
        "preprocessor_steps": [
            type(step[1]).__name__ for step in self.preprocessor.steps
        ],
        "feature_selector": type(self.feature_selector).__name__
        if self.feature_selector
        else None,
        "algorithm_pre_selector": type(self.algorithm_pre_selector).__name__
        if self.algorithm_pre_selector
        else None,
    }

get_from_configuration(configuration, pre_prefix='', feature_groups=None, budget=None, max_feature_time=None, **kwargs) classmethod

Create a SelectorPipeline from a configuration.

Parameters

configuration : Configuration or dict Configuration object. pre_prefix : str, default="" Prefix for nested lookups. feature_groups : dict or None, default=None Feature groups definition. budget : float or None, default=None Total budget. max_feature_time : float or None, default=None Maximum feature computation time. **kwargs : Any Additional keyword arguments.

Returns

partial Partial function for SelectorPipeline.

Source code in asf/selectors/selector_pipeline.py

@classmethod
def get_from_configuration(
    cls,
    configuration: Configuration | dict[str, Any],
    pre_prefix: str = "",
    feature_groups: dict[str, Any] | None = None,
    budget: float | None = None,
    max_feature_time: float | None = None,
    **kwargs: Any,
) -> partial[SelectorPipeline]:
    """
    Create a SelectorPipeline from a configuration.

    Parameters
    ----------
    configuration : Configuration or dict
        Configuration object.
    pre_prefix : str, default=""
        Prefix for nested lookups.
    feature_groups : dict or None, default=None
        Feature groups definition.
    budget : float or None, default=None
        Total budget.
    max_feature_time : float or None, default=None
        Maximum feature computation time.
    **kwargs : Any
        Additional keyword arguments.

    Returns
    -------
    partial
        Partial function for SelectorPipeline.
    """
    if not CONFIGSPACE_AVAILABLE:
        raise RuntimeError("ConfigSpace is not available.")

    prefix = f"{pre_prefix}:{cls.PREFIX}:" if pre_prefix else f"{cls.PREFIX}:"
    clean_config: dict[str, Any] = {}
    cs = getattr(configuration, "config_space", None)

    def resolve(hp_name: str, val: Any) -> Any:
        if isinstance(val, (type, partial)) or callable(val):
            return val
        if cs and val:
            hp = cs.get(hp_name)
            if hp:
                return cls._resolve_class_from_hp(hp, str(val))
        return val if isinstance(val, type) else None

    # 1. Presolver (do this first to get the budget)
    use_ps = configuration.get(f"{prefix}use_presolver")
    clean_config["use_presolver"] = use_ps
    ps_val = configuration.get(f"{prefix}presolver")
    ps_cls = resolve(f"{prefix}presolver", ps_val)

    presolver_budget = 0.0
    if use_ps and ps_cls:
        if hasattr(ps_cls, "get_from_configuration"):
            clean_config["presolver"] = ps_cls.get_from_configuration(
                configuration,
                pre_prefix=f"{prefix}presolver",
                total_budget=budget,
                **kwargs,
            )
            # Try to extract the budget from the configuration to subtract it from the selector budget
            # The parameter name is f"{prefix}presolver:{ps_cls.PREFIX}:presolver_budget"
            ps_prefix = f"{prefix}presolver:{ps_cls.PREFIX}:"
            presolver_budget = configuration.get(
                f"{ps_prefix}presolver_budget", 0.0
            )
        else:
            clean_config["presolver"] = ps_cls()

    # 2. Selector
    s_val = configuration.get(f"{prefix}selector")
    s_cls = resolve(f"{prefix}selector", s_val)

    selector_budget = budget
    if budget is not None:
        selector_budget = max(0.0, budget - presolver_budget)

    if s_cls:
        if hasattr(s_cls, "get_from_configuration"):
            clean_config["selector"] = s_cls.get_from_configuration(
                configuration,
                pre_prefix=f"{prefix}selector",
                budget=selector_budget,
                **kwargs,
            )
        else:
            clean_config["selector"] = (
                s_cls(budget=selector_budget) if selector_budget else s_cls()
            )

    # 3. Preprocessors
    if cs:
        for hp in list(cs.values()):
            if hp.name.startswith(f"{prefix}preprocessor:"):
                val = configuration.get(hp.name)
                if val and val != "False":
                    res = cls._resolve_class_from_hp(hp, str(val))
                    if res:
                        key = hp.name[len(prefix) :]
                        if hasattr(res, "get_from_configuration"):
                            clean_config[key] = res.get_from_configuration(  # type: ignore[misc]
                                configuration, pre_prefix=hp.name, **kwargs
                            )
                        else:
                            res_any: Any = res
                            clean_config[key] = (
                                res_any() if callable(res_any) else res_any
                            )
                else:
                    clean_config[hp.name[len(prefix) :]] = False

    # 4. Algorithm Pre-selector
    aps_val = configuration.get(f"{prefix}algorithm_pre_selector")
    aps_cls = resolve(f"{prefix}algorithm_pre_selector", aps_val)
    if aps_cls:
        if hasattr(aps_cls, "get_from_configuration"):
            clean_config["algorithm_pre_selector"] = aps_cls.get_from_configuration(
                configuration,
                pre_prefix=f"{prefix}algorithm_pre_selector",
                **kwargs,
            )
        else:
            clean_config["algorithm_pre_selector"] = aps_cls()

    # 5. Max feature time
    mft = configuration.get(f"{prefix}max_feature_time")
    if mft is not None:
        clean_config["max_feature_time"] = mft

    return cls._get_from_clean_configuration(
        clean_config=clean_config,
        configuration=configuration,
        pre_prefix=pre_prefix,
        feature_groups=feature_groups,
        max_feature_time=max_feature_time,
        **kwargs,
    )

load(path) staticmethod

Load a pipeline from a file.

Parameters

path : str or Path File path to load from.

Returns

SelectorPipeline The loaded pipeline.

Source code in asf/selectors/selector_pipeline.py

@staticmethod
def load(path: str | Path) -> SelectorPipeline:
    """
    Load a pipeline from a file.

    Parameters
    ----------
    path : str or Path
        File path to load from.

    Returns
    -------
    SelectorPipeline
        The loaded pipeline.
    """
    import joblib

    return joblib.load(path)

predict(features, performance=None, **kwargs)

Make predictions.

Parameters

features : pd.DataFrame The input features. performance : pd.DataFrame or None, default=None Performance data for oracle selectors. **kwargs : Any Additional keyword arguments.

Returns

dict Predictions mapping instance IDs to schedules.

Source code in asf/selectors/selector_pipeline.py

def predict(
    self,
    features: pd.DataFrame,
    performance: pd.DataFrame | None = None,
    **kwargs: Any,
) -> dict[Any, list[tuple[str, float] | tuple[str, float, float]]]:
    """
    Make predictions.

    Parameters
    ----------
    features : pd.DataFrame
        The input features.
    performance : pd.DataFrame or None, default=None
        Performance data for oracle selectors.
    **kwargs : Any
        Additional keyword arguments.

    Returns
    -------
    dict
        Predictions mapping instance IDs to schedules.
    """
    X = self.preprocessor.transform(features)

    scheds: list[Any] = []
    if self.pre_solving:
        # Presolver prediction returns a single schedule applied to all test instances
        scheds = list(self.pre_solving.predict())

    if self.feature_selector:
        X = self.feature_selector.transform(X)

    X = self._filter_features(X)

    predictions = self.selector.predict(X, performance=performance)

    feature_steps: list[Any] = []
    if self.feature_groups is not None:
        if isinstance(self.feature_groups, dict):
            feature_steps = list(self.feature_groups.keys())
        elif isinstance(self.feature_groups, list):
            feature_steps = self.feature_groups

    if self.max_feature_time is not None and feature_steps:
        feature_steps = [
            (str(fg), float(self.max_feature_time)) for fg in feature_steps
        ]

    final_preds: dict[Any, list[tuple[str, float] | tuple[str, float, float]]] = {}
    assert isinstance(predictions, dict)
    for instance_id in X.index:
        if instance_id in predictions:
            prediction = list(predictions[instance_id])
        elif str(instance_id) in predictions:
            prediction = list(predictions[str(instance_id)])
        else:
            prediction = []
        final_preds[instance_id] = scheds + feature_steps + prediction

    return final_preds

save(path)

Save the pipeline to a file.

Parameters

path : str or Path File path to save the pipeline.

Source code in asf/selectors/selector_pipeline.py

def save(self, path: str | Path) -> None:
    """
    Save the pipeline to a file.

    Parameters
    ----------
    path : str or Path
        File path to save the pipeline.
    """
    import joblib

    joblib.dump(self, path)

SimpleRanking

Bases: ConfigurableMixin, AbstractModelBasedSelector

Algorithm Selection via Ranking.

Attributes

classifier : AbstractPredictor or None The trained ranking model.

Source code in asf/selectors/simple_ranking.py

class SimpleRanking(ConfigurableMixin, AbstractModelBasedSelector):
    """
    Algorithm Selection via Ranking.

    Attributes
    ----------
    classifier : AbstractPredictor or None
        The trained ranking model.
    """

    PREFIX = "simple_ranking"
    RETURN_TYPE = "single"

    def __init__(
        self,
        model_class: type[AbstractPredictor] = XGBoostRankerWrapper,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the SimpleRanking.

        Parameters
        ----------
        model_class : type[AbstractPredictor], default=XGBoostRankerWrapper
            The class of the ranking model to be used.
        **kwargs : Any
            Additional keyword arguments.
        """
        AbstractModelBasedSelector.__init__(self, model_class, **kwargs)
        self.classifier: AbstractPredictor | None = None

    def _fit(
        self,
        features: pd.DataFrame,
        performance: pd.DataFrame,
        **kwargs: Any,
    ) -> None:
        """
        Fit the ranking model.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame
            The algorithm performance data.
        """
        if self.algorithm_features is None:
            encoder = OneHotEncoder(sparse_output=False)
            self.algorithm_features = pd.DataFrame(
                encoder.fit_transform(np.array(self.algorithms).reshape(-1, 1)),
                index=list(self.algorithms),
                columns=[f"algo_{i}" for i in range(len(self.algorithms))],
            )

        performance = performance[self.algorithms]
        features = features[list(self.features)]

        # Reset index to have instance names as a column for merging
        features_reset = features.reset_index().rename(
            columns={features.index.name or "index": "INSTANCE_ID"}
        )
        self.algorithm_features.index.name = "ALGORITHM"

        # Create cross-product of instances and algorithms
        total_features = pd.merge(
            features_reset, self.algorithm_features.reset_index(), how="cross"
        )

        stacked_perf = performance.stack().reset_index()
        stacked_perf.columns = ["INSTANCE_ID", "ALGORITHM", "PERFORMANCE"]

        merged = total_features.merge(
            stacked_perf, on=["INSTANCE_ID", "ALGORITHM"], how="left"
        )

        # Calculate ranks per instance
        gdfs = []
        for _name, gdf in merged.groupby("INSTANCE_ID"):
            gdf["rank"] = gdf["PERFORMANCE"].rank(
                ascending=not self.maximize, method="min"
            )
            gdfs.append(gdf)
        merged = pd.concat(gdfs)

        # Features for training
        X = merged.drop(columns=["INSTANCE_ID", "ALGORITHM", "PERFORMANCE", "rank"])
        y = merged["rank"]

        q_encoder = OrdinalEncoder()
        qid = q_encoder.fit_transform(
            merged["INSTANCE_ID"].to_numpy().reshape(-1, 1)
        ).flatten()

        self.classifier = self.model_class()
        if self.classifier is None:
            raise RuntimeError("Classifier could not be initialized.")

        self.classifier.fit(X, y, qid=qid)

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict the best algorithm for each instance.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.

        Returns
        -------
        dict
            Mapping from instance names to algorithm schedules.
        """
        if features is None:
            raise ValueError("SimpleRanking require features for prediction.")
        if self.classifier is None:
            raise RuntimeError("Classifier has not been fitted.")
        if self.algorithm_features is None:
            raise RuntimeError("Algorithm features missing.")

        f_cols = list(self.features)
        inst_name = features.index.name or "index"
        features_reset = features.reset_index().rename(
            columns={inst_name: "INSTANCE_ID"}
        )

        total_features = pd.merge(
            features_reset, self.algorithm_features.reset_index(), how="cross"
        )

        X = total_features[f_cols + list(self.algorithm_features.columns)]
        predictions = self.classifier.predict(X)
        results: dict[str, list[tuple[str, float]]] = {}
        for i, instance_name in enumerate(features.index):
            mask = total_features["INSTANCE_ID"] == instance_name
            # Local best for this group
            best_idx = int(np.argmin(predictions[mask]))
            best_algo = total_features[mask].iloc[best_idx]["ALGORITHM"]
            results[str(instance_name)] = [(str(best_algo), float(self.budget or 0))]

        return results

    @staticmethod
    def _define_hyperparameters(
        model_class: list[type[AbstractPredictor]] | None = None,
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for SimpleRanking.

        Parameters
        ----------
        model_class : list[type[AbstractPredictor]] or None, default=None
            List of model classes to choose from.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        if model_class is None:
            model_class = [XGBoostRankerWrapper]

        model_class_param = ClassChoice(
            name="model_class",
            choices=cast(list[type | bool], model_class),
            default=model_class[0],
        )

        return [model_class_param], [], []

__init__(model_class=XGBoostRankerWrapper, **kwargs)

Initialize the SimpleRanking.

Parameters

model_class : type[AbstractPredictor], default=XGBoostRankerWrapper The class of the ranking model to be used. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/simple_ranking.py

def __init__(
    self,
    model_class: type[AbstractPredictor] = XGBoostRankerWrapper,
    **kwargs: Any,
) -> None:
    """
    Initialize the SimpleRanking.

    Parameters
    ----------
    model_class : type[AbstractPredictor], default=XGBoostRankerWrapper
        The class of the ranking model to be used.
    **kwargs : Any
        Additional keyword arguments.
    """
    AbstractModelBasedSelector.__init__(self, model_class, **kwargs)
    self.classifier: AbstractPredictor | None = None

SingleBestSolver

Bases: ConfigurableMixin, AbstractSelector

Single Best Solver (SBS) selector.

Always selects the algorithm with the best average performance across all training instances. This represents the baseline performance achievable without any instance-specific selection.

Attributes

best_algorithm : str or None The name of the algorithm with the best aggregate performance.

Source code in asf/selectors/baselines.py

class SingleBestSolver(ConfigurableMixin, AbstractSelector):
    """
    Single Best Solver (SBS) selector.

    Always selects the algorithm with the best average performance across all
    training instances. This represents the baseline performance achievable
    without any instance-specific selection.

    Attributes
    ----------
    best_algorithm : str or None
        The name of the algorithm with the best aggregate performance.
    """

    PREFIX = "sbs"
    RETURN_TYPE = "single"

    def __init__(
        self,
        budget: float | None = None,
        maximize: bool = False,
        feature_groups: list[str] | None = None,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the SingleBestSolver.

        Parameters
        ----------
        budget : float or None, default=None
            The budget for the selector.
        maximize : bool, default=False
            Indicates whether to maximize the performance metric.
        feature_groups : list[str] or None, default=None
            Groups of features to be considered.
        **kwargs : Any
            Additional keyword arguments.
        """
        super().__init__(
            budget=budget,
            maximize=maximize,
            feature_groups=feature_groups,
            **kwargs,
        )
        self.best_algorithm: str | None = None

    def _fit(
        self,
        features: pd.DataFrame,
        performance: pd.DataFrame,
        **kwargs: Any,
    ) -> None:
        """
        Find the single best algorithm based on aggregate performance.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame
            The performance data.
        **kwargs : Any
            Additional keyword arguments.
        """
        # Apply PAR10 penalty for comparison
        if self.budget is not None:
            perf_penalized = np.where(
                performance <= self.budget, performance, self.budget * 10
            )
        else:
            perf_penalized = performance.values

        # Aggregate performance across all instances
        perf_sum = np.sum(perf_penalized, axis=0)

        if self.maximize:
            best_idx = np.argmax(perf_sum)
        else:
            best_idx = np.argmin(perf_sum)

        self.best_algorithm = performance.columns[best_idx]

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
                Predict the single best algorithm for all instances.

                Parameters
                ----------
                features : pd.DataFrame or None
                    The input features.
                performance : pd.DataFrame or None, default=None
                    The performance data.

                Returns
        -------
                dict
                    Dictionary mapping instance IDs to the single best algorithm.
        """
        indices = features.index if features is not None else [0]
        return {
            str(instance): [(str(self.best_algorithm), float(self.budget or 0))]
            for instance in indices
        }

    @staticmethod
    def _define_hyperparameters(
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for SingleBestSolver.

        Returns
        -------
        tuple
            Empty hyperparameters, conditions, and forbiddens.
        """
        return [], [], []

__init__(budget=None, maximize=False, feature_groups=None, **kwargs)

Initialize the SingleBestSolver.

Parameters

budget : float or None, default=None The budget for the selector. maximize : bool, default=False Indicates whether to maximize the performance metric. feature_groups : list[str] or None, default=None Groups of features to be considered. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/baselines.py

def __init__(
    self,
    budget: float | None = None,
    maximize: bool = False,
    feature_groups: list[str] | None = None,
    **kwargs: Any,
) -> None:
    """
    Initialize the SingleBestSolver.

    Parameters
    ----------
    budget : float or None, default=None
        The budget for the selector.
    maximize : bool, default=False
        Indicates whether to maximize the performance metric.
    feature_groups : list[str] or None, default=None
        Groups of features to be considered.
    **kwargs : Any
        Additional keyword arguments.
    """
    super().__init__(
        budget=budget,
        maximize=maximize,
        feature_groups=feature_groups,
        **kwargs,
    )
    self.best_algorithm: str | None = None

StackingSelector

Bases: ConfigurableMixin, AbstractSelector

Stacking ensemble for algorithm selectors.

Trains base selectors and uses their outputs as features for a meta-selector. For selectors with generate_features(), can optionally use the generated features instead of final predictions.

Attributes

base_selectors : list[AbstractSelector] List of base selectors. meta_selector : AbstractSelector Meta-selector trained on base selector outputs. use_generated_features : bool If True, use generate_features() output for selectors that support it. cv : int Number of cross-validation folds for generating meta-features. use_original_features : bool If True, include original features in meta-features. random_state : int or None Random state for cross-validation. base_selectors_ : list[AbstractSelector] Fitted base selectors. meta_selector_ : AbstractSelector Fitted meta-selector.

Source code in asf/selectors/ensembles.py

class StackingSelector(ConfigurableMixin, AbstractSelector):
    """Stacking ensemble for algorithm selectors.

    Trains base selectors and uses their outputs as features for a meta-selector.
    For selectors with generate_features(), can optionally use the generated
    features instead of final predictions.

    Attributes
    ----------
    base_selectors : list[AbstractSelector]
        List of base selectors.
    meta_selector : AbstractSelector
        Meta-selector trained on base selector outputs.
    use_generated_features : bool
        If True, use generate_features() output for selectors that support it.
    cv : int
        Number of cross-validation folds for generating meta-features.
    use_original_features : bool
        If True, include original features in meta-features.
    random_state : int or None
        Random state for cross-validation.
    base_selectors_ : list[AbstractSelector]
        Fitted base selectors.
    meta_selector_ : AbstractSelector
        Fitted meta-selector.
    """

    PREFIX = "stacking"
    RETURN_TYPE = "single"

    def __init__(
        self,
        base_selectors: (
            list[AbstractSelector] | Callable[[], list[AbstractSelector]] | None
        ) = None,
        meta_selector: AbstractSelector | Callable[[], AbstractSelector] | None = None,
        use_generated_features: bool = False,
        cv: int = 5,
        use_original_features: bool = True,
        random_state: int | None = None,
        **kwargs: Any,
    ) -> None:
        """Initialize the StackingSelector.

        Parameters
        ----------
        base_selectors : list[AbstractSelector] or Callable or None
            List of base selectors.
        meta_selector : AbstractSelector or Callable or None
            Meta-selector for final predictions.
        use_generated_features : bool, default=False
            If True, use generate_features() output for selectors that have it.
            If False, use one-hot encoded final predictions.
        cv : int, default=5
            Number of cross-validation folds.
        use_original_features : bool, default=True
            If True, concatenate original features with stacked features.
        random_state : int or None, default=None
            Random state for cross-validation.
        **kwargs : Any
            Additional keyword arguments passed to AbstractSelector.
        """
        super().__init__(**kwargs)

        if callable(base_selectors) and not isinstance(base_selectors, list):
            self.base_selectors = base_selectors()
        else:
            self.base_selectors = (
                list(base_selectors) if base_selectors is not None else []
            )

        if callable(meta_selector) and not isinstance(meta_selector, AbstractSelector):
            self.meta_selector = meta_selector()
        else:
            self.meta_selector = meta_selector

        if not self.base_selectors:
            raise ValueError("base_selectors list cannot be empty")
        if self.meta_selector is None:
            raise ValueError("meta_selector cannot be None")

        self.use_generated_features = bool(use_generated_features)
        self.cv = int(cv)
        self.use_original_features = bool(use_original_features)
        self.random_state = random_state

        self.base_selectors_: list[AbstractSelector] = []
        self.meta_selector_: AbstractSelector | None = None
        self._encoder: OneHotEncoder | None = None

    def _get_selector_features(
        self,
        selector: AbstractSelector,
        features: pd.DataFrame,
    ) -> pd.DataFrame:
        """Get features from a selector for stacking.

        Parameters
        ----------
        selector : AbstractSelector
            The fitted selector.
        features : pd.DataFrame
            Input features.

        Returns
        -------
        pd.DataFrame
            Features generated by the selector.
        """
        if self.use_generated_features and _has_generate_features(selector):
            # Use generate_features output
            gen_features = selector.generate_features(features)  # type: ignore[attr-defined]
            if isinstance(gen_features, pd.DataFrame):
                return gen_features
            return pd.DataFrame(gen_features, index=features.index)

        # Fall back to one-hot encoded predictions
        preds = selector.predict(features)
        if isinstance(preds, dict):
            preds_dict: dict[str, Any] = cast(Any, preds)
            labels = []
            for idx in features.index:
                idx_str = str(idx)
                schedule = preds_dict.get(idx_str) if idx_str in preds_dict else None
                if schedule and isinstance(schedule, list) and len(schedule) > 0:
                    first_item = schedule[0]
                    if isinstance(first_item, tuple) and len(first_item) >= 1:
                        labels.append(str(first_item[0]))
                    else:
                        labels.append(
                            self.algorithms[0] if self.algorithms else "unknown"
                        )
                else:
                    labels.append(self.algorithms[0] if self.algorithms else "unknown")
        else:
            labels = [self.algorithms[0]] * len(features)

        # One-hot encode
        if self._encoder is None:
            self._encoder = OneHotEncoder(
                sparse_output=False,
                categories=[self.algorithms],
                handle_unknown="ignore",
            )
            self._encoder.fit(np.array(self.algorithms).reshape(-1, 1))

        encoded = self._encoder.transform(np.array(labels).reshape(-1, 1))
        return pd.DataFrame(encoded, index=features.index)

    def _build_meta_features(
        self,
        selectors: list[AbstractSelector],
        features: pd.DataFrame,
    ) -> pd.DataFrame:
        """Build meta-features from all base selectors.

        Parameters
        ----------
        selectors : list[AbstractSelector]
            List of fitted selectors.
        features : pd.DataFrame
            Input features.

        Returns
        -------
        pd.DataFrame
            Concatenated meta-features from all selectors.
        """
        meta_features_list = []

        for i, selector in enumerate(selectors):
            sel_features = self._get_selector_features(selector, features)
            # Prefix columns to avoid collisions
            sel_features.columns = pd.Index(
                [f"sel{i}_{col}" for col in sel_features.columns]
            )
            meta_features_list.append(sel_features)

        meta_features = pd.concat(meta_features_list, axis=1)

        if self.use_original_features:
            meta_features = pd.concat(
                [features.reset_index(drop=True), meta_features.reset_index(drop=True)],
                axis=1,
            )
            meta_features.index = features.index

        return meta_features

    def _fit(
        self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs: Any
    ) -> None:
        """Fit the stacking ensemble.

        Parameters
        ----------
        features : pd.DataFrame
            Training features.
        performance : pd.DataFrame
            Algorithm performance data.
        **kwargs : Any
            Additional keyword arguments.
        """
        n_samples = len(features)
        n_splits = min(self.cv, n_samples)

        # Initialize encoder with all algorithms
        self._encoder = OneHotEncoder(
            sparse_output=False, categories=[self.algorithms], handle_unknown="ignore"
        )
        self._encoder.fit(np.array(self.algorithms).reshape(-1, 1))

        # Cross-validation to generate meta-features without data leakage
        kf = KFold(n_splits=n_splits, shuffle=True, random_state=self.random_state)

        # Store out-of-fold predictions for each selector
        oof_features: list[pd.DataFrame] = [
            pd.DataFrame(index=features.index) for _ in self.base_selectors
        ]

        for train_idx, val_idx in kf.split(np.arange(n_samples)):
            train_ix = features.index[train_idx]
            val_ix = features.index[val_idx]

            X_train = features.loc[train_ix]
            y_train = performance.loc[train_ix]
            X_val = features.loc[val_ix]

            for i, selector in enumerate(self.base_selectors):
                fold_selector = _clone_selector(selector)
                fold_selector.fit(X_train, y_train, **kwargs)

                val_features = self._get_selector_features(fold_selector, X_val)
                val_features.columns = pd.Index(
                    [f"sel{i}_{col}" for col in val_features.columns]
                )

                for col in val_features.columns:
                    if col not in oof_features[i].columns:
                        oof_features[i][col] = np.nan
                    oof_features[i].loc[val_ix, col] = val_features[col].values

        # Combine out-of-fold features
        meta_train_features = pd.concat(oof_features, axis=1)
        meta_train_features = meta_train_features.fillna(0)

        if self.use_original_features:
            meta_train_features = pd.concat(
                [
                    features.reset_index(drop=True),
                    meta_train_features.reset_index(drop=True),
                ],
                axis=1,
            )
            meta_train_features.index = features.index

        # Fit base selectors on full data
        self.base_selectors_ = []
        for selector in self.base_selectors:
            fitted = _clone_selector(selector)
            fitted.fit(features, performance, **kwargs)
            self.base_selectors_.append(fitted)

        # Fit meta-selector
        assert self.meta_selector is not None
        self.meta_selector_ = _clone_selector(self.meta_selector)
        self.meta_selector_.fit(meta_train_features, performance, **kwargs)

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """Predict using the stacking ensemble.

        Parameters
        ----------
        features : pd.DataFrame or None
            Input features for prediction.
        performance : pd.DataFrame or None, default=None
            Performance data (unused).

        Returns
        -------
        dict
            Mapping from instance names to algorithm schedules.
        """
        if features is None:
            raise ValueError("StackingSelector requires features for prediction.")

        if not self.base_selectors_ or self.meta_selector_ is None:
            raise RuntimeError("StackingSelector has not been fitted.")

        # Build meta-features from fitted base selectors
        meta_features = self._build_meta_features(self.base_selectors_, features)

        # Get predictions from meta-selector
        preds = self.meta_selector_.predict(meta_features)

        if isinstance(preds, dict):
            return cast(
                dict[str, list[tuple[str, int | float]]],
                {str(k): v for k, v in preds.items()},
            )

        # Handle other return types
        result: dict[str, list[tuple[str, float]]] = {}
        for idx in features.index:
            result[str(idx)] = []
        return result

    @staticmethod
    def _define_hyperparameters(
        base_selector_classes: list[type] | None = None,
        meta_selector_classes: list[type] | None = None,
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """Define hyperparameters for StackingSelector.

        Parameters
        ----------
        base_selector_classes : list[type] or None, default=None
            List of base selector classes.
        meta_selector_classes : list[type] or None, default=None
            List of meta selector classes.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        hyperparameters: list[Any] = [
            Categorical("use_generated_features", items=[True, False], default=False),
            Integer("cv", bounds=(2, 10), default=5),
            Categorical("use_original_features", items=[True, False], default=True),
        ]

        if meta_selector_classes:
            hyperparameters.append(
                ClassChoice(
                    "meta_selector",
                    choices=cast(list[type | bool], meta_selector_classes),
                )
            )

        return hyperparameters, [], []

__init__(base_selectors=None, meta_selector=None, use_generated_features=False, cv=5, use_original_features=True, random_state=None, **kwargs)

Initialize the StackingSelector.

Parameters

base_selectors : list[AbstractSelector] or Callable or None List of base selectors. meta_selector : AbstractSelector or Callable or None Meta-selector for final predictions. use_generated_features : bool, default=False If True, use generate_features() output for selectors that have it. If False, use one-hot encoded final predictions. cv : int, default=5 Number of cross-validation folds. use_original_features : bool, default=True If True, concatenate original features with stacked features. random_state : int or None, default=None Random state for cross-validation. **kwargs : Any Additional keyword arguments passed to AbstractSelector.

Source code in asf/selectors/ensembles.py

def __init__(
    self,
    base_selectors: (
        list[AbstractSelector] | Callable[[], list[AbstractSelector]] | None
    ) = None,
    meta_selector: AbstractSelector | Callable[[], AbstractSelector] | None = None,
    use_generated_features: bool = False,
    cv: int = 5,
    use_original_features: bool = True,
    random_state: int | None = None,
    **kwargs: Any,
) -> None:
    """Initialize the StackingSelector.

    Parameters
    ----------
    base_selectors : list[AbstractSelector] or Callable or None
        List of base selectors.
    meta_selector : AbstractSelector or Callable or None
        Meta-selector for final predictions.
    use_generated_features : bool, default=False
        If True, use generate_features() output for selectors that have it.
        If False, use one-hot encoded final predictions.
    cv : int, default=5
        Number of cross-validation folds.
    use_original_features : bool, default=True
        If True, concatenate original features with stacked features.
    random_state : int or None, default=None
        Random state for cross-validation.
    **kwargs : Any
        Additional keyword arguments passed to AbstractSelector.
    """
    super().__init__(**kwargs)

    if callable(base_selectors) and not isinstance(base_selectors, list):
        self.base_selectors = base_selectors()
    else:
        self.base_selectors = (
            list(base_selectors) if base_selectors is not None else []
        )

    if callable(meta_selector) and not isinstance(meta_selector, AbstractSelector):
        self.meta_selector = meta_selector()
    else:
        self.meta_selector = meta_selector

    if not self.base_selectors:
        raise ValueError("base_selectors list cannot be empty")
    if self.meta_selector is None:
        raise ValueError("meta_selector cannot be None")

    self.use_generated_features = bool(use_generated_features)
    self.cv = int(cv)
    self.use_original_features = bool(use_original_features)
    self.random_state = random_state

    self.base_selectors_: list[AbstractSelector] = []
    self.meta_selector_: AbstractSelector | None = None
    self._encoder: OneHotEncoder | None = None

VirtualBestSolver

Bases: ConfigurableMixin, AbstractSelector

Virtual Best Solver (VBS) / Oracle selector.

Always selects the best algorithm for each specific instance. This represents the upper bound of performance achievable by any algorithm selector (requires oracle knowledge of true performance).

Note: This selector "cheats" by using the test performance data.

Source code in asf/selectors/baselines.py

class VirtualBestSolver(ConfigurableMixin, AbstractSelector):
    """
    Virtual Best Solver (VBS) / Oracle selector.

    Always selects the best algorithm for each specific instance.
    This represents the upper bound of performance achievable by any
    algorithm selector (requires oracle knowledge of true performance).

    Note: This selector "cheats" by using the test performance data.
    """

    PREFIX = "vbs"
    RETURN_TYPE = "single"

    def __init__(
        self,
        budget: float | None = None,
        maximize: bool = False,
        feature_groups: list[str] | None = None,
        **kwargs: Any,
    ) -> None:
        """
        Initialize the VirtualBestSolver.

        Parameters
        ----------
        budget : float or None, default=None
            The budget for the selector.
        maximize : bool, default=False
            Indicates whether to maximize the performance metric.
        feature_groups : list[str] or None, default=None
            Groups of features to be considered.
        **kwargs : Any
            Additional keyword arguments.
        """
        super().__init__(
            budget=budget,
            maximize=maximize,
            feature_groups=feature_groups,
            **kwargs,
        )
        self._performance: pd.DataFrame | None = None

    def _fit(
        self,
        features: pd.DataFrame,
        performance: pd.DataFrame,
        **kwargs: Any,
    ) -> None:
        """
        Store the performance data for oracle predictions.

        Parameters
        ----------
        features : pd.DataFrame
            The input features.
        performance : pd.DataFrame
            The performance data.
        **kwargs : Any
            Additional keyword arguments.
        """
        self._performance = performance

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """
        Predict the best algorithm for each instance (oracle).

        If performance data is provided at prediction time, use it.
        Otherwise, fall back to training performance.

        Parameters
        ----------
        features : pd.DataFrame or None
            The input features.
        performance : pd.DataFrame or None, default=None
            The performance data.

        Returns
        -------
        dict
            Dictionary mapping instance IDs to the best algorithm.
        """
        # Use provided performance or fall back to stored
        perf = performance if performance is not None else self._performance

        if perf is None:
            raise ValueError(
                "VirtualBestSolver requires performance data. "
                "Either provide it at fit time or pass it to predict."
            )

        indices = features.index if features is not None else perf.index

        result: dict[str, list[tuple[str, float]]] = {}
        for instance in indices:
            if instance not in perf.index:
                # Fall back to first algorithm if instance not found
                result[str(instance)] = [
                    (str(self.algorithms[0]), float(self.budget or 0))
                ]
                continue

            instance_perf = perf.loc[instance]

            # Apply PAR10 penalty for comparison
            if self.budget is not None:
                instance_perf_penalized = np.where(
                    instance_perf <= self.budget, instance_perf, self.budget * 10
                )
            else:
                instance_perf_penalized = instance_perf.values

            if self.maximize:
                best_idx = int(np.argmax(instance_perf_penalized))
            else:
                best_idx = int(np.argmin(instance_perf_penalized))

            best_algorithm = str(perf.columns[best_idx])
            result[str(instance)] = [(best_algorithm, float(self.budget or 0))]

        return result

    @staticmethod
    def _define_hyperparameters(
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """
        Define hyperparameters for VirtualBestSolver.

        Returns
        -------
        tuple
            Empty hyperparameters, conditions, and forbiddens.
        """
        return [], [], []

__init__(budget=None, maximize=False, feature_groups=None, **kwargs)

Initialize the VirtualBestSolver.

Parameters

budget : float or None, default=None The budget for the selector. maximize : bool, default=False Indicates whether to maximize the performance metric. feature_groups : list[str] or None, default=None Groups of features to be considered. **kwargs : Any Additional keyword arguments.

Source code in asf/selectors/baselines.py

def __init__(
    self,
    budget: float | None = None,
    maximize: bool = False,
    feature_groups: list[str] | None = None,
    **kwargs: Any,
) -> None:
    """
    Initialize the VirtualBestSolver.

    Parameters
    ----------
    budget : float or None, default=None
        The budget for the selector.
    maximize : bool, default=False
        Indicates whether to maximize the performance metric.
    feature_groups : list[str] or None, default=None
        Groups of features to be considered.
    **kwargs : Any
        Additional keyword arguments.
    """
    super().__init__(
        budget=budget,
        maximize=maximize,
        feature_groups=feature_groups,
        **kwargs,
    )
    self._performance: pd.DataFrame | None = None

VotingSelector

Bases: ConfigurableMixin, AbstractSelector

Voting ensemble for algorithm selectors.

Combines predictions from multiple base selectors using hard voting (majority vote).

Attributes

selectors : list[AbstractSelector] List of base selectors. weights : list[float] or None Weights for each selector's vote. selectors_ : list[AbstractSelector] Fitted selectors after calling fit().

Source code in asf/selectors/ensembles.py

class VotingSelector(ConfigurableMixin, AbstractSelector):
    """Voting ensemble for algorithm selectors.

    Combines predictions from multiple base selectors using hard voting
    (majority vote).

    Attributes
    ----------
    selectors : list[AbstractSelector]
        List of base selectors.
    weights : list[float] or None
        Weights for each selector's vote.
    selectors_ : list[AbstractSelector]
        Fitted selectors after calling fit().
    """

    PREFIX = "voting"
    RETURN_TYPE = "single"

    def __init__(
        self,
        selectors: (
            list[AbstractSelector] | Callable[[], list[AbstractSelector]] | None
        ) = None,
        weights: list[float] | None = None,
        **kwargs: Any,
    ) -> None:
        """Initialize the VotingSelector.

        Parameters
        ----------
        selectors : list[AbstractSelector] or Callable or None
            List of base selectors. Can be instances or a callable
            that returns a list.
        weights : list[float] or None, default=None
            Weights for each selector's vote. If None, equal weights.
        **kwargs : Any
            Additional keyword arguments passed to AbstractSelector.
        """
        super().__init__(**kwargs)

        if callable(selectors) and not isinstance(selectors, list):
            self.selectors = selectors()
        else:
            self.selectors = list(selectors) if selectors is not None else []

        if not self.selectors:
            raise ValueError("selectors list cannot be empty")

        self.weights = weights
        if self.weights is not None and len(self.weights) != len(self.selectors):
            raise ValueError("weights must have same length as selectors")

        self.selectors_: list[AbstractSelector] = []

    def _fit(
        self, features: pd.DataFrame, performance: pd.DataFrame, **kwargs: Any
    ) -> None:
        """Fit the voting ensemble.

        Parameters
        ----------
        features : pd.DataFrame
            Training features.
        performance : pd.DataFrame
            Algorithm performance data.
        **kwargs : Any
            Additional keyword arguments.
        """
        self.selectors_ = []

        for selector in self.selectors:
            fitted = _clone_selector(selector)
            fitted.fit(features, performance, **kwargs)
            self.selectors_.append(fitted)

    def _predict(
        self,
        features: pd.DataFrame | None,
        performance: pd.DataFrame | None = None,
    ) -> dict[str, list[tuple[str, float]]]:
        """Predict using hard voting.

        Parameters
        ----------
        features : pd.DataFrame or None
            Input features for prediction.
        performance : pd.DataFrame or None, default=None
            Performance data (unused).

        Returns
        -------
        dict
            Mapping from instance names to algorithm schedules.
        """
        if features is None:
            raise ValueError("VotingSelector requires features for prediction.")

        if not self.selectors_:
            raise RuntimeError("VotingSelector has not been fitted.")

        weights = self.weights or [1.0] * len(self.selectors_)

        # Collect weighted votes
        all_votes: dict[str, dict[str, float]] = {
            str(idx): {} for idx in features.index
        }

        for selector, weight in zip(self.selectors_, weights):
            preds = selector.predict(features)
            if isinstance(preds, dict):
                for instance, schedule in preds.items():
                    if schedule:
                        algo = str(schedule[0][0])
                        all_votes[str(instance)][algo] = (
                            all_votes[str(instance)].get(algo, 0.0) + weight
                        )

        # Select winner by weighted votes
        result: dict[str, list[tuple[str, float]]] = {}
        for instance in features.index:
            votes = all_votes[str(instance)]
            if votes:
                best_algo = max(votes.keys(), key=lambda a: votes[a])
                result[str(instance)] = [(best_algo, float(self.budget or 0))]
            else:
                result[str(instance)] = []

        return result

    @staticmethod
    def _define_hyperparameters(
        selector_classes: list[type] | None = None,
        **kwargs: Any,
    ) -> tuple[list[Any], list[Any], list[Any]]:
        """Define hyperparameters for VotingSelector.

        Parameters
        ----------
        selector_classes : list[type] or None, default=None
            List of selector classes to include.
        **kwargs : Any
            Additional keyword arguments.

        Returns
        -------
        tuple
            Tuple of (hyperparameters, conditions, forbiddens).
        """
        if not CONFIGSPACE_AVAILABLE:
            return [], [], []

        # VotingSelector takes a list of selectors, not configurable via ConfigSpace
        return [], [], []

__init__(selectors=None, weights=None, **kwargs)

Initialize the VotingSelector.

Parameters

selectors : list[AbstractSelector] or Callable or None List of base selectors. Can be instances or a callable that returns a list. weights : list[float] or None, default=None Weights for each selector's vote. If None, equal weights. **kwargs : Any Additional keyword arguments passed to AbstractSelector.

Source code in asf/selectors/ensembles.py

def __init__(
    self,
    selectors: (
        list[AbstractSelector] | Callable[[], list[AbstractSelector]] | None
    ) = None,
    weights: list[float] | None = None,
    **kwargs: Any,
) -> None:
    """Initialize the VotingSelector.

    Parameters
    ----------
    selectors : list[AbstractSelector] or Callable or None
        List of base selectors. Can be instances or a callable
        that returns a list.
    weights : list[float] or None, default=None
        Weights for each selector's vote. If None, equal weights.
    **kwargs : Any
        Additional keyword arguments passed to AbstractSelector.
    """
    super().__init__(**kwargs)

    if callable(selectors) and not isinstance(selectors, list):
        self.selectors = selectors()
    else:
        self.selectors = list(selectors) if selectors is not None else []

    if not self.selectors:
        raise ValueError("selectors list cannot be empty")

    self.weights = weights
    if self.weights is not None and len(self.weights) != len(self.selectors):
        raise ValueError("weights must have same length as selectors")

    self.selectors_: list[AbstractSelector] = []

tune_selector(X, y, selector_class, features_running_time, algorithm_features=None, selector_kwargs=None, preprocessing_class=None, pre_solving_class=None, feature_selector=None, algorithm_pre_selector=None, max_algorithm_pre_selector=None, budget=None, maximize=False, feature_groups=None, output_dir='./smac_output', smac_metric=running_time_selector_performance, smac_kwargs=None, smac_scenario_kwargs=None, runcount_limit=100, timeout=float('inf'), seed=0, cv=10, groups=None, max_feature_time=None)

Tunes a selector model using SMAC.

Parameters

X : pd.DataFrame Instance feature matrix. y : pd.DataFrame Algorithm performance matrix. selector_class : type or list Selector classes to tune. features_running_time : pd.DataFrame Running times for computing feature groups. algorithm_features : pd.DataFrame or None, optional Features for algorithms. selector_kwargs : dict or None, optional Arguments for selector instantiation. preprocessing_class : list or None, optional List of preprocessor classes. pre_solving_class : list or None, optional List of presolver classes. feature_selector : Any or None, optional Feature selection component. algorithm_pre_selector : Any or None, optional Algorithm pre-selection component. max_algorithm_pre_selector : int or None, optional Constraint for pre-selection. budget : float or None, optional Global cutoff time. maximize : bool, default=False Whether to maximize the performance metric. feature_groups : dict or None, optional Definition of feature groups. output_dir : str, default="./smac_output" SMAC output directory. smac_metric : callable, default=running_time_selector_performance Evaluation metric for SMAC. smac_kwargs : callable or None, optional Additional arguments for SMAC facade. smac_scenario_kwargs : dict or None, optional Additional arguments for SMAC scenario. runcount_limit : int, default=100 Limit for trials. timeout : float, default=inf Wall-clock time limit. seed : int, default=0 Random seed. cv : int, default=10 Number of cross-validation folds. groups : np.ndarray or None, optional Group labels for CV. max_feature_time : float or None, optional Budget per feature group.

Returns

SelectorPipeline Best pipeline found by SMAC.

Source code in asf/selectors/selector_tuner.py

def tune_selector(
    X: pd.DataFrame,
    y: pd.DataFrame,
    selector_class: type[AbstractSelector]
    | list[type[AbstractSelector]]
    | list[tuple[type[AbstractSelector], dict[str, Any]]],
    features_running_time: pd.DataFrame,
    algorithm_features: pd.DataFrame | None = None,
    selector_kwargs: dict[str, Any] | None = None,
    preprocessing_class: list[type[TransformerMixin]] | None = None,
    pre_solving_class: list[type[Any]] | None = None,
    feature_selector: Any | None = None,
    algorithm_pre_selector: Any | None = None,
    max_algorithm_pre_selector: int | None = None,
    budget: float | None = None,
    maximize: bool = False,
    feature_groups: dict[str, Any] | None = None,
    output_dir: str = "./smac_output",
    smac_metric: Callable[
        ..., float | dict[str, float]
    ] = running_time_selector_performance,
    smac_kwargs: Callable[[Scenario], dict[str, Any]] | None = None,
    smac_scenario_kwargs: dict[str, Any] | None = None,
    runcount_limit: int = 100,
    timeout: float = float("inf"),
    seed: int = 0,
    cv: int = 10,
    groups: np.ndarray | None = None,
    max_feature_time: float | None = None,
) -> SelectorPipeline:
    """
    Tunes a selector model using SMAC.

    Parameters
    ----------
    X : pd.DataFrame
        Instance feature matrix.
    y : pd.DataFrame
        Algorithm performance matrix.
    selector_class : type or list
        Selector classes to tune.
    features_running_time : pd.DataFrame
        Running times for computing feature groups.
    algorithm_features : pd.DataFrame or None, optional
        Features for algorithms.
    selector_kwargs : dict or None, optional
        Arguments for selector instantiation.
    preprocessing_class : list or None, optional
        List of preprocessor classes.
    pre_solving_class : list or None, optional
        List of presolver classes.
    feature_selector : Any or None, optional
        Feature selection component.
    algorithm_pre_selector : Any or None, optional
        Algorithm pre-selection component.
    max_algorithm_pre_selector : int or None, optional
        Constraint for pre-selection.
    budget : float or None, optional
        Global cutoff time.
    maximize : bool, default=False
        Whether to maximize the performance metric.
    feature_groups : dict or None, optional
        Definition of feature groups.
    output_dir : str, default="./smac_output"
        SMAC output directory.
    smac_metric : callable, default=running_time_selector_performance
        Evaluation metric for SMAC.
    smac_kwargs : callable or None, optional
        Additional arguments for SMAC facade.
    smac_scenario_kwargs : dict or None, optional
        Additional arguments for SMAC scenario.
    runcount_limit : int, default=100
        Limit for trials.
    timeout : float, default=inf
        Wall-clock time limit.
    seed : int, default=0
        Random seed.
    cv : int, default=10
        Number of cross-validation folds.
    groups : np.ndarray or None, optional
        Group labels for CV.
    max_feature_time : float or None, optional
        Budget per feature group.

    Returns
    -------
    SelectorPipeline
        Best pipeline found by SMAC.
    """
    _logger = logging.getLogger(__name__)

    if not SMAC_AVAILABLE:
        raise RuntimeError("SMAC is not installed.")
    if not CONFIGSPACE_AVAILABLE:
        raise RuntimeError("ConfigSpace is not installed.")

    if pre_solving_class is not None and budget is None:
        raise ValueError("Budget must be provided if using pre-solving.")

    sel_list = selector_class if isinstance(selector_class, list) else [selector_class]
    sel_kwargs = selector_kwargs or {}
    sc_kwargs = smac_scenario_kwargs or {}

    cs = SelectorPipeline.get_configuration_space(
        selector_class=sel_list,
        preprocessing_class=preprocessing_class,
        pre_solving_class=pre_solving_class,
        feature_groups=feature_groups,
        algorithm_pre_selector=algorithm_pre_selector,
        max_feature_time=max_feature_time,
        budget=budget,
        max_algorithm_pre_selector=max_algorithm_pre_selector,
        n_algorithms=y.shape[1] if hasattr(y, "shape") else None,
        **sel_kwargs,
    )

    cs = convert_class_choices_to_categorical(cs)

    scenario = Scenario(
        configspace=cs,
        n_trials=runcount_limit,
        walltime_limit=timeout,
        deterministic=True,
        output_directory=Path(output_dir),
        seed=seed,
        **sc_kwargs,
    )

    def target_function(config: Configuration, seed: int) -> float:
        if groups is not None:
            kfold = GroupKFoldShuffle(n_splits=cv, shuffle=True, random_state=seed)
        else:
            kfold = KFold(n_splits=cv, shuffle=True, random_state=seed)

        scores = []
        for train_idx, test_idx in kfold.split(X, y, groups):
            X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
            y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]
            rt_test = features_running_time.iloc[test_idx]

            pipeline = _create_pipeline(
                config,
                budget,
                maximize,
                sel_kwargs,
                feature_groups,
                max_feature_time=max_feature_time,
            )

            pipeline.fit(X_train, y_train, algorithm_features=algorithm_features)
            y_pred = pipeline.predict(X_test)
            assert isinstance(y_pred, dict)  # Added assertion for y_pred type

            score = smac_metric(y_pred, y_test, budget, rt_test)
            if isinstance(score, dict):
                score = float(np.mean(list(score.values())))
            scores.append(float(score))

        final_score = float(np.mean(scores))
        return -final_score if maximize else final_score

    facade_kwargs = smac_kwargs(scenario) if smac_kwargs is not None else {}
    smac = HyperparameterOptimizationFacade(scenario, target_function, **facade_kwargs)
    best_config = smac.optimize()

    if isinstance(best_config, list):
        best_config = best_config[0]
    assert isinstance(best_config, (dict, Configuration))
    # Cast to handle union type
    final_config = cast(Configuration | dict[str, Any], best_config)
    return _create_pipeline(
        final_config,
        budget,
        maximize,
        sel_kwargs,
        feature_groups,
        max_feature_time=max_feature_time,
    )