LightGBM: Light Gradient Boosting Machine

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

import numpy as np
from typing import Tuple, List
 
class FeatureBinner:
    """
    Implements feature discretization strategies for histogram-based GBDT.
    """
    
    def __init__(self, max_bins: int = 256, strategy: str = 'quantile'):
        """
        Initialize binner.
        
        Parameters:
        -----------
        max_bins : int
            Maximum number of bins per feature
        strategy : str
            'quantile' for equal-frequency, 'uniform' for equal-width
        """
        self.max_bins = max_bins
        self.strategy = strategy
        self.bin_edges_ = None
        
    def fit(self, X: np.ndarray) -> 'FeatureBinner':
        """
        Learn bin boundaries from data.
        
        Parameters:
        -----------
        X : array of shape (n_samples, n_features)
        """
        n_features = X.shape[1]
        self.bin_edges_ = []
        
        for f in range(n_features):
            values = X[:, f]
            unique_values = np.unique(values)
            
            # If fewer unique values than max_bins, use unique values as boundaries
            if len(unique_values) <= self.max_bins:
                edges = unique_values
            elif self.strategy == 'quantile':
                # Equal-frequency binning
                percentiles = np.linspace(0, 100, self.max_bins + 1)
                edges = np.percentile(values, percentiles[1:-1])
                edges = np.unique(edges)  # Remove duplicates
            elif self.strategy == 'uniform':
                # Equal-width binning
                edges = np.linspace(values.min(), values.max(), self.max_bins + 1)[1:-1]
            else:
                raise ValueError(f"Unknown strategy: {self.strategy}")
                
            self.bin_edges_.append(edges)
        
        return self
    
    def transform(self, X: np.ndarray) -> np.ndarray:
        """
        Transform features to bin indices.
        
        Returns:
        --------
        X_binned : array of uint8/uint16 bin indices
        """
        n_samples, n_features = X.shape
        X_binned = np.zeros((n_samples, n_features), dtype=np.uint8)
        
        for f in range(n_features):
            # np.searchsorted gives the bin index for each value
            X_binned[:, f] = np.searchsorted(self.bin_edges_[f], X[:, f])
        
        return X_binned
    
    def get_n_bins(self) -> List[int]:
        """Get number of bins for each feature."""
        return [len(edges) + 1 for edges in self.bin_edges_]
 
 
# Demonstration
def demonstrate_binning():
    np.random.seed(42)
    
    # Create features with different distributions
    n_samples = 10000
    
    # Normal distribution
    feat_normal = np.random.randn(n_samples)
    
    # Skewed distribution (exponential)
    feat_skewed = np.random.exponential(scale=2, size=n_samples)
    
    # Uniform distribution
    feat_uniform = np.random.uniform(0, 100, n_samples)
    
    # Categorical-like (few unique values)
    feat_categorical = np.random.choice([1, 2, 3, 5, 8, 13, 21], n_samples)
    
    X = np.column_stack([feat_normal, feat_skewed, feat_uniform, feat_categorical])
    
    print("Feature Distributions:")
    print(f"  Normal:      min={feat_normal.min():.2f}, max={feat_normal.max():.2f}")
    print(f"  Skewed:      min={feat_skewed.min():.2f}, max={feat_skewed.max():.2f}")
    print(f"  Uniform:     min={feat_uniform.min():.2f}, max={feat_uniform.max():.2f}")
    print(f"  Categorical: unique values = {len(np.unique(feat_categorical))}")
    
    # Quantile-based binning
    binner_quantile = FeatureBinner(max_bins=16, strategy='quantile')
    binner_quantile.fit(X)
    X_binned_quantile = binner_quantile.transform(X)
    
    # Uniform binning
    binner_uniform = FeatureBinner(max_bins=16, strategy='uniform')
    binner_uniform.fit(X)
    X_binned_uniform = binner_uniform.transform(X)
    
    print(f"\nQuantile Binning (16 bins):")
    for i, name in enumerate(['Normal', 'Skewed', 'Uniform', 'Categorical']):
        n_bins = binner_quantile.get_n_bins()[i]
        bin_counts = np.bincount(X_binned_quantile[:, i], minlength=n_bins)
        print(f"  {name}: {n_bins} bins, samples per bin: min={bin_counts.min()}, max={bin_counts.max()}")
    
    print(f"\nUniform Binning (16 bins):")
    for i, name in enumerate(['Normal', 'Skewed', 'Uniform', 'Categorical']):
        n_bins = binner_uniform.get_n_bins()[i]
        bin_counts = np.bincount(X_binned_uniform[:, i], minlength=n_bins)
        non_empty = np.sum(bin_counts > 0)
        print(f"  {name}: {n_bins} bins, {non_empty} non-empty, " +
              f"samples per non-empty bin: min={bin_counts[bin_counts>0].min()}, max={bin_counts.max()}")
 
 
