Automated Model Selection

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

import numpy as np
from sklearn.model_selection import cross_val_score
from typing import Dict, Any, List, Tuple
 
class CASHProblem:
    """
    Formal representation of the CASH problem.
    
    Encapsulates the combined algorithm selection and hyperparameter
    optimization as a black-box optimization problem.
    """
    
    def __init__(self, algorithms: Dict[str, Any], 
                 hyperparameter_spaces: Dict[str, Dict],
                 X, y, cv=5, scoring='accuracy'):
        """
        Parameters:
            algorithms: Dict mapping algorithm names to (estimator_class, default_params)
            hyperparameter_spaces: Dict mapping algorithm names to their param distributions
            X: Feature matrix
            y: Target vector
            cv: Cross-validation folds
            scoring: Scoring metric
        """
        self.algorithms = algorithms
        self.hyperparameter_spaces = hyperparameter_spaces
        self.X = X
        self.y = y
        self.cv = cv
        self.scoring = scoring
        
        # Track evaluations
        self.evaluation_history = []
        
    def evaluate_configuration(self, algorithm_name: str, 
                                hyperparameters: Dict[str, Any]) -> float:
        """
        Evaluate a single configuration (algorithm + hyperparameters).
        
        This is the black-box objective function that CASH optimizers
        query repeatedly.
        
        Parameters:
            algorithm_name: Which algorithm to use
            hyperparameters: Hyperparameter settings for that algorithm
            
        Returns:
            Negative cross-validation score (we minimize)
        """
        if algorithm_name not in self.algorithms:
            raise ValueError(f"Unknown algorithm: {algorithm_name}")
        
        estimator_class, default_params = self.algorithms[algorithm_name]
        
        # Merge default params with provided hyperparameters
        params = {**default_params, **hyperparameters}
        
        try:
            # Instantiate and evaluate
            estimator = estimator_class(**params)
            scores = cross_val_score(estimator, self.X, self.y, 
                                      cv=self.cv, scoring=self.scoring)
            score = scores.mean()
            
            # Track evaluation
            self.evaluation_history.append({
                'algorithm': algorithm_name,
                'hyperparameters': hyperparameters,
                'score': score,
                'std': scores.std()
            })
            
            return -score  # Negative because we minimize
            
        except Exception as e:
            # Invalid configuration - return worst possible score
            return 0.0  # Return 0 accuracy (worst)
    
    def get_configuration_space_size(self) -> Dict[str, int]:
        """
        Estimate the size of the configuration space.
        
        Returns approximate number of configurations per algorithm.
        """
        sizes = {}
        for algo_name, param_space in self.hyperparameter_spaces.items():
            # Very rough estimate: product of parameter cardinalities
            size = 1
            for param_name, param_dist in param_space.items():
                if hasattr(param_dist, '__len__'):
                    size *= len(param_dist)
                else:
                    size *= 100  # Continuous parameter - discretize
            sizes[algo_name] = size
        return sizes
    
    def get_best_configuration(self) -> Tuple[str, Dict, float]:
        """
        Get the best configuration found so far.
        """
        if not self.evaluation_history:
            return None, None, None
        
        best = max(self.evaluation_history, key=lambda x: x['score'])
        return best['algorithm'], best['hyperparameters'], best['score']

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117

from ConfigSpace import ConfigurationSpace, Configuration
from ConfigSpace import CategoricalHyperparameter, UniformIntegerHyperparameter
from ConfigSpace import UniformFloatHyperparameter, EqualsCondition, InCondition
 
