Grid Search

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
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245

import time
import numpy as np
from typing import Dict, List, Any, Tuple
from dataclasses import dataclass
 
@dataclass
class CostEstimate:
    """Comprehensive cost estimate for a grid search."""
    total_configs: int
    total_trains: int  # configs × cv_folds
    estimated_time_sequential: float  # seconds
    estimated_time_parallel: float  # seconds with parallelization
    parallel_workers: int
    time_per_train: float  # seconds
    
    def __str__(self) -> str:
        def format_time(seconds: float) -> str:
            if seconds < 60:
                return f"{seconds:.1f} seconds"
            elif seconds < 3600:
                return f"{seconds/60:.1f} minutes"
            elif seconds < 86400:
                return f"{seconds/3600:.1f} hours"
            else:
                return f"{seconds/86400:.1f} days"
        
        return f"""
Grid Search Cost Estimate
========================
Configurations: {self.total_configs:,}
Total Training Runs: {self.total_trains:,}
Time per Training: {format_time(self.time_per_train)}
 
Sequential Time: {format_time(self.estimated_time_sequential)}
Parallel Time ({self.parallel_workers} workers): {format_time(self.estimated_time_parallel)}
"""
 
 
def estimate_training_time(
    model_class,
    X, y,
    sample_configs: List[Dict[str, Any]],
    n_samples: int = 3,
    cv_folds: int = 1
) -> Tuple[float, float]:
    """
    Empirically estimate training time by timing sample configurations.
    
    Strategy:
    1. Train model with a few representative configurations
    2. Measure wall-clock time for each
    3. Return mean and std of training times
    
    For heterogeneous training times (e.g., varying n_estimators),
    sample from different regions of the parameter space.
    """
    times = []
    
    for config in sample_configs[:n_samples]:
        start = time.time()
        
        for _ in range(cv_folds):
            model = model_class(**config)
            model.fit(X, y)
        
        elapsed = time.time() - start
        times.append(elapsed / cv_folds)
    
    return np.mean(times), np.std(times)
 
 
def estimate_grid_search_cost(
    param_grid: Dict[str, List[Any]],
    model_class,
    X, y,
    cv_folds: int = 5,
    parallel_workers: int = 1,
    sample_size: int = 5
) -> CostEstimate:
    """
    Comprehensive cost estimation for a planned grid search.
    
    Methodology:
    1. Calculate total configurations from grid dimensions
    2. Sample configurations spanning the parameter space
    3. Time training on sample configurations
    4. Extrapolate to full grid
    
    Accuracy Considerations:
    - Training time may vary across configurations (e.g., more estimators = longer)
    - We sample from different regions to capture this variance
    - Estimate is typically within 20% for homogeneous training times
    """
    import random
    from itertools import product
    
    # Calculate grid size
    param_names = list(param_grid.keys())
    param_values = list(param_grid.values())
    total_configs = 1
    for values in param_values:
        total_configs *= len(values)
    
    # Generate sample configurations for timing
    # Strategy: sample from different regions of the grid
    all_combos = list(product(*param_values))
    
    if len(all_combos) <= sample_size:
        sample_combos = all_combos
    else:
        # Sample evenly across the grid
        indices = np.linspace(0, len(all_combos)-1, sample_size, dtype=int)
        sample_combos = [all_combos[i] for i in indices]
    
    sample_configs = [
        dict(zip(param_names, combo)) for combo in sample_combos
    ]
    
    # Time sample configurations
    print(f"Timing {len(sample_configs)} sample configurations...")
    mean_time, std_time = estimate_training_time(
        model_class, X, y, sample_configs, 
        n_samples=len(sample_configs), cv_folds=1
    )
    
    print(f"Estimated time per train: {mean_time:.2f}s ± {std_time:.2f}s")
    
    # Extrapolate to full grid
    total_trains = total_configs * cv_folds
    sequential_time = total_trains * mean_time
    parallel_time = sequential_time / parallel_workers
    
    return CostEstimate(
        total_configs=total_configs,
        total_trains=total_trains,
        estimated_time_sequential=sequential_time,
        estimated_time_parallel=parallel_time,
        parallel_workers=parallel_workers,
        time_per_train=mean_time
    )
 
 
