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

import numpy as np
from typing import Tuple, List
 
def random_forest_split(X: np.ndarray, y: np.ndarray, 
                         feature_indices: List[int]) -> Tuple[int, float, float]:
    """
    Random Forest splitting: exhaustive threshold search.
    
    For each candidate feature, evaluate ALL possible split thresholds
    and select the one that maximizes information gain.
    
    Time Complexity: O(max_features × n × log(n)) for sorting-based implementation
    """
    best_feature, best_threshold, best_gain = None, None, -np.inf
    
    for feature_idx in feature_indices:
        feature_values = X[:, feature_idx]
        
        # Sort unique values to find all candidate thresholds
        unique_values = np.unique(feature_values)
        
        # Evaluate all possible split points (midpoints between consecutive values)
        for i in range(len(unique_values) - 1):
            threshold = (unique_values[i] + unique_values[i + 1]) / 2
            gain = compute_information_gain(X, y, feature_idx, threshold)
            
            if gain > best_gain:
                best_gain = gain
                best_feature = feature_idx
                best_threshold = threshold
    
    return best_feature, best_threshold, best_gain
 
 
def extra_trees_split(X: np.ndarray, y: np.ndarray,
                       feature_indices: List[int]) -> Tuple[int, float, float]:
    """
    Extra-Trees splitting: random threshold selection.
    
    For each candidate feature, draw ONE random threshold from the
    feature's range and evaluate only that split.
    
    Time Complexity: O(max_features × n) - no sorting required!
    """
    best_feature, best_threshold, best_gain = None, None, -np.inf
    
    for feature_idx in feature_indices:
        feature_values = X[:, feature_idx]
        
        # Draw a single random threshold from the feature's range
        min_val, max_val = feature_values.min(), feature_values.max()
        
        if min_val == max_val:  # Constant feature, skip
            continue
            
        threshold = np.random.uniform(min_val, max_val)
        gain = compute_information_gain(X, y, feature_idx, threshold)
        
        if gain > best_gain:
            best_gain = gain
            best_feature = feature_idx
            best_threshold = threshold
    
    return best_feature, best_threshold, best_gain
 
 
def compute_information_gain(X, y, feature_idx, threshold) -> float:
    """Compute information gain for a given split (simplified for illustration)."""
    left_mask = X[:, feature_idx] <= threshold
    right_mask = ~left_mask
    
    if left_mask.sum() == 0 or right_mask.sum() == 0:
        return -np.inf  # Invalid split
    
    # Gini impurity computation
    def gini(labels):
        if len(labels) == 0:
            return 0
        proportions = np.bincount(labels) / len(labels)
        return 1 - np.sum(proportions ** 2)
    
    n = len(y)
    n_left, n_right = left_mask.sum(), right_mask.sum()
    
    parent_gini = gini(y)
    child_gini = (n_left / n) * gini(y[left_mask]) + (n_right / n) * gini(y[right_mask])
    
    return parent_gini - child_gini

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

import numpy as np
from sklearn.ensemble import RandomForestClassifier, ExtraTreesClassifier
import time
 
def benchmark_training_time(n_samples_list, n_features=50, n_estimators=100):
    """
    Benchmark and compare training times between 
    Random Forest and Extra-Trees as dataset size grows.
    """
    results = []
    
    for n_samples in n_samples_list:
        # Generate synthetic dataset
        X = np.random.randn(n_samples, n_features)
        y = (X[:, 0] + X[:, 1] > 0).astype(int)
        
        # Time Random Forest
        rf = RandomForestClassifier(n_estimators=n_estimators, random_state=42)
        start = time.time()
        rf.fit(X, y)
        rf_time = time.time() - start
        
        # Time Extra-Trees
        et = ExtraTreesClassifier(n_estimators=n_estimators, random_state=42)
        start = time.time()
        et.fit(X, y)
        et_time = time.time() - start
        
        speedup = rf_time / et_time
        results.append({
            'n_samples': n_samples,
            'rf_time': rf_time,
            'et_time': et_time,
            'speedup': speedup,
            'theoretical_speedup': np.log2(n_samples)
        })
        
        print(f"n={n_samples:>6}: RF={rf_time:.2f}s, ET={et_time:.2f}s, "
              f"Speedup={speedup:.2f}x (theoretical ≈ {np.log2(n_samples):.1f}x)")
    
    return results
 
