AutoML Best Practices

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

def estimate_automl_budget(
    dataset_rows: int,
    num_features: int,
    problem_type: str,
    cv_folds: int = 5,
    algorithms_to_search: int = 20,
    hyperparameter_trials_per_algo: int = 50
) -> dict:
    """
    Estimate AutoML compute budget based on problem characteristics.
    
    Returns estimates in CPU-hours for tabular problems.
    For deep learning/NAS, multiply by ~10-100x and consider GPU requirements.
    """
    
    # Base training time estimates (seconds per 1000 rows)
    BASE_TRAINING_TIMES = {
        'linear': 0.1,        # LogisticRegression, Ridge, Lasso
        'tree': 0.5,          # DecisionTree
        'ensemble': 2.0,      # RandomForest, XGBoost, LightGBM
        'svm': 5.0,           # SVMs (scales poorly with data size)
        'neural': 10.0,       # Small MLPs
    }
    
    # Estimate base training time per model
    avg_training_time_per_1k = sum(BASE_TRAINING_TIMES.values()) / len(BASE_TRAINING_TIMES)
    base_training_seconds = (dataset_rows / 1000) * avg_training_time_per_1k
    
    # Scale by features (roughly sqrt relationship)
    feature_multiplier = (num_features / 50) ** 0.5
    scaled_training_seconds = base_training_seconds * feature_multiplier
    
    # Total evaluations
    total_evaluations = algorithms_to_search * hyperparameter_trials_per_algo * cv_folds
    
    # Total compute in seconds
    total_seconds = scaled_training_seconds * total_evaluations
    
    # Convert to hours
    cpu_hours = total_seconds / 3600
    
    return {
        'estimated_cpu_hours': round(cpu_hours, 1),
        'minimum_budget': round(cpu_hours * 0.25, 1),
        'recommended_budget': round(cpu_hours, 1),
        'extended_budget': round(cpu_hours * 3, 1),
        'total_evaluations': total_evaluations,
        'avg_seconds_per_eval': round(scaled_training_seconds, 2),
    }
 
# Example usage
budget = estimate_automl_budget(
    dataset_rows=100_000,
    num_features=100,
    problem_type='classification',
    cv_folds=5,
    algorithms_to_search=15,
    hyperparameter_trials_per_algo=40
)
 
print(f"Estimated CPU-hours: {budget['estimated_cpu_hours']}")
print(f"Recommended budget: {budget['recommended_budget']} CPU-hours")
print(f"Total evaluations planned: {budget['total_evaluations']}")

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

# Auto-sklearn: Time budget configuration
from autosklearn.classification import AutoSklearnClassifier
 
# Strategy 1: Fixed total time limit
clf_fixed = AutoSklearnClassifier(
    time_left_for_this_task=3600,    # Total time: 1 hour
    per_run_time_limit=300,           # Max 5 minutes per model
    ensemble_size=20,
    ensemble_nbest=10,
    memory_limit=8192,                # 8GB RAM limit
)
 
# Strategy 2: Aggressive early stopping for quick iteration
clf_quick = AutoSklearnClassifier(
    time_left_for_this_task=600,      # 10 minutes total
    per_run_time_limit=60,            # Max 1 minute per model
    initial_configurations_via_metalearning=10,  # Warm start
    ensemble_size=5,
)
 
# Strategy 3: Extended search for production quality
clf_production = AutoSklearnClassifier(
    time_left_for_this_task=14400,    # 4 hours total
    per_run_time_limit=600,           # 10 minutes per model
    ensemble_size=50,
    ensemble_nbest=25,
    resampling_strategy='cv',
    resampling_strategy_arguments={'folds': 5},
)
 
# AutoGluon: Preset-based time allocation
from autogluon.tabular import TabularPredictor
 
# Quick prototyping
predictor_quick = TabularPredictor(label='target').fit(
    train_data,
    time_limit=300,           # 5 minutes
    presets='medium_quality',
)
 
# Production-ready
predictor_prod = TabularPredictor(label='target').fit(
    train_data,
    time_limit=3600,          # 1 hour
    presets='best_quality',
    auto_stack=True,
)
 
# Custom per-model limits with AutoGluon
predictor_custom = TabularPredictor(label='target').fit(
    train_data,
    time_limit=7200,          # 2 hours total
    hyperparameters={
        'GBM': {'num_boost_round': 10000, 'ag_args_fit': {'max_time_limit': 600}},
        'NN_TORCH': {'epochs': 100, 'ag_args_fit': {'max_time_limit': 1200}},
        'CAT': {'iterations': 5000, 'ag_args_fit': {'max_time_limit': 600}},
    },
)

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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182

import numpy as np
from typing import List, Callable
 