# ============================================================
# Cost Estimation with Heterogeneous Training Times
# ============================================================
def estimate_heterogeneous_cost(
    param_grid: Dict[str, List[Any]],
    time_model: Dict[str, float],
    cv_folds: int = 5
) -> Dict:
    """
    Estimate cost when training time depends on hyperparameter values.
    
    Many hyperparameters directly affect training time:
    - n_estimators: Linear relationship (2x estimators ≈ 2x time)
    - max_depth: Often linear or log relationship
    - hidden_layers: Quadratic or higher (more parameters)
    
    Parameters:
    -----------
    time_model: Dict mapping hyperparameter values to time multipliers
        Example: {'n_estimators': {100: 1.0, 200: 2.0, 500: 5.0}}
    """
    from itertools import product
    
    param_names = list(param_grid.keys())
    param_values = list(param_grid.values())
    
    total_time = 0.0
    base_time = 1.0  # Normalize to base configuration
    
    for combo in product(*param_values):
        config = dict(zip(param_names, combo))
        
        # Calculate time multiplier for this configuration
        multiplier = 1.0
        for param, value in config.items():
            if param in time_model and value in time_model[param]:
                multiplier *= time_model[param][value]
        
        total_time += base_time * multiplier * cv_folds
    
    return {
        'total_time_units': total_time,
        'num_configs': len(list(product(*param_values))),
        'effective_average_multiplier': total_time / (len(list(product(*param_values))) * cv_folds)
    }
 
 
# ============================================================
# Example: Complete Cost Analysis
# ============================================================
def demonstrate_cost_analysis():
    """Demonstrate comprehensive cost estimation."""
    from sklearn.ensemble import GradientBoostingClassifier
    from sklearn.datasets import make_classification
    import multiprocessing
    
    # Generate sample data
    X, y = make_classification(n_samples=1000, n_features=20, random_state=42)
    
    # Define parameter grid
    param_grid = {
        'n_estimators': [50, 100, 200, 300],
        'max_depth': [3, 5, 7, 10],
        'learning_rate': [0.01, 0.05, 0.1, 0.2],
        'min_samples_split': [2, 5, 10]
    }
    
    # Calculate grid dimensions
    print("=== Grid Analysis ===")
    total_configs = 1
    for param, values in param_grid.items():
        print(f"{param}: {len(values)} values")
        total_configs *= len(values)
    print(f"\nTotal configurations: {total_configs}")
    print(f"With 5-fold CV: {total_configs * 5} training runs")
    
    # Estimate cost
    print("\n=== Cost Estimation ===")
    available_cores = multiprocessing.cpu_count()
    
    estimate = estimate_grid_search_cost(
        param_grid=param_grid,
        model_class=GradientBoostingClassifier,
        X=X, y=y,
        cv_folds=5,
        parallel_workers=available_cores,
        sample_size=5
    )
    
    print(estimate)
    
    # Provide recommendation
    print("=== Recommendation ===")
    if estimate.estimated_time_parallel < 3600:
        print("✓ Grid search is feasible (under 1 hour)")
    elif estimate.estimated_time_parallel < 86400:
        print("⚠ Grid search is expensive but feasible (under 1 day)")
        print("  Consider: Coarser grid, random search, or early stopping")
    else:
        print("✗ Grid search is prohibitively expensive")
        print("  Strongly consider: Random search, Bayesian optimization")
    
    return estimate

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

import numpy as np
from typing import Dict, Any
 