if __name__ == "__main__":
    demonstrate_binning()

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

import numpy as np
from dataclasses import dataclass
from typing import List, Tuple, Optional
 
@dataclass
class HistogramBin:
    """Single bin in a histogram."""
    gradient_sum: float = 0.0
    hessian_sum: float = 0.0
    count: int = 0
 
@dataclass
class SplitInfo:
    """Information about a candidate split."""
    feature: int
    bin_threshold: int  # Split at values <= bin_threshold
    gain: float
    left_gradient: float
    left_hessian: float
    left_count: int
 
 
class HistogramSplitFinder:
    """
    Histogram-based split finding for gradient boosting.
    """
    
    def __init__(self, n_bins: int = 256, lambda_l2: float = 1.0, gamma: float = 0.0):
        self.n_bins = n_bins
        self.lambda_l2 = lambda_l2
        self.gamma = gamma
        
    def build_histograms(self, X_binned: np.ndarray, g: np.ndarray, 
                         h: np.ndarray, sample_indices: np.ndarray) -> np.ndarray:
        """
        Build histograms for all features.
        
        Parameters:
        -----------
        X_binned : array of shape (n_samples, n_features), dtype uint8
            Binned feature values
        g : array of shape (n_samples,)
            Gradients
        h : array of shape (n_samples,)
            Hessians
        sample_indices : array of sample indices in this node
        
        Returns:
        --------
        histograms : array of shape (n_features, n_bins, 3)
            histograms[f, b, 0] = gradient sum
            histograms[f, b, 1] = hessian sum
            histograms[f, b, 2] = count
        """
        n_features = X_binned.shape[1]
        histograms = np.zeros((n_features, self.n_bins, 3))
        
        for i in sample_indices:
            for f in range(n_features):
                b = X_binned[i, f]
                histograms[f, b, 0] += g[i]
                histograms[f, b, 1] += h[i]
                histograms[f, b, 2] += 1
        
        return histograms
    
    def find_best_split(self, histograms: np.ndarray, 
                        min_samples_leaf: int = 1) -> Optional[SplitInfo]:
        """
        Find the best split across all features.
        
        Parameters:
        -----------
        histograms : Histogram array from build_histograms
        min_samples_leaf : Minimum samples required in each child
        
        Returns:
        --------
        best_split : SplitInfo or None if no valid split found
        """
        n_features = histograms.shape[0]
        best_split = None
        best_gain = 0.0
        
        for f in range(n_features):
            # Total statistics for this feature
            G_total = histograms[f, :, 0].sum()
            H_total = histograms[f, :, 1].sum()
            count_total = int(histograms[f, :, 2].sum())
            
            if count_total == 0:
                continue
            
            # Parent score (before split)
            parent_score = self._compute_score(G_total, H_total)
            
            # Scan bins to find best split
            G_L, H_L, count_L = 0.0, 0.0, 0
            
            for b in range(self.n_bins - 1):  # Can't split after last bin
                G_L += histograms[f, b, 0]
                H_L += histograms[f, b, 1]
                count_L += int(histograms[f, b, 2])
                
                count_R = count_total - count_L
                
                # Check minimum samples constraint
                if count_L < min_samples_leaf or count_R < min_samples_leaf:
                    continue
                
                G_R = G_total - G_L
                H_R = H_total - H_L
                
                # Compute gain
                left_score = self._compute_score(G_L, H_L)
                right_score = self._compute_score(G_R, H_R)
                gain = left_score + right_score - parent_score - self.gamma
                
                if gain > best_gain:
                    best_gain = gain
                    best_split = SplitInfo(
                        feature=f,
                        bin_threshold=b,
                        gain=gain,
                        left_gradient=G_L,
                        left_hessian=H_L,
                        left_count=count_L
                    )
        
        return best_split
    
    def _compute_score(self, G: float, H: float) -> float:
        """Compute the score (negative loss) for a leaf."""
        return (G ** 2) / (H + self.lambda_l2)
 
 