def create_cash_configuration_space():
    """
    Create a hierarchical configuration space for CASH.
    
    Uses ConfigSpace library (used by Auto-sklearn) to properly
    represent conditional hyperparameters.
    """
    cs = ConfigurationSpace()
    
    # Root: Algorithm choice
    algorithm = CategoricalHyperparameter(
        "algorithm", 
        ["RandomForest", "SVM", "NeuralNetwork", "GradientBoosting"]
    )
    cs.add_hyperparameter(algorithm)
    
    # ================= Random Forest =================
    rf_n_estimators = UniformIntegerHyperparameter(
        "rf_n_estimators", lower=10, upper=500, default_value=100
    )
    rf_max_depth = UniformIntegerHyperparameter(
        "rf_max_depth", lower=1, upper=50, default_value=10
    )
    rf_min_samples_split = UniformIntegerHyperparameter(
        "rf_min_samples_split", lower=2, upper=20, default_value=2
    )
    
    cs.add_hyperparameters([rf_n_estimators, rf_max_depth, rf_min_samples_split])
    
    # Conditionals: These only matter if algorithm = RandomForest
    cs.add_condition(EqualsCondition(rf_n_estimators, algorithm, "RandomForest"))
    cs.add_condition(EqualsCondition(rf_max_depth, algorithm, "RandomForest"))
    cs.add_condition(EqualsCondition(rf_min_samples_split, algorithm, "RandomForest"))
    
    # ================= SVM =================
    svm_kernel = CategoricalHyperparameter(
        "svm_kernel", ["linear", "rbf", "poly"], default_value="rbf"
    )
    svm_C = UniformFloatHyperparameter(
        "svm_C", lower=1e-3, upper=1e3, log=True, default_value=1.0
    )
    svm_gamma = UniformFloatHyperparameter(
        "svm_gamma", lower=1e-4, upper=1e1, log=True, default_value=0.1
    )
    svm_degree = UniformIntegerHyperparameter(
        "svm_degree", lower=2, upper=5, default_value=3
    )
    svm_coef0 = UniformFloatHyperparameter(
        "svm_coef0", lower=0.0, upper=10.0, default_value=0.0
    )
    
    cs.add_hyperparameters([svm_kernel, svm_C, svm_gamma, svm_degree, svm_coef0])
    
    # SVM hyperparams conditional on algorithm = SVM
    cs.add_condition(EqualsCondition(svm_kernel, algorithm, "SVM"))
    cs.add_condition(EqualsCondition(svm_C, algorithm, "SVM"))
    
    # gamma only matters for RBF kernel
    cs.add_condition(EqualsCondition(svm_gamma, svm_kernel, "rbf"))
    
    # degree and coef0 only matter for polynomial kernel
    cs.add_condition(EqualsCondition(svm_degree, svm_kernel, "poly"))
    cs.add_condition(EqualsCondition(svm_coef0, svm_kernel, "poly"))
    
    # ================= Neural Network =================
    nn_hidden_layers = UniformIntegerHyperparameter(
        "nn_hidden_layers", lower=1, upper=5, default_value=2
    )
    nn_hidden_size = UniformIntegerHyperparameter(
        "nn_hidden_size", lower=16, upper=512, default_value=64
    )
    nn_activation = CategoricalHyperparameter(
        "nn_activation", ["relu", "tanh", "sigmoid"], default_value="relu"
    )
    nn_learning_rate = UniformFloatHyperparameter(
        "nn_learning_rate", lower=1e-5, upper=1e-1, log=True, default_value=1e-3
    )
    
    cs.add_hyperparameters([nn_hidden_layers, nn_hidden_size, 
                           nn_activation, nn_learning_rate])
    
    # Neural network params conditional on algorithm = NeuralNetwork
    for hp in [nn_hidden_layers, nn_hidden_size, nn_activation, nn_learning_rate]:
        cs.add_condition(EqualsCondition(hp, algorithm, "NeuralNetwork"))
    
    # ================= Gradient Boosting =================
    gb_n_estimators = UniformIntegerHyperparameter(
        "gb_n_estimators", lower=50, upper=500, default_value=100
    )
    gb_learning_rate = UniformFloatHyperparameter(
        "gb_learning_rate", lower=0.01, upper=1.0, log=True, default_value=0.1
    )
    gb_max_depth = UniformIntegerHyperparameter(
        "gb_max_depth", lower=1, upper=15, default_value=3
    )
    
    cs.add_hyperparameters([gb_n_estimators, gb_learning_rate, gb_max_depth])
    
    for hp in [gb_n_estimators, gb_learning_rate, gb_max_depth]:
        cs.add_condition(EqualsCondition(hp, algorithm, "GradientBoosting"))
    
    return cs
 