def compute_ensemble_cost_multiplier(config: Dict[str, Any], 
                                     base_n_estimators: int = 100) -> float:
    """
    Compute cost multiplier for tree ensemble methods.
    
    Cost scales approximately linearly with n_estimators
    and sublinearly with max_depth (trees are pruned).
    """
    n_estimators = config.get('n_estimators', base_n_estimators)
    max_depth = config.get('max_depth', 6)
    
    # n_estimators: linear scaling
    estimator_factor = n_estimators / base_n_estimators
    
    # max_depth: sublinear (log-ish) due to pruning
    depth_factor = np.log2(max_depth + 1) / np.log2(7)  # Normalized to depth=6
    
    return estimator_factor * depth_factor
 
 
def compute_nn_cost_multiplier(config: Dict[str, Any]) -> float:
    """
    Compute cost multiplier for neural networks.
    
    Cost dominated by:
    - Forward/backward pass: O(Σ h_i * h_{i+1}) per sample
    - Number of epochs
    - Batch size (smaller = more updates)
    """
    hidden_sizes = config.get('hidden_layer_sizes', (100,))
    epochs = config.get('max_iter', 200)
    
    # Network parameter count (proxy for forward/backward cost)
    # Include input (assumed 100) and output (assumed 1)
    layers = [100] + list(hidden_sizes) + [1]
    param_count = sum(layers[i] * layers[i+1] for i in range(len(layers)-1))
    base_params = 100 * 100 + 100 * 1  # Single 100-unit layer baseline
    
    param_factor = param_count / base_params
    epoch_factor = epochs / 200
    
    return param_factor * epoch_factor
 
 
def compute_svm_cost_multiplier(config: Dict[str, Any], 
                                n_samples: int = 10000) -> float:
    """
    Compute cost multiplier for SVM.
    
    RBF kernel SVM training is O(n²) to O(n³), but heavily
    influenced by C (regularization) and gamma (kernel width).
    
    Larger C = looser solution = faster convergence
    Larger gamma = more complex decision boundary = slower
    """
    C = config.get('C', 1.0)
    gamma = config.get('gamma', 'scale')
    
    # C effect: inverse relationship (larger C = faster)
    c_factor = 1 / np.log10(C + 1)
    
    # gamma effect: complex, but generally larger = slower
    if gamma != 'scale':
        gamma_factor = np.log10(gamma * 1000 + 1)
    else:
        gamma_factor = 1.0
    
    return c_factor * gamma_factor
 
 
def estimate_heterogeneous_grid_time(
    param_grid: Dict[str, list],
    cost_function: callable,
    base_time_seconds: float,
    cv_folds: int = 5
) -> Dict:
    """
    Estimate total grid search time accounting for varying training costs.
    
    Returns detailed breakdown showing which configurations dominate cost.
    """
    from itertools import product
    
    param_names = list(param_grid.keys())
    param_values = list(param_grid.values())
    
    config_costs = []
    
    for combo in product(*param_values):
        config = dict(zip(param_names, combo))
        multiplier = cost_function(config)
        time_estimate = base_time_seconds * multiplier * cv_folds
        
        config_costs.append({
            'config': config,
            'multiplier': multiplier,
            'estimated_time': time_estimate
        })
    
    # Sort by cost
    config_costs.sort(key=lambda x: x['estimated_time'], reverse=True)
    
    total_time = sum(c['estimated_time'] for c in config_costs)
    
    # Find what fraction of time is spent on top 20% of configs
    top_20_pct = int(len(config_costs) * 0.2)
    top_20_time = sum(c['estimated_time'] for c in config_costs[:top_20_pct])
    
    return {
        'total_time_seconds': total_time,
        'num_configs': len(config_costs),
        'average_time_per_config': total_time / len(config_costs),
        'top_20_pct_time_fraction': top_20_time / total_time,
        'most_expensive_configs': config_costs[:5],
        'least_expensive_configs': config_costs[-5:],
    }
 
 