# Demonstration
def demonstrate_histogram_splitting():
    np.random.seed(42)
    
    # Create synthetic data
    n_samples = 10000
    n_features = 5
    n_bins = 32
    
    # Simulate binned features
    X_binned = np.random.randint(0, n_bins, (n_samples, n_features), dtype=np.uint8)
    
    # Create a target with structure
    y = (X_binned[:, 0] > 15).astype(float) + 0.5 * (X_binned[:, 1] > 20).astype(float)
    y += 0.1 * np.random.randn(n_samples)
    
    # For squared loss: g = pred - y, h = 1
    pred = np.zeros(n_samples)
    g = pred - y
    h = np.ones(n_samples)
    
    # Build histograms for root node (all samples)
    sample_indices = np.arange(n_samples)
    
    finder = HistogramSplitFinder(n_bins=n_bins, lambda_l2=1.0, gamma=0.0)
    
    import time
    start = time.time()
    histograms = finder.build_histograms(X_binned, g, h, sample_indices)
    hist_time = time.time() - start
    
    start = time.time()
    best_split = finder.find_best_split(histograms)
    split_time = time.time() - start
    
    print("Histogram-based Split Finding Demo")
    print("=" * 50)
    print(f"Samples: {n_samples}, Features: {n_features}, Bins: {n_bins}")
    print(f"\nHistogram construction: {1000*hist_time:.2f} ms")
    print(f"Split finding:          {1000*split_time:.2f} ms")
    
    if best_split:
        print(f"\nBest Split Found:")
        print(f"  Feature: {best_split.feature}")
        print(f"  Bin threshold: {best_split.bin_threshold}")
        print(f"  Gain: {best_split.gain:.4f}")
        print(f"  Left samples: {best_split.left_count}")
        print(f"  Right samples: {len(sample_indices) - best_split.left_count}")
    
    # Show histogram for best feature
    print(f"\nHistogram for feature {best_split.feature}:")
    print(f"  Bins with samples: {(histograms[best_split.feature, :, 2] > 0).sum()}")
    
    return histograms, best_split
 
 
if __name__ == "__main__":
    demonstrate_histogram_splitting()

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

import numpy as np
import time
 
def demonstrate_subtraction_trick():
    """
    Demonstrate the histogram subtraction trick.
    """
    np.random.seed(42)
    
    # Simulation parameters
    n_samples = 100000
    n_features = 50
    n_bins = 256
    
    # Simulate binned data
    X_binned = np.random.randint(0, n_bins, (n_samples, n_features), dtype=np.uint8)
    g = np.random.randn(n_samples)
    h = np.ones(n_samples)
    
    # Simulate a split: 30% left, 70% right
    split_point = int(n_samples * 0.3)
    left_indices = np.arange(split_point)
    right_indices = np.arange(split_point, n_samples)
    all_indices = np.arange(n_samples)
    
    print("Subtraction Trick Demonstration")
    print("=" * 50)
    print(f"Samples: {n_samples}, Features: {n_features}, Bins: {n_bins}")
    print(f"Left child: {len(left_indices)} samples ({100*len(left_indices)/n_samples:.0f}%)")
    print(f"Right child: {len(right_indices)} samples ({100*len(right_indices)/n_samples:.0f}%)")
    
    def build_histogram(indices):
        """Build histogram for given sample indices."""
        hist = np.zeros((n_features, n_bins, 3))
        for i in indices:
            for f in range(n_features):
                b = X_binned[i, f]
                hist[f, b, 0] += g[i]
                hist[f, b, 1] += h[i]
                hist[f, b, 2] += 1
        return hist
    
    # Method 1: Build both histograms directly
    start = time.time()
    hist_left_direct = build_histogram(left_indices)
    hist_right_direct = build_histogram(right_indices)
    time_direct = time.time() - start
    
    # Method 2: Build parent, then smaller child, then subtract
    start = time.time()
    hist_parent = build_histogram(all_indices)
    # Build smaller child (left in this case)
    hist_left_sub = build_histogram(left_indices)
    # Derive larger child by subtraction
    hist_right_sub = hist_parent - hist_left_sub
    time_subtraction = time.time() - start
    
    print(f"\nDirect method:      {1000 * time_direct:.1f} ms")
    print(f"Subtraction method: {1000 * time_subtraction:.1f} ms")
    print(f"Speedup:            {time_direct / time_subtraction:.2f}x")
    
    # Verify correctness
    assert np.allclose(hist_right_direct, hist_right_sub), "Subtraction result mismatch!"
    print("\n✓ Subtraction result matches direct computation")
    
    # Analyze savings more precisely
    direct_work = len(left_indices) + len(right_indices)  # = n_samples
    sub_work = n_samples + len(left_indices)  # Parent + smaller child
    
    # But we would have computed parent anyway for split finding!
    # So the marginal cost is just the smaller child
    marginal_sub_work = len(left_indices)
    
    print(f"\nWork analysis:")
    print(f"  Direct: process {direct_work:,} sample-features")
    print(f"  Subtraction (marginal): process {marginal_sub_work:,} sample-features")
    print(f"  Theoretical speedup: {direct_work / marginal_sub_work:.2f}x")
    
    # Best case: balanced split → 2x speedup
    # Typical case with 30-70 split → about 1.4x speedup
    # Worst case: one child has 1 sample → nearly 2x speedup
    
    return hist_parent, hist_left_sub, hist_right_sub
 
 