class SuccessiveHalving:
    """
    Successive Halving for hyperparameter optimization.
    
    Efficiently allocates budget by progressively eliminating
    poorly performing configurations.
    """
    
    def __init__(
        self,
        total_budget: int,
        num_configs: int,
        reduction_factor: int = 3,
        min_budget_per_config: int = 1
    ):
        """
        Args:
            total_budget: Total compute budget (e.g., epochs, iterations)
            num_configs: Number of initial configurations to evaluate
            reduction_factor: Fraction of configs to keep each round (default: keep 1/3)
            min_budget_per_config: Minimum budget for first round
        """
        self.total_budget = total_budget
        self.num_configs = num_configs
        self.reduction_factor = reduction_factor
        self.min_budget = min_budget_per_config
        
        # Calculate schedule
        self.schedule = self._compute_schedule()
        
    def _compute_schedule(self) -> List[dict]:
        """Compute the successive halving schedule."""
        schedule = []
        n_configs = self.num_configs
        budget = self.min_budget
        
        while n_configs >= 1:
            schedule.append({
                'num_configs': int(n_configs),
                'budget_per_config': int(budget),
            })
            n_configs = n_configs / self.reduction_factor
            budget = budget * self.reduction_factor
            
        return schedule
    
    def run(
        self,
        configs: List[dict],
        evaluate_fn: Callable[[dict, int], float],
    ) -> dict:
        """
        Run successive halving.
        
        Args:
            configs: List of configurations to evaluate
            evaluate_fn: Function(config, budget) -> score (higher is better)
            
        Returns:
            Best configuration found
        """
        active_configs = configs[:self.num_configs]
        
        for round_idx, round_info in enumerate(self.schedule):
            budget = round_info['budget_per_config']
            n_to_keep = round_info['num_configs']
            
            print(f"Round {round_idx}: Evaluating {len(active_configs)} configs "
                  f"with budget {budget}")
            
            # Evaluate all active configurations
            scores = []
            for config in active_configs:
                score = evaluate_fn(config, budget)
                scores.append((score, config))
            
            # Sort by score (descending) and keep top performers
            scores.sort(key=lambda x: x[0], reverse=True)
            active_configs = [cfg for _, cfg in scores[:n_to_keep]]
            
            print(f"  Best score this round: {scores[0][0]:.4f}")
            
        return active_configs[0] if active_configs else None
 
 
class SearchLevelEarlyStopping:
    """
    Early stopping for the overall AutoML search based on
    improvement history.
    """
    
    def __init__(
        self,
        patience: int = 20,
        min_improvement: float = 0.001,
        min_evaluations: int = 50,
    ):
        """
        Args:
            patience: Number of evaluations without improvement before stopping
            min_improvement: Minimum relative improvement to count as progress
            min_evaluations: Minimum evaluations before early stopping is considered
        """
        self.patience = patience
        self.min_improvement = min_improvement
        self.min_evaluations = min_evaluations
        
        self.best_score = float('-inf')
        self.evaluations_since_improvement = 0
        self.total_evaluations = 0
        self.history = []
        
    def update(self, score: float) -> bool:
        """
        Update with new evaluation result.
        
        Returns:
            True if search should continue, False if should stop
        """
        self.total_evaluations += 1
        self.history.append(score)
        
        # Check for improvement
        relative_improvement = (score - self.best_score) / max(abs(self.best_score), 1e-10)
        
        if relative_improvement > self.min_improvement:
            self.best_score = score
            self.evaluations_since_improvement = 0
        else:
            self.evaluations_since_improvement += 1
            
        # Decide whether to continue
        if self.total_evaluations < self.min_evaluations:
            return True  # Keep searching
            
        if self.evaluations_since_improvement >= self.patience:
            print(f"Early stopping: No improvement for {self.patience} evaluations")
            return False  # Stop searching
            
        return True  # Keep searching
    
    def get_statistics(self) -> dict:
        """Get search statistics."""
        return {
            'total_evaluations': self.total_evaluations,
            'best_score': self.best_score,
            'evaluations_since_improvement': self.evaluations_since_improvement,
            'improvement_rate': len([h for h in self.history if h >= self.best_score * 0.99]) / max(len(self.history), 1),
        }
 
 
# Usage example
def example_usage():
    # Successive Halving example
    sh = SuccessiveHalving(
        total_budget=1000,
        num_configs=81,
        reduction_factor=3,
        min_budget_per_config=10
    )
    
    print("Successive Halving Schedule:")
    for i, round_info in enumerate(sh.schedule):
        print(f"  Round {i}: {round_info['num_configs']} configs, "
              f"{round_info['budget_per_config']} budget each")
    
    # Early stopping example
    stopper = SearchLevelEarlyStopping(patience=10, min_evaluations=30)
    
    # Simulate search with diminishing returns
    for i in range(100):
        # Simulate score that improves initially then plateaus
        score = 0.5 + 0.4 * (1 - np.exp(-i/10)) + np.random.normal(0, 0.01)
        
        if not stopper.update(score):
            print(f"Stopped at evaluation {i}")
            break
    
    print(f"Final stats: {stopper.get_statistics()}")

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
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165