# Example execution (representative results):
# n= 1000: RF=0.42s, ET=0.18s, Speedup=2.33x (theoretical ≈ 10.0x)
# n= 5000: RF=2.15s, ET=0.76s, Speedup=2.83x (theoretical ≈ 12.3x)
# n=10000: RF=4.82s, ET=1.53s, Speedup=3.15x (theoretical ≈ 13.3x)
# n=50000: RF=28.4s, ET=7.21s, Speedup=3.94x (theoretical ≈ 15.6x)
 
# Note: Observed speedups are often less than theoretical O(log n) due to:
# - Constant factors in implementation
# - Memory access patterns
# - Python/C boundary overhead
# - Cache efficiency differences

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

import numpy as np
from typing import Optional, Tuple, List
from dataclasses import dataclass
 
@dataclass
class TreeNode:
    """Node in an Extra-Tree."""
    feature_index: Optional[int] = None  # Split feature (None for leaves)
    threshold: Optional[float] = None     # Split threshold (None for leaves)
    left: Optional['TreeNode'] = None     # Left child (values <= threshold)
    right: Optional['TreeNode'] = None    # Right child (values > threshold)
    value: Optional[np.ndarray] = None    # Leaf prediction (class distribution or mean)
    is_leaf: bool = False
 
 
class ExtraTree:
    """
    A single Extremely Randomized Tree.
    
    Key differences from standard decision trees:
    1. No bootstrap sampling (use full training set)
    2. Random threshold selection instead of optimal search
    3. Feature subsampling at each node
    """
    
    def __init__(
        self,
        max_features: str = 'sqrt',
        min_samples_split: int = 2,
        min_samples_leaf: int = 1,
        max_depth: Optional[int] = None,
        random_state: Optional[int] = None
    ):
        self.max_features = max_features
        self.min_samples_split = min_samples_split
        self.min_samples_leaf = min_samples_leaf
        self.max_depth = max_depth
        self.random_state = random_state
        self.rng = np.random.RandomState(random_state)
        self.root = None
        self.n_features_ = None
        self.n_classes_ = None
    
    def _get_max_features(self, n_features: int) -> int:
        """Compute number of features to consider at each split."""
        if self.max_features == 'sqrt':
            return max(1, int(np.sqrt(n_features)))
        elif self.max_features == 'log2':
            return max(1, int(np.log2(n_features)))
        elif isinstance(self.max_features, int):
            return min(self.max_features, n_features)
        elif isinstance(self.max_features, float):
            return max(1, int(self.max_features * n_features))
        else:
            return n_features  # 'None' or 'all'
    
    def _compute_impurity(self, y: np.ndarray) -> float:
        """Compute Gini impurity for classification."""
        if len(y) == 0:
            return 0.0
        counts = np.bincount(y, minlength=self.n_classes_)
        proportions = counts / len(y)
        return 1.0 - np.sum(proportions ** 2)
    
    def _compute_split_quality(
        self, 
        y: np.ndarray, 
        y_left: np.ndarray, 
        y_right: np.ndarray
    ) -> float:
        """Compute information gain from a split."""
        n = len(y)
        n_left, n_right = len(y_left), len(y_right)
        
        if n_left < self.min_samples_leaf or n_right < self.min_samples_leaf:
            return -np.inf  # Invalid split
        
        impurity_parent = self._compute_impurity(y)
        impurity_left = self._compute_impurity(y_left)
        impurity_right = self._compute_impurity(y_right)
        
        weighted_child_impurity = (
            (n_left / n) * impurity_left + 
            (n_right / n) * impurity_right
        )
        
        return impurity_parent - weighted_child_impurity
    
    def _pick_random_split(
        self, 
        X: np.ndarray, 
        y: np.ndarray
    ) -> Tuple[Optional[int], Optional[float], float]:
        """
        Extra-Trees core: randomly select features and thresholds.
        
        Algorithm:
        1. Sample k features uniformly at random
        2. For each feature, draw a random threshold from [min, max]
        3. Return the feature/threshold pair with best split quality
        """
        n_samples, n_features = X.shape
        k = self._get_max_features(n_features)
        
        # Randomly select k candidate features
        candidate_features = self.rng.choice(
            n_features, size=min(k, n_features), replace=False
        )
        
        best_feature = None
        best_threshold = None
        best_quality = -np.inf
        
        for feature_idx in candidate_features:
            feature_values = X[:, feature_idx]
            min_val, max_val = feature_values.min(), feature_values.max()
            
            # Skip constant features
            if min_val >= max_val:
                continue
            
            # EXTRA-TREES KEY STEP: Random threshold instead of optimal search
            threshold = self.rng.uniform(min_val, max_val)
            
            # Evaluate this random split
            left_mask = feature_values <= threshold
            right_mask = ~left_mask
            
            quality = self._compute_split_quality(
                y, y[left_mask], y[right_mask]
            )
            
            if quality > best_quality:
                best_quality = quality
                best_feature = feature_idx
                best_threshold = threshold
        
        return best_feature, best_threshold, best_quality
    
    def _build_tree(
        self, 
        X: np.ndarray, 
        y: np.ndarray, 
        depth: int = 0
    ) -> TreeNode:
        """Recursively build the Extra-Tree."""
        n_samples = len(y)
        
        # Stopping conditions
        if (n_samples < self.min_samples_split or
            (self.max_depth is not None and depth >= self.max_depth) or
            len(np.unique(y)) == 1):
            # Create leaf node
            leaf = TreeNode(is_leaf=True)
            leaf.value = np.bincount(y, minlength=self.n_classes_) / n_samples
            return leaf
        
        # Find best random split
        feature_idx, threshold, quality = self._pick_random_split(X, y)
        
        if feature_idx is None or quality <= 0:
            # No valid split found, create leaf
            leaf = TreeNode(is_leaf=True)
            leaf.value = np.bincount(y, minlength=self.n_classes_) / n_samples
            return leaf
        
        # Create internal node and recurse
        left_mask = X[:, feature_idx] <= threshold
        right_mask = ~left_mask
        
        node = TreeNode(
            feature_index=feature_idx,
            threshold=threshold,
            is_leaf=False
        )
        node.left = self._build_tree(X[left_mask], y[left_mask], depth + 1)
        node.right = self._build_tree(X[right_mask], y[right_mask], depth + 1)
        
        return node
    
    def fit(self, X: np.ndarray, y: np.ndarray) -> 'ExtraTree':
        """Fit the Extra-Tree to training data."""
        self.n_features_ = X.shape[1]
        self.n_classes_ = len(np.unique(y))
        # NOTE: Extra-Trees do NOT use bootstrap sampling!
        # They train on the full dataset to reduce bias
        self.root = self._build_tree(X, y)
        return self
    
    def _predict_sample(self, x: np.ndarray, node: TreeNode) -> np.ndarray:
        """Predict class probabilities for a single sample."""
        if node.is_leaf:
            return node.value
        
        if x[node.feature_index] <= node.threshold:
            return self._predict_sample(x, node.left)
        else:
            return self._predict_sample(x, node.right)
    
    def predict_proba(self, X: np.ndarray) -> np.ndarray:
        """Predict class probabilities for all samples."""
        return np.array([self._predict_sample(x, self.root) for x in X])
    
    def predict(self, X: np.ndarray) -> np.ndarray:
        """Predict class labels for all samples."""
        proba = self.predict_proba(X)
        return np.argmax(proba, axis=1)

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