# Example: Analyze gradient boosting grid
def analyze_gb_grid_costs():
    """Demonstrate heterogeneous cost analysis for GradientBoosting."""
    
    param_grid = {
        'n_estimators': [50, 100, 200, 500, 1000],
        'max_depth': [3, 5, 7, 10],
        'learning_rate': [0.01, 0.05, 0.1, 0.2],
    }
    
    base_time = 2.0  # seconds for n_estimators=100, max_depth=6
    
    result = estimate_heterogeneous_grid_time(
        param_grid=param_grid,
        cost_function=compute_ensemble_cost_multiplier,
        base_time_seconds=base_time,
        cv_folds=5
    )
    
    print("=== Heterogeneous Cost Analysis ===")
    print(f"Total configurations: {result['num_configs']}")
    print(f"Total estimated time: {result['total_time_seconds']/3600:.2f} hours")
    print(f"Average time per config: {result['average_time_per_config']:.1f} seconds")
    print(f"\nTop 20% of configs account for {result['top_20_pct_time_fraction']*100:.1f}% of time")
    
    print("\nMost expensive configurations:")
    for c in result['most_expensive_configs']:
        print(f"  {c['config']} -> {c['estimated_time']:.1f}s")
    
    print("\nLeast expensive configurations:")
    for c in result['least_expensive_configs']:
        print(f"  {c['config']} -> {c['estimated_time']:.1f}s")
    
    return result

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
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203

import numpy as np
from typing import Dict, List, Any, Tuple
from itertools import product
 
def coarse_to_fine_grid_search(
    param_ranges: Dict[str, Tuple[float, float, str]],
    objective_fn: callable,
    coarse_points: int = 3,
    fine_points: int = 5,
    zoom_factor: float = 0.5
) -> Dict:
    """
    Two-phase coarse-to-fine grid search.
    
    Phase 1: Coarse grid to identify promising region
    Phase 2: Fine grid zoomed into best region
    
    Parameters:
    -----------
    param_ranges: Dict mapping param names to (min, max, scale)
        scale is 'linear' or 'log'
    zoom_factor: Fraction of original range to use in fine phase
    
    This typically achieves better results than single-phase grid search
    with the same total budget, especially in high dimensions.
    """
    
    def create_grid(ranges: Dict[str, Tuple[float, float, str]], 
                   n_points: int) -> Dict[str, List[float]]:
        grid = {}
        for param, (lo, hi, scale) in ranges.items():
            if scale == 'log':
                grid[param] = list(np.logspace(np.log10(lo), np.log10(hi), n_points))
            else:
                grid[param] = list(np.linspace(lo, hi, n_points))
        return grid
    
    def evaluate_grid(grid: Dict[str, List[float]]) -> List[Dict]:
        results = []
        param_names = list(grid.keys())
        for combo in product(*grid.values()):
            config = dict(zip(param_names, combo))
            score = objective_fn(config)
            results.append({'config': config, 'score': score})
        return results
    
    # Phase 1: Coarse search
    print(f"Phase 1: Coarse grid ({coarse_points} points per dim)")
    coarse_grid = create_grid(param_ranges, coarse_points)
    coarse_results = evaluate_grid(coarse_grid)
    
    # Find best from coarse phase
    best_coarse = min(coarse_results, key=lambda x: x['score'])
    print(f"  Best coarse: {best_coarse['score']:.6f}")
    print(f"  Config: {best_coarse['config']}")
    
    # Phase 2: Fine search around best region
    print(f"\nPhase 2: Fine grid ({fine_points} points per dim)")
    fine_ranges = {}
    for param, (lo, hi, scale) in param_ranges.items():
        center = best_coarse['config'][param]
        range_width = (hi - lo) * zoom_factor
        
        if scale == 'log':
            # In log space
            log_center = np.log10(center)
            log_range = np.log10(hi / lo) * zoom_factor
            new_lo = 10 ** max(np.log10(lo), log_center - log_range/2)
            new_hi = 10 ** min(np.log10(hi), log_center + log_range/2)
        else:
            new_lo = max(lo, center - range_width/2)
            new_hi = min(hi, center + range_width/2)
        
        fine_ranges[param] = (new_lo, new_hi, scale)
    
    fine_grid = create_grid(fine_ranges, fine_points)
    fine_results = evaluate_grid(fine_grid)
    
    # Find best overall
    all_results = coarse_results + fine_results
    best_overall = min(all_results, key=lambda x: x['score'])
    
    total_configs = len(coarse_results) + len(fine_results)
    equivalent_single_phase = int(total_configs ** (1/len(param_ranges)))
    
    print(f"\n=== Summary ===")
    print(f"Total configurations evaluated: {total_configs}")
    print(f"Equivalent single-phase resolution: {equivalent_single_phase} per dim")
    print(f"Actual fine resolution: {fine_points} per dim")
    print(f"Best score: {best_overall['score']:.6f}")
    
    return {
        'best_config': best_overall['config'],
        'best_score': best_overall['score'],
        'coarse_results': coarse_results,
        'fine_results': fine_results,
        'total_configs': total_configs
    }
 
 