import numpy as np
from sklearn.model_selection import cross_val_score
from sklearn.ensemble import RandomForestClassifier
from typing import Dict, Any, Tuple
 
class MultiFidelityOptimizer:
    """
    Multi-fidelity hyperparameter optimization using data subsampling.
    
    Evaluates configurations on progressively larger data subsets,
    eliminating poorly performing configurations early.
    """
    
    def __init__(
        self,
        fidelity_levels: list = [0.1, 0.3, 0.5, 1.0],
        survival_rate: float = 0.5,
        cv_folds: int = 3,
    ):
        """
        Args:
            fidelity_levels: Data fraction at each stage (increasing)
            survival_rate: Fraction of configs that survive each round
            cv_folds: Cross-validation folds for evaluation
        """
        self.fidelity_levels = fidelity_levels
        self.survival_rate = survival_rate
        self.cv_folds = cv_folds
        
    def _subsample(self, X, y, fraction: float):
        """Subsample data to given fraction."""
        n_samples = int(len(X) * fraction)
        indices = np.random.choice(len(X), n_samples, replace=False)
        return X[indices], y[indices]
    
    def _evaluate_config(
        self,
        config: Dict[str, Any],
        X, y,
        fidelity: float
    ) -> float:
        """Evaluate a configuration at given fidelity level."""
        X_sub, y_sub = self._subsample(X, y, fidelity)
        
        model = RandomForestClassifier(**config, random_state=42)
        scores = cross_val_score(model, X_sub, y_sub, cv=self.cv_folds)
        
        return scores.mean()
    
    def optimize(
        self,
        configs: list,
        X, y,
    ) -> Tuple[Dict[str, Any], float]:
        """
        Run multi-fidelity optimization.
        
        Args:
            configs: List of configuration dicts
            X: Feature matrix
            y: Target vector
            
        Returns:
            Tuple of (best_config, best_score)
        """
        active_configs = configs.copy()
        
        for fidelity in self.fidelity_levels:
            n_survive = max(1, int(len(active_configs) * self.survival_rate))
            
            print(f"
Fidelity {fidelity:.0%}: Evaluating {len(active_configs)} configs")
            
            # Evaluate all active configs
            results = []
            for config in active_configs:
                score = self._evaluate_config(config, X, y, fidelity)
                results.append((score, config))
                
            # Sort and keep top performers
            results.sort(key=lambda x: x[0], reverse=True)
            
            # Report top score at this fidelity
            print(f"  Top score at {fidelity:.0%} fidelity: {results[0][0]:.4f}")
            
            # Keep survivors for next round
            active_configs = [cfg for _, cfg in results[:n_survive]]
            
        # Final result is best config at full fidelity
        return active_configs[0], results[0][0]
 
 
# Example: Hyperband implementation sketch
class Hyperband:
    """
    Hyperband: A Novel Bandit-Based Approach to Hyperparameter Optimization.
    
    Combines successive halving with multiple brackets to handle
    the exploration-exploitation tradeoff in budget allocation.
    """
    
    def __init__(
        self,
        max_budget: int,
        reduction_factor: float = 3,
    ):
        self.max_budget = max_budget  # R in paper
        self.eta = reduction_factor    # η in paper
        
        # Number of successive halving brackets
        self.s_max = int(np.log(max_budget) / np.log(reduction_factor))
        self.B = (self.s_max + 1) * max_budget  # Total budget
        
    def get_brackets(self) -> list:
        """
        Generate Hyperband brackets.
        
        Each bracket represents a successive halving run with
        different initial configurations and budgets.
        """
        brackets = []
        
        for s in range(self.s_max, -1, -1):
            # Number of configurations in this bracket
            n = int(np.ceil((self.B / self.max_budget) * 
                           (self.eta ** s) / (s + 1)))
            # Minimum budget per config in this bracket
            r = self.max_budget * (self.eta ** (-s))
            
            bracket = {
                'bracket_id': self.s_max - s,
                'initial_configs': n,
                'min_budget': r,
                'num_rounds': s + 1,
            }
            brackets.append(bracket)
            
        return brackets
    
    def print_schedule(self):
        """Print Hyperband schedule for visualization."""
        print(f"Hyperband Schedule (max_budget={self.max_budget}, η={self.eta})")
        print(f"Total budget per run: {self.B}")
        print()
        
        for bracket in self.get_brackets():
            print(f"Bracket {bracket['bracket_id']}:")
            print(f"  Initial configs: {bracket['initial_configs']}")
            print(f"  Minimum budget: {bracket['min_budget']:.1f}")
            print(f"  Rounds: {bracket['num_rounds']}")
            
            # Show successive halving within bracket
            n = bracket['initial_configs']
            r = bracket['min_budget']
            for round_idx in range(bracket['num_rounds']):
                print(f"    Round {round_idx}: {int(n)} configs × {int(r)} budget")
                n = n / self.eta
                r = r * self.eta
            print()
 
 
# Demo
if __name__ == "__main__":
    hb = Hyperband(max_budget=81, reduction_factor=3)
    hb.print_schedule()

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

"""
Cost-Aware AutoML Configuration Examples
"""
 
# AWS SageMaker Autopilot with cost controls
import sagemaker
from sagemaker.automl.automl import AutoML
 
automl = AutoML(
    role='SageMakerRole',
    target_attribute_name='target',
    output_path='s3://bucket/output',
    
    # Cost control: limit job duration
    max_runtime_per_training_job_in_seconds=600,
    total_job_runtime_in_seconds=7200,
    
    # Cost control: limit candidates
    max_candidates=50,
    
    # Use spot instances for training
    problem_type='BinaryClassification',
    job_objective={'MetricName': 'AUC'},
)
 
# Google Cloud Vertex AI with budget constraints
from google.cloud import aiplatform
 
aiplatform.init(project='my-project', location='us-central1')
 
# Create AutoML job with budget awareness
job = aiplatform.AutoMLTabularTrainingJob(
    display_name='cost-aware-automl',
    optimization_prediction_type='classification',
    optimization_objective='maximize-au-roc',
)
 
model = job.run(
    dataset=dataset,
    target_column='target',
    
    # Budget constraints
    budget_milli_node_hours=1000,  # 1 node-hour maximum
    
    # Disable neural architecture search to reduce cost
    disable_early_stopping=False,
)
 
# H2O AutoML with resource limits
import h2o
from h2o.automl import H2OAutoML
 
h2o.init(max_mem_size='8G')  # Limit memory
 
aml = H2OAutoML(
    max_runtime_secs=3600,         # 1 hour limit
    max_models=20,                  # Model limit
    
    # Exclude expensive algorithm families to control cost
    exclude_algos=['DeepLearning'],  # Skip neural networks
    
    # Use less cross-validation
    nfolds=3,                        # 3-fold instead of default 5
    
    # Early stopping
    stopping_metric='AUC',
    stopping_rounds=5,
    stopping_tolerance=0.001,
)
 
aml.train(x=features, y=target, training_frame=train)
 
 
def estimate_cloud_cost(
    cpu_hours: float,
    gpu_hours: float = 0,
    storage_gb_months: float = 10,
    instance_type: str = 'spot'
) -> dict:
    """
    Estimate cloud costs for AutoML run.
    
    Based on approximate AWS pricing (varies by region/time).
    """
    PRICES = {
        'cpu_ondemand': 0.17,      # per vCPU-hour
        'cpu_spot': 0.05,          # per vCPU-hour
        'cpu_reserved': 0.10,      # per vCPU-hour
        'gpu_ondemand': 3.06,      # per GPU-hour (V100)
        'gpu_spot': 0.90,          # per GPU-hour
        'storage': 0.023,          # per GB-month (S3)
    }
    
    if instance_type == 'spot':
        cpu_cost = cpu_hours * PRICES['cpu_spot']
        gpu_cost = gpu_hours * PRICES['gpu_spot']
    elif instance_type == 'reserved':
        cpu_cost = cpu_hours * PRICES['cpu_reserved']
        gpu_cost = gpu_hours * PRICES['gpu_ondemand'] * 0.6  # ~40% discount
    else:  # on-demand
        cpu_cost = cpu_hours * PRICES['cpu_ondemand']
        gpu_cost = gpu_hours * PRICES['gpu_ondemand']
    
    storage_cost = storage_gb_months * PRICES['storage']
    
    total = cpu_cost + gpu_cost + storage_cost
    
    return {
        'cpu_cost': round(cpu_cost, 2),
        'gpu_cost': round(gpu_cost, 2),
        'storage_cost': round(storage_cost, 2),
        'total_cost': round(total, 2),
        'instance_type': instance_type,
    }
 
# Example
print(estimate_cloud_cost(cpu_hours=100, gpu_hours=10, instance_type='spot'))
# {'cpu_cost': 5.0, 'gpu_cost': 9.0, 'storage_cost': 0.23, 'total_cost': 14.23, 'instance_type': 'spot'}