from sklearn.ensemble import ExtraTreesClassifier
from sklearn.model_selection import RandomizedSearchCV
import numpy as np
 
def tune_extra_trees(X_train, y_train, n_iter=50, cv=5):
    """
    Hyperparameter tuning for Extra-Trees using randomized search.
    
    Note: The search space is tailored for Extra-Trees characteristics.
    """
    
    # Parameter distributions optimized for Extra-Trees
    param_distributions = {
        # Fewer trees often suffice due to higher diversity
        'n_estimators': [50, 100, 200, 300, 500],
        
        # Extra-Trees can benefit from considering more features
        # since random thresholds prevent overfitting
        'max_features': ['sqrt', 'log2', 0.3, 0.5, 0.7, None],
        
        # Depth control
        'max_depth': [None, 10, 20, 30, 50],
        
        # Lower values are safer for Extra-Trees
        'min_samples_split': [2, 5, 10],
        'min_samples_leaf': [1, 2, 4],
        
        # Usually keep False, but worth testing
        'bootstrap': [False, True],
    }
    
    base_model = ExtraTreesClassifier(
        random_state=42,
        n_jobs=-1  # Use all cores
    )
    
    search = RandomizedSearchCV(
        base_model,
        param_distributions,
        n_iter=n_iter,
        cv=cv,
        scoring='accuracy',
        random_state=42,
        n_jobs=-1,
        verbose=1
    )
    
    search.fit(X_train, y_train)
    
    print(f"Best CV Score: {search.best_score_:.4f}")
    print(f"Best Parameters: {search.best_params_}")
    
    return search.best_estimator_, search.best_params_
 