def successive_halving_grid_search(
    param_grid: Dict[str, List[Any]],
    objective_fn: callable,
    budget_per_config: int,
    reduction_factor: int = 3,
    min_budget: int = 1
) -> Dict:
    """
    Successive Halving applied to grid search.
    
    Instead of uniform budget allocation, we:
    1. Start with all configurations at minimum budget
    2. Evaluate all, keep top 1/η fraction
    3. Increase budget for survivors, repeat
    
    This is more budget-efficient but trades off completeness—
    good configs with high variance might be eliminated early.
    
    Parameters:
    -----------
    budget_per_config: Resource unit (e.g., epochs) for full evaluation
    reduction_factor: η, fraction eliminated each round
    min_budget: Starting budget per configuration
    """
    from math import ceil, log
    
    # Generate all configurations
    param_names = list(param_grid.keys())
    all_configs = [
        dict(zip(param_names, combo))
        for combo in product(*param_grid.values())
    ]
    
    n_configs = len(all_configs)
    max_rounds = ceil(log(budget_per_config / min_budget, reduction_factor))
    
    print(f"Successive Halving Grid Search")
    print(f"  Configurations: {n_configs}")
    print(f"  Max budget: {budget_per_config}")
    print(f"  Reduction factor: {reduction_factor}")
    print(f"  Rounds: {max_rounds}")
    
    # Initialize: all configs start in the race
    survivors = [(config, None) for config in all_configs]
    current_budget = min_budget
    results_history = []
    
    for round_idx in range(max_rounds + 1):
        n_survivors = len(survivors)
        print(f"\nRound {round_idx}: {n_survivors} configs, budget={current_budget}")
        
        # Evaluate all survivors at current budget
        round_results = []
        for config, prev_score in survivors:
            # objective_fn should accept budget parameter
            score = objective_fn(config, budget=current_budget)
            round_results.append((config, score))
        
        # Sort by score (lower is better)
        round_results.sort(key=lambda x: x[1])
        results_history.append({
            'round': round_idx,
            'budget': current_budget,
            'n_configs': n_survivors,
            'results': round_results
        })
        
        # Keep top 1/η fraction
        n_promote = max(1, n_survivors // reduction_factor)
        survivors = round_results[:n_promote]
        
        print(f"  Best score: {round_results[0][1]:.6f}")
        print(f"  Promoting {n_promote} configs")
        
        # Increase budget for next round
        current_budget = min(current_budget * reduction_factor, budget_per_config)
        
        if n_promote == 1:
            break
    
    # Winner
    best_config, best_score = survivors[0]
    
    # Calculate total budget spent
    total_budget = sum(
        r['n_configs'] * r['budget'] for r in results_history
    )
    uniform_budget = n_configs * budget_per_config
    
    print(f"\n=== Summary ===")
    print(f"Best config: {best_config}")
    print(f"Best score: {best_score:.6f}")
    print(f"Total budget spent: {total_budget}")
    print(f"vs uniform allocation: {uniform_budget}")
    print(f"Savings: {100*(1 - total_budget/uniform_budget):.1f}%")
    
    return {
        'best_config': best_config,
        'best_score': best_score,
        'history': results_history,
        'budget_spent': total_budget,
        'budget_saved_fraction': 1 - total_budget/uniform_budget
    }

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
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227

import numpy as np
from sklearn.model_selection import GridSearchCV, StratifiedShuffleSplit
from typing import Dict, List, Any
 
def grid_search_with_subsampling(
    estimator,
    param_grid: Dict[str, List[Any]],
    X, y,
    subsample_fraction: float = 0.2,
    cv: int = 3,
    final_cv: int = 5,
    n_jobs: int = -1
):
    """
    Two-phase grid search with subsampling.
    
    Phase 1: Grid search on subsampled data (fast)
    Phase 2: Validate top candidates on full data (thorough)
    
    This can provide 5-10x speedup with minimal loss in quality,
    especially for large datasets where subsample is still representative.
    """
    from sklearn.model_selection import cross_val_score
    
    n_samples = len(y)
    n_subsample = int(n_samples * subsample_fraction)
    
    print(f"=== Subsampled Grid Search ===")
    print(f"Full data: {n_samples} samples")
    print(f"Subsample: {n_subsample} samples ({subsample_fraction*100:.0f}%)")
    
    # Create subsample (stratified for classification)
    from sklearn.model_selection import train_test_split
    X_sub, _, y_sub, _ = train_test_split(
        X, y, train_size=subsample_fraction, stratify=y, random_state=42
    )
    
    # Phase 1: Grid search on subsample
    print(f"\nPhase 1: Grid search on subsample ({cv}-fold CV)")
    grid_search = GridSearchCV(
        estimator, param_grid, cv=cv, n_jobs=n_jobs, 
        scoring='accuracy', verbose=1
    )
    grid_search.fit(X_sub, y_sub)
    
    # Get top N candidates
    results_df = pd.DataFrame(grid_search.cv_results_)
    top_n = 5
    top_candidates = results_df.nsmallest(top_n, 'rank_test_score')
    
    print(f"\nTop {top_n} candidates from Phase 1:")
    for idx, row in top_candidates.iterrows():
        print(f"  {row['params']}: {row['mean_test_score']:.4f}")
    
    # Phase 2: Validate on full data
    print(f"\nPhase 2: Validating top candidates on full data ({final_cv}-fold CV)")
    full_data_results = []
    
    for idx, row in top_candidates.iterrows():
        params = row['params']
        estimator_clone = estimator.set_params(**params)
        scores = cross_val_score(
            estimator_clone, X, y, cv=final_cv, scoring='accuracy'
        )
        result = {
            'params': params,
            'subsample_score': row['mean_test_score'],
            'full_data_score': scores.mean(),
            'full_data_std': scores.std()
        }
        full_data_results.append(result)
        print(f"  {params}: {scores.mean():.4f} ± {scores.std():.4f}")
    
    # Select best based on full data
    best_result = max(full_data_results, key=lambda x: x['full_data_score'])
    
    print(f"\n=== Final Result ===")
    print(f"Best params: {best_result['params']}")
    print(f"Subsample score: {best_result['subsample_score']:.4f}")
    print(f"Full data score: {best_result['full_data_score']:.4f}")
    
    return best_result
 
 
def early_stopping_grid_search(
    estimator_class,
    param_grid: Dict[str, List[Any]],
    X, y,
    cv: int = 5,
    patience: int = 10,
    min_improvement: float = 0.001
):
    """
    Grid search with early stopping per configuration.
    
    For iterative models (neural networks, boosting), we can:
    1. Start with small n_iterations/epochs
    2. Monitor validation performance
    3. Stop early if performance plateaus
    
    This dramatically reduces cost for configurations that converge quickly
    or that are clearly suboptimal.
    """
    from sklearn.model_selection import StratifiedKFold
    from itertools import product
    
    param_names = list(param_grid.keys())
    all_configs = list(product(*param_grid.values()))
    
    results = []
    skf = StratifiedKFold(n_splits=cv, shuffle=True, random_state=42)
    
    for combo in all_configs:
        config = dict(zip(param_names, combo))
        
        fold_scores = []
        total_iterations = 0
        
        for fold_idx, (train_idx, val_idx) in enumerate(skf.split(X, y)):
            X_train, X_val = X[train_idx], X[val_idx]
            y_train, y_val = y[train_idx], y[val_idx]
            
            # Initialize model with early stopping capability
            model = estimator_class(
                **config,
                early_stopping=True,
                validation_fraction=0.1,
                n_iter_no_change=patience,
                tol=min_improvement
            )
            
            model.fit(X_train, y_train)
            score = model.score(X_val, y_val)
            fold_scores.append(score)
            
            # Track actual iterations used
            if hasattr(model, 'n_iter_'):
                total_iterations += model.n_iter_
        
        results.append({
            'config': config,
            'mean_score': np.mean(fold_scores),
            'std_score': np.std(fold_scores),
            'avg_iterations': total_iterations / cv
        })
    
    # Sort by score
    results.sort(key=lambda x: x['mean_score'], reverse=True)
    
    return results
 
 
def progressive_resolution_search(
    param_ranges: Dict[str, tuple],
    objective_fn: callable,
    initial_resolution: int = 2,
    max_resolution: int = 6,
    top_k: int = 5
):
    """
    Progressively increase grid resolution, focusing on promising regions.
    
    Start with very coarse grid (2 points per dim), identify top regions,
    then increase resolution only in those regions.
    
    This is a form of adaptive grid refinement that significantly reduces
    total evaluations compared to uniform fine grid.
    """
    from itertools import product
    
    all_results = []
    current_ranges = param_ranges.copy()
    
    for resolution in range(initial_resolution, max_resolution + 1):
        print(f"\n=== Resolution {resolution} ===")
        
        # Create grid at current resolution
        grid = {}
        for param, (lo, hi, scale) in current_ranges.items():
            if scale == 'log':
                grid[param] = list(np.logspace(
                    np.log10(lo), np.log10(hi), resolution
                ))
            else:
                grid[param] = list(np.linspace(lo, hi, resolution))
        
        # Evaluate grid
        round_results = []
        for combo in product(*grid.values()):
            config = dict(zip(grid.keys(), combo))
            score = objective_fn(config)
            round_results.append({'config': config, 'score': score})
            all_results.append({'config': config, 'score': score, 'resolution': resolution})
        
        # Sort and find top k
        round_results.sort(key=lambda x: x['score'])
        top_configs = round_results[:top_k]
        
        print(f"Evaluated {len(round_results)} configs")
        print(f"Best score: {top_configs[0]['score']:.6f}")
        
        # Narrow ranges to neighborhood of top configs
        if resolution < max_resolution:
            new_ranges = {}
            for param, (orig_lo, orig_hi, scale) in param_ranges.items():
                values = [c['config'][param] for c in top_configs]
                if scale == 'log':
                    log_vals = [np.log10(v) for v in values]
                    log_lo, log_hi = min(log_vals), max(log_vals)
                    margin = (log_hi - log_lo) * 0.5
                    new_lo = 10 ** max(np.log10(orig_lo), log_lo - margin)
                    new_hi = 10 ** min(np.log10(orig_hi), log_hi + margin)
                else:
                    margin = (max(values) - min(values)) * 0.5
                    new_lo = max(orig_lo, min(values) - margin)
                    new_hi = min(orig_hi, max(values) + margin)
                new_ranges[param] = (new_lo, new_hi, scale)
            
            current_ranges = new_ranges
    
    best = min(all_results, key=lambda x: x['score'])
    print(f"\n=== Final Best ===")
    print(f"Config: {best['config']}")
    print(f"Score: {best['score']:.6f}")
    print(f"Total evaluations: {len(all_results)}")
    
    return best, all_results