# Usage
cs = create_cash_configuration_space()
print(f"Configuration space has {len(cs)} hyperparameters")
print(f"Conditionals ensure only relevant params are active")
 
# Sample valid configurations
for i in range(3):
    config = cs.sample_configuration()
    print(f"
Sampled: {dict(config)}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

import numpy as np
from typing import Callable, Dict, Any, List, Tuple
from concurrent.futures import ProcessPoolExecutor
 
class RandomSearchCASH:
    """
    Random search for the CASH problem.
    
    Despite its simplicity, random search is a strong baseline
    that's hard to beat with small evaluation budgets.
    """
    
    def __init__(self, config_space, n_workers=1, random_state=42):
        """
        Parameters:
            config_space: ConfigSpace object defining the search space
            n_workers: Number of parallel workers
            random_state: Random seed for reproducibility
        """
        self.config_space = config_space
        self.n_workers = n_workers
        self.rng = np.random.RandomState(random_state)
        
    def optimize(self, objective_func: Callable, 
                 n_evaluations: int) -> Tuple[Dict, float]:
        """
        Run random search.
        
        Parameters:
            objective_func: Function that takes a config dict, returns loss
            n_evaluations: Number of configurations to try
            
        Returns:
            (best_config, best_loss)
        """
        # Sample all configurations upfront
        configs = [self.config_space.sample_configuration() 
                   for _ in range(n_evaluations)]
        
        # Evaluate (potentially in parallel)
        if self.n_workers == 1:
            results = [objective_func(dict(config)) for config in configs]
        else:
            with ProcessPoolExecutor(max_workers=self.n_workers) as executor:
                results = list(executor.map(
                    lambda c: objective_func(dict(c)), configs
                ))
        
        # Find best
        best_idx = np.argmin(results)
        
        return dict(configs[best_idx]), results[best_idx]
    
    def optimize_with_budget(self, objective_func: Callable,
                              time_budget_seconds: float,
                              min_evaluations: int = 10) -> Tuple[Dict, float]:
        """
        Run random search with a time budget instead of fixed evaluations.
        
        Useful for fair comparison with adaptive methods.
        """
        import time
        
        best_config = None
        best_loss = float('inf')
        n_evaluated = 0
        start_time = time.time()
        
        while True:
            # Check budget
            elapsed = time.time() - start_time
            if elapsed >= time_budget_seconds and n_evaluated >= min_evaluations:
                break
            
            # Sample and evaluate
            config = self.config_space.sample_configuration()
            loss = objective_func(dict(config))
            n_evaluated += 1
            
            if loss < best_loss:
                best_loss = loss
                best_config = dict(config)
        
        print(f"Random search: {n_evaluated} evaluations in {elapsed:.1f}s")
        return best_config, best_loss

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103

from smac import HyperparameterOptimizationFacade, Scenario
from ConfigSpace import ConfigurationSpace
import numpy as np
 
def run_smac_optimization(config_space: ConfigurationSpace,
                           objective_func,
                           n_trials: int = 100,
                           n_workers: int = 1,
                           seed: int = 42):
    """
    Run SMAC (Sequential Model-based Algorithm Configuration) for CASH.
    
    SMAC is the state-of-the-art Bayesian optimizer for hierarchical,
    conditional configuration spaces.
    
    Parameters:
        config_space: ConfigSpace defining algorithms and hyperparameters
        objective_func: Function mapping config dict to loss (minimize)
        n_trials: Number of configurations to evaluate
        n_workers: Parallel workers
        seed: Random seed
        
    Returns:
        Best configuration found
    """
    
    # Define the scenario
    scenario = Scenario(
        configspace=config_space,
        n_trials=n_trials,       # Total evaluations
        n_workers=n_workers,      # Parallelism
        seed=seed,
        deterministic=True
    )
    
    # Wrap objective for SMAC interface
    def smac_objective(config, seed=None):
        """SMAC-compatible objective wrapper."""
        return objective_func(dict(config))
    
    # Create SMAC optimizer
    smac = HyperparameterOptimizationFacade(
        scenario=scenario,
        target_function=smac_objective,
        overwrite=True  # Overwrite previous runs
    )
    
    # Run optimization
    incumbent = smac.optimize()
    
    return dict(incumbent)
 
 
# Example usage with Auto-sklearn style CASH problem
def example_cash_with_smac():
    """Demonstrate SMAC on a CASH problem."""
    from sklearn.datasets import make_classification
    from sklearn.model_selection import cross_val_score
    from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
    from sklearn.svm import SVC
    
    # Generate dataset
    X, y = make_classification(n_samples=500, n_features=20, 
                               n_informative=10, random_state=42)
    
    # Build configuration space (simplified)
    cs = create_cash_configuration_space()  # From previous example
    
    # Define objective
    def objective(config):
        algo = config['algorithm']
        
        if algo == 'RandomForest':
            clf = RandomForestClassifier(
                n_estimators=config.get('rf_n_estimators', 100),
                max_depth=config.get('rf_max_depth', None),
                random_state=42
            )
        elif algo == 'GradientBoosting':
            clf = GradientBoostingClassifier(
                n_estimators=config.get('gb_n_estimators', 100),
                learning_rate=config.get('gb_learning_rate', 0.1),
                max_depth=config.get('gb_max_depth', 3),
                random_state=42
            )
        elif algo == 'SVM':
            clf = SVC(
                kernel=config.get('svm_kernel', 'rbf'),
                C=config.get('svm_C', 1.0),
                gamma=config.get('svm_gamma', 'scale'),
                random_state=42
            )
        else:
            raise ValueError(f"Unknown algorithm: {algo}")
        
        # Return negative accuracy (since we minimize)
        scores = cross_val_score(clf, X, y, cv=5, scoring='accuracy')
        return -scores.mean()
    
    # Run SMAC
    best_config = run_smac_optimization(cs, objective, n_trials=50)
    print(f"Best configuration: {best_config}")
    print(f"Best score: {-objective(best_config):.4f}")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118

import numpy as np
import time
from typing import Dict
 
def analyze_cash_complexity(algorithms: Dict, 
                             hyperparameter_ranges: Dict,
                             dataset_size: int,
                             cv_folds: int = 5):
    """
    Analyze the computational complexity of a CASH problem.
    
    Provides estimates of:
    - Configuration space size
    - Estimated time per evaluation
    - Total search time for different budgets
    """
    
    # Configuration space size
    total_configs = 0
    space_details = {}
    
    for algo_name, param_ranges in hyperparameter_ranges.items():
        algo_configs = 1
        for param_name, param_range in param_ranges.items():
            if isinstance(param_range, list):
                algo_configs *= len(param_range)
            elif isinstance(param_range, tuple) and len(param_range) == 2:
                # Continuous: estimate 100 effective values
                algo_configs *= 100
        
        space_details[algo_name] = algo_configs
        total_configs += algo_configs
    
    print("=== CASH Complexity Analysis ===
")
    print(f"Number of algorithms: {len(algorithms)}")
    print(f"
Configurations per algorithm:")
    for algo, count in space_details.items():
        print(f"  {algo}: ~{count:,} configurations")
    print(f"
Total configuration space: ~{total_configs:,} configurations")
    
    # Estimate time per evaluation
    # Very rough: assume O(n * d) training, where d ~ n/10
    d_approx = max(10, dataset_size // 10)
    
    time_estimates = {
        'RandomForest': dataset_size * d_approx * 100 / 1e8,  # ~100 trees
        'SVM': (dataset_size ** 2) * d_approx / 1e9,          # Kernel computation
        'GradientBoosting': dataset_size * d_approx * 100 / 1e8,
        'NeuralNetwork': dataset_size * d_approx * 1000 / 1e8, # ~1000 iterations
    }
    
    avg_time = np.mean([time_estimates.get(algo, 1.0) 
                        for algo in algorithms.keys()])
    
    print(f"
=== Time Estimates (dataset size={dataset_size}) ===")
    print(f"Estimated time per evaluation: ~{avg_time * cv_folds:.1f} seconds")
    print(f"  (with {cv_folds}-fold CV)")
    
    # Total time for different evaluation budgets
    budgets = [50, 100, 500, 1000]
    print(f"
=== Total Search Time Estimates ===")
    for budget in budgets:
        total_time = budget * avg_time * cv_folds
        hours = total_time / 3600
        print(f"  {budget} evaluations: ~{hours:.1f} hours")
    
    # Coverage
    print(f"
=== Search Coverage ===")
    for budget in budgets:
        coverage = budget / total_configs * 100
        print(f"  {budget} evaluations: {coverage:.4f}% of space explored")
    
    return {
        'total_configs': total_configs,
        'avg_time_per_eval': avg_time * cv_folds,
        'space_details': space_details
    }
 
# Example usage
algorithms = {
    'RandomForest': None,
    'SVM': None,
    'GradientBoosting': None,
    'NeuralNetwork': None
}
 
hyperparameter_ranges = {
    'RandomForest': {
        'n_estimators': (10, 500),
        'max_depth': (1, 50),
        'min_samples_split': (2, 20)
    },
    'SVM': {
        'kernel': ['linear', 'rbf', 'poly'],
        'C': (1e-3, 1e3),
        'gamma': (1e-4, 1e1)
    },
    'GradientBoosting': {
        'n_estimators': (50, 500),
        'learning_rate': (0.01, 1.0),
        'max_depth': (1, 15)
    },
    'NeuralNetwork': {
        'hidden_layers': (1, 5),
        'hidden_size': (16, 512),
        'learning_rate': (1e-5, 1e-1),
        'activation': ['relu', 'tanh', 'sigmoid']
    }
}
 
analysis = analyze_cash_complexity(algorithms, hyperparameter_ranges, 
                                   dataset_size=10000)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

import autosklearn.classification
from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import train_test_split
import numpy as np
 
def run_autosklearn_cash(time_budget_minutes=10):
    """
    Run Auto-sklearn, demonstrating CASH in action.
    
    Auto-sklearn internally solves the CASH problem using SMAC
    with meta-learning warm-starting.
    """
    
    # Load dataset
    X, y = load_breast_cancer(return_X_y=True)
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    
    print(f"Dataset: {X_train.shape[0]} training samples, "
          f"{X_train.shape[1]} features")
    print(f"Time budget: {time_budget_minutes} minutes")
    
    # Initialize Auto-sklearn
    # This sets up the CASH problem internally
    automl = autosklearn.classification.AutoSklearnClassifier(
        time_left_for_this_task=time_budget_minutes * 60,  # Total seconds
        per_run_time_limit=60,  # Max seconds per configuration
        memory_limit=4096,      # MB
        
        # CASH settings
        include={
            'classifier': [
                'random_forest',
                'gradient_boosting', 
                'adaboost',
                'extra_trees',
                'libsvm_svc',
                'liblinear_svc',
                'k_nearest_neighbors',
                'decision_tree',
            ]
        },
        
        # Meta-learning for warm-starting
        initial_configurations_via_metalearning=25,
        
        # Ensembling
        ensemble_kwargs={'ensemble_size': 50},
        
        # Reproducibility
        seed=42,
        n_jobs=1
    )
    
    # Fit - this runs SMAC to solve the CASH problem
    print("
Starting Auto-sklearn (CASH optimization)...")
    automl.fit(X_train, y_train)
    
    # Results
    print("
=== CASH Results ===")
    print(f"Test accuracy: {automl.score(X_test, y_test):.4f}")
    
    # Show examined configurations
    print(f"
Configurations examined: {len(automl.cv_results_['mean_test_score'])}")
    
    # Show best configuration (the CASH solution)
    print("
Best configuration found:")
    print(automl.show_models())
    
    # Leaderboard
    print("
Top 5 configurations:")
    leaderboard = automl.leaderboard(detailed=True, 
                                      ensemble_size=False)[['rank', 'type', 'cost']]
    print(leaderboard.head())
    
    return automl
 
# Run CASH with Auto-sklearn
# automl = run_autosklearn_cash(time_budget_minutes=10)