# Key insights for Extra-Trees tuning:
# 
# 1. max_features: Try higher values than you would for RF
#    - RF optimal is typically sqrt(d)
#    - ET can often benefit from sqrt(d) to 0.7*d
# 
# 2. n_estimators: You may need fewer trees
#    - Higher tree diversity means faster convergence
#    - 100-200 trees often sufficient (vs 200-500 for RF)
#
# 3. min_samples_split/leaf: Can be more aggressive
#    - Random thresholds provide implicit regularization
#    - Lower values (1-2) often work well

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
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273

import numpy as np
import pandas as pd
from sklearn.ensemble import ExtraTreesClassifier, ExtraTreesRegressor
from sklearn.model_selection import cross_val_score, train_test_split
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.pipeline import Pipeline
from sklearn.metrics import classification_report, mean_squared_error
import joblib
from typing import Union, Dict, Any
 
class ExtraTreesModel:
    """
    Production-ready Extra-Trees wrapper with best practices.
    
    Features:
    - Automatic task detection (classification vs regression)
    - Built-in validation pipeline
    - Feature importance analysis
    - Model persistence
    - Prediction confidence estimation
    """
    
    def __init__(
        self,
        task: str = 'auto',
        n_estimators: int = 200,
        max_features: Union[str, float] = 'sqrt',
        max_depth: int = None,
        min_samples_leaf: int = 1,
        bootstrap: bool = False,
        n_jobs: int = -1,
        random_state: int = 42
    ):
        self.task = task
        self.n_estimators = n_estimators
        self.max_features = max_features
        self.max_depth = max_depth
        self.min_samples_leaf = min_samples_leaf
        self.bootstrap = bootstrap
        self.n_jobs = n_jobs
        self.random_state = random_state
        
        self.model = None
        self.label_encoder = None
        self.feature_names = None
    
    def _detect_task(self, y: np.ndarray) -> str:
        """Auto-detect whether this is classification or regression."""
        if self.task != 'auto':
            return self.task
        
        # Heuristic: if fewer unique values than 10% of samples, classify
        unique_ratio = len(np.unique(y)) / len(y)
        if unique_ratio < 0.1 or len(np.unique(y)) <= 20:
            return 'classification'
        return 'regression'
    
    def _create_model(self, task: str):
        """Instantiate the appropriate Extra-Trees model."""
        params = {
            'n_estimators': self.n_estimators,
            'max_features': self.max_features,
            'max_depth': self.max_depth,
            'min_samples_leaf': self.min_samples_leaf,
            'bootstrap': self.bootstrap,
            'n_jobs': self.n_jobs,
            'random_state': self.random_state
        }
        
        if task == 'classification':
            return ExtraTreesClassifier(**params)
        else:
            return ExtraTreesRegressor(**params)
    
    def fit(
        self, 
        X: Union[np.ndarray, pd.DataFrame], 
        y: Union[np.ndarray, pd.Series],
        feature_names: list = None
    ) -> 'ExtraTreesModel':
        """
        Fit the Extra-Trees model.
        
        Args:
            X: Feature matrix
            y: Target vector
            feature_names: Optional feature names for importance analysis
        """
        # Store feature names
        if isinstance(X, pd.DataFrame):
            self.feature_names = X.columns.tolist()
            X = X.values
        else:
            self.feature_names = feature_names or [f'feature_{i}' for i in range(X.shape[1])]
        
        # Convert target if needed
        if isinstance(y, pd.Series):
            y = y.values
        
        # Detect and set task
        detected_task = self._detect_task(y)
        
        # Encode labels for classification
        if detected_task == 'classification':
            self.label_encoder = LabelEncoder()
            y = self.label_encoder.fit_transform(y)
        
        # Create and fit model
        self.model = self._create_model(detected_task)
        self.model.fit(X, y)
        
        return self
    
    def predict(self, X: Union[np.ndarray, pd.DataFrame]) -> np.ndarray:
        """Generate predictions."""
        if isinstance(X, pd.DataFrame):
            X = X.values
        
        predictions = self.model.predict(X)
        
        # Decode labels for classification
        if self.label_encoder is not None:
            predictions = self.label_encoder.inverse_transform(predictions)
        
        return predictions
    
    def predict_proba(self, X: Union[np.ndarray, pd.DataFrame]) -> np.ndarray:
        """Return prediction probabilities (classification only)."""
        if not hasattr(self.model, 'predict_proba'):
            raise ValueError("Probability predictions only available for classification")
        
        if isinstance(X, pd.DataFrame):
            X = X.values
        
        return self.model.predict_proba(X)
    
    def predict_with_confidence(
        self, 
        X: Union[np.ndarray, pd.DataFrame]
    ) -> Dict[str, np.ndarray]:
        """
        Return predictions with confidence estimates.
        
        For classification: returns predicted class and probability
        For regression: returns prediction and standard deviation across trees
        """
        if isinstance(X, pd.DataFrame):
            X = X.values
        
        if hasattr(self.model, 'predict_proba'):
            # Classification
            proba = self.model.predict_proba(X)
            predictions = np.argmax(proba, axis=1)
            confidence = np.max(proba, axis=1)
            
            if self.label_encoder is not None:
                predictions = self.label_encoder.inverse_transform(predictions)
            
            return {
                'predictions': predictions,
                'confidence': confidence,
                'probabilities': proba
            }
        else:
            # Regression - use individual tree predictions for uncertainty
            tree_predictions = np.array([
                tree.predict(X) for tree in self.model.estimators_
            ])
            
            return {
                'predictions': self.model.predict(X),
                'std': np.std(tree_predictions, axis=0),
                'tree_predictions': tree_predictions
            }
    
    def get_feature_importance(self, top_k: int = None) -> pd.DataFrame:
        """
        Get feature importance scores.
        
        Note: Extra-Trees importance may be more uniform than RF
        due to random thresholds spreading importance across features.
        """
        importance = self.model.feature_importances_
        
        df = pd.DataFrame({
            'feature': self.feature_names,
            'importance': importance
        }).sort_values('importance', ascending=False)
        
        if top_k is not None:
            df = df.head(top_k)
        
        return df
    
    def cross_validate(
        self, 
        X: Union[np.ndarray, pd.DataFrame], 
        y: Union[np.ndarray, pd.Series],
        cv: int = 5
    ) -> Dict[str, float]:
        """Perform cross-validation and return scores."""
        if isinstance(X, pd.DataFrame):
            X = X.values
        if isinstance(y, pd.Series):
            y = y.values
        
        # Create fresh model for CV
        task = self._detect_task(y)
        model = self._create_model(task)
        
        if task == 'classification':
            scoring = 'accuracy'
        else:
            scoring = 'neg_mean_squared_error'
        
        scores = cross_val_score(model, X, y, cv=cv, scoring=scoring)
        
        return {
            'mean': scores.mean(),
            'std': scores.std(),
            'scores': scores
        }
    
    def save(self, filepath: str):
        """Save model to disk."""
        joblib.dump({
            'model': self.model,
            'label_encoder': self.label_encoder,
            'feature_names': self.feature_names
        }, filepath)
    
    @classmethod
    def load(cls, filepath: str) -> 'ExtraTreesModel':
        """Load model from disk."""
        data = joblib.load(filepath)
        instance = cls()
        instance.model = data['model']
        instance.label_encoder = data['label_encoder']
        instance.feature_names = data['feature_names']
        return instance
 
 
# Example usage
if __name__ == "__main__":
    from sklearn.datasets import make_classification
    
    # Generate sample data
    X, y = make_classification(
        n_samples=5000, n_features=20, n_informative=10,
        n_redundant=5, random_state=42
    )
    
    # Split data
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    
    # Create and train model
    et_model = ExtraTreesModel(n_estimators=100, max_features=0.5)
    et_model.fit(X_train, y_train)
    
    # Cross-validate
    cv_results = et_model.cross_validate(X_train, y_train)
    print(f"CV Accuracy: {cv_results['mean']:.4f} (+/- {cv_results['std']:.4f})")
    
    # Predictions with confidence
    results = et_model.predict_with_confidence(X_test)
    print(f"\nTest Accuracy: {(results['predictions'] == y_test).mean():.4f}")
    print(f"Mean Confidence: {results['confidence'].mean():.4f}")
    
    # Feature importance
    importance = et_model.get_feature_importance(top_k=10)
    print(f"\nTop 10 Features:\n{importance}")