def analyze_subtraction_savings():
    """
    Analyze subtraction trick savings for different split ratios.
    """
    print("\nSubtraction Trick Savings by Split Ratio")
    print("=" * 50)
    print(f"{'Left %':<10} {'Right %':<10} {'Direct Work':<15} {'Sub Work':<15} {'Speedup':<10}")
    print("-" * 50)
    
    n = 100000
    
    for left_pct in [0.1, 0.2, 0.3, 0.4, 0.5]:
        left_n = int(n * left_pct)
        right_n = n - left_n
        
        direct_work = left_n + right_n  # Build both
        smaller = min(left_n, right_n)
        sub_work = smaller  # Only build smaller (parent already exists)
        
        speedup = direct_work / sub_work
        
        print(f"{100*left_pct:<10.0f} {100*(1-left_pct):<10.0f} "
              f"{direct_work:<15,} {sub_work:<15,} {speedup:<10.1f}x")
    
    print("\nNote: Speedup is relative to building BOTH children.")
    print("The parent histogram was already built during split finding.")
 
 
if __name__ == "__main__":
    demonstrate_subtraction_trick()
    analyze_subtraction_savings()

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

import lightgbm as lgb
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
import numpy as np
import time
 
def compare_bin_counts():
    """
    Compare LightGBM performance with different bin counts.
    """
    # Generate dataset
    X, y = make_classification(
        n_samples=100000, n_features=50, n_informative=20,
        n_redundant=10, random_state=42
    )
    X_train, X_val, y_train, y_val = train_test_split(
        X, y, test_size=0.2, random_state=42
    )
    
    train_data = lgb.Dataset(X_train, label=y_train)
    val_data = lgb.Dataset(X_val, label=y_val, reference=train_data)
    
    bin_counts = [15, 31, 63, 127, 255, 511]
    results = []
    
    for max_bin in bin_counts:
        params = {
            'objective': 'binary',
            'metric': 'auc',
            'boosting_type': 'gbdt',
            'num_leaves': 31,
            'learning_rate': 0.05,
            'max_bin': max_bin,
            'verbose': -1,
        }
        
        start = time.time()
        model = lgb.train(
            params, train_data,
            num_boost_round=100,
            valid_sets=[val_data],
            callbacks=[lgb.log_evaluation(period=0)]
        )
        elapsed = time.time() - start
        
        results.append({
            'max_bin': max_bin,
            'time': elapsed,
            'auc': model.best_score['valid_0']['auc']
        })
    
    # Print results
    print("Effect of max_bin on Training")
    print("=" * 60)
    print(f"{'max_bin':<12} {'Time (s)':<15} {'Val AUC':<15} {'Speedup':<15}")
    print("-" * 60)
    
    baseline_time = results[-1]['time']  # 255 bins as baseline
    for r in results:
        speedup = baseline_time / r['time']
        print(f"{r['max_bin']:<12} {r['time']:<15.2f} {r['auc']:<15.6f} {speedup:<15.2f}x")
    
    print("\nObservations:")
    print("- Fewer bins = faster training")
    print("- AUC is relatively stable across bin counts")
    print("- 255 bins (default) offers good accuracy/speed tradeoff")
    
    return results
 
 
def demonstrate_feature_specific_binning():
    """
    Show that different features may benefit from different bin counts.
    """
    np.random.seed(42)
    n = 10000
    
    # Feature A: continuous with many unique values
    feat_continuous = np.random.randn(n) * 100
    unique_continuous = len(np.unique(feat_continuous))
    
    # Feature B: ordinal with 10 levels
    feat_ordinal = np.random.randint(0, 10, n)
    unique_ordinal = len(np.unique(feat_ordinal))
    
    # Feature C: binary
    feat_binary = np.random.randint(0, 2, n)
    unique_binary = len(np.unique(feat_binary))
    
    print("Feature-Specific Discretization Analysis")
    print("=" * 50)
    print(f"Continuous feature: {unique_continuous:,} unique values → benefits from 255 bins")
    print(f"Ordinal feature: {unique_ordinal} unique values → 10 bins sufficient")
    print(f"Binary feature: {unique_binary} unique values → 2 bins sufficient")
    
    print("\nLightGBM automatically adjusts bin count based on unique values.")
    print("If a feature has < max_bin unique values, it uses fewer bins.")
 
 
if __name__ == "__main__":
    compare_bin_counts()
    print("\n")
    demonstrate_feature_specific_binning()