High-Cardinality Features

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

import pandas as pd
from sklearn.preprocessing import LabelEncoder, OrdinalEncoder
 
# Sample data
df = pd.DataFrame({
    'fruit': ['apple', 'banana', 'cherry', 'apple', 'date', 'banana'],
    'size': ['small', 'medium', 'large', 'small', 'medium', 'large']
})
 
# Method 1: LabelEncoder (single column, learns mapping automatically)
label_encoder = LabelEncoder()
df['fruit_encoded'] = label_encoder.fit_transform(df['fruit'])
 
# Access the mapping
print("Label classes:", label_encoder.classes_)
# Output: ['apple' 'banana' 'cherry' 'date']
 
# Inverse transform
original = label_encoder.inverse_transform([0, 1, 2, 3])
print("Decoded:", original)
 
# Method 2: OrdinalEncoder (multiple columns, explicit category ordering)
# Useful when you want to enforce a specific order
ordinal_encoder = OrdinalEncoder(
    categories=[['small', 'medium', 'large']],  # Explicit ordering
    handle_unknown='use_encoded_value',
    unknown_value=-1  # Assign -1 to unseen categories
)
df['size_encoded'] = ordinal_encoder.fit_transform(df[['size']])
 
# Check the result
print(df)
 
# Handling unseen categories at inference time
new_data = pd.DataFrame({'size': ['tiny', 'large', 'xl']})
new_encoded = ordinal_encoder.transform(new_data)
print("With unknowns:", new_encoded)  # [-1, 2, -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

import pandas as pd
import numpy as np
from sklearn.preprocessing import OneHotEncoder
from scipy import sparse
 
# Sample data
df = pd.DataFrame({
    'color': ['red', 'blue', 'green', 'red', 'blue', 'green', 'yellow'],
    'size': ['S', 'M', 'L', 'M', 'S', 'L', 'M']
})
 
# Method 1: pandas get_dummies (convenient but creates dense arrays)
df_encoded = pd.get_dummies(df, columns=['color', 'size'], prefix=['color', 'size'])
print("Pandas get_dummies result:")
print(df_encoded)
 
# Method 2: sklearn OneHotEncoder (production-grade, handles unknowns)
encoder = OneHotEncoder(
    sparse_output=True,     # Return sparse matrix for memory efficiency
    handle_unknown='ignore', # Silently ignore unseen categories (all zeros)
    drop='first',            # Drop first category to avoid multicollinearity
    min_frequency=2,         # Combine rare categories (frequency < 2)
    max_categories=10        # Limit max categories (combine rest as 'infrequent')
)
 
# Fit and transform
encoded_sparse = encoder.fit_transform(df[['color', 'size']])
print(f"
Sparse matrix shape: {encoded_sparse.shape}")
print(f"Sparse matrix density: {encoded_sparse.nnz / np.prod(encoded_sparse.shape):.2%}")
print(f"Memory usage: {encoded_sparse.data.nbytes + encoded_sparse.indices.nbytes + encoded_sparse.indptr.nbytes} bytes")
 
# Get feature names for interpretability
feature_names = encoder.get_feature_names_out(['color', 'size'])
print(f"
Feature names: {feature_names}")
 
# Handle unseen categories at inference time
new_data = pd.DataFrame({'color': ['purple', 'red'], 'size': ['XL', 'S']})
new_encoded = encoder.transform(new_data)
print(f"
New data encoding (note zeros for unseen 'purple' and 'XL'):")
print(new_encoded.toarray())
 
# Inverse transform (only possible without drop='first')
encoder_no_drop = OneHotEncoder(sparse_output=False, handle_unknown='ignore')
encoder_no_drop.fit(df[['color', 'size']])
encoded = encoder_no_drop.transform(df[['color', 'size']])
decoded = encoder_no_drop.inverse_transform(encoded)
print(f"
Decoded: {decoded}")

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

import pandas as pd
import numpy as np
import category_encoders as ce
 
# Sample high-cardinality data
np.random.seed(42)
n_samples = 10000
n_categories = 1000  # High cardinality
 
df = pd.DataFrame({
    'category': [f'cat_{i}' for i in np.random.randint(0, n_categories, n_samples)],
    'target': np.random.randn(n_samples)
})
 
# Binary encoding using category_encoders library
binary_encoder = ce.BinaryEncoder(cols=['category'])
df_encoded = binary_encoder.fit_transform(df[['category']])
 
print(f"Original: {n_categories} unique categories")
print(f"Binary encoding columns: {df_encoded.shape[1]}")
print(f"Expected: ceil(log2({n_categories})) = {int(np.ceil(np.log2(n_categories)))}")
print(f"
Column names: {df_encoded.columns.tolist()}")
print(f"
Sample encodings:")
print(df_encoded.head(10))
 
# Memory comparison
one_hot_memory = n_samples * n_categories * 1  # bytes (assuming uint8)
binary_memory = n_samples * int(np.ceil(np.log2(n_categories))) * 1
print(f"
Memory: One-Hot = {one_hot_memory / 1e6:.2f} MB, Binary = {binary_memory / 1e6:.2f} MB")
print(f"Reduction factor: {one_hot_memory / binary_memory:.1f}x")
 
# Manual implementation for understanding
def binary_encode(categories: pd.Series) -> pd.DataFrame:
    """Manual binary encoding implementation."""
    # Step 1: Integer encode
    unique_cats = categories.unique()
    cat_to_int = {cat: i for i, cat in enumerate(unique_cats)}
    integers = categories.map(cat_to_int)
    
    # Step 2: Determine number of bits needed
    n_bits = int(np.ceil(np.log2(len(unique_cats) + 1)))
    
    # Step 3: Convert to binary representation
    result = pd.DataFrame()
    for bit in range(n_bits):
        result[f'bit_{n_bits - 1 - bit}'] = ((integers >> (n_bits - 1 - bit)) & 1).astype(int)
    
    return result
 
# Test manual implementation
df_manual = binary_encode(df['category'])
print(f"
Manual implementation shape: {df_manual.shape}")

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

import pandas as pd
import numpy as np
from sklearn.base import BaseEstimator, TransformerMixin
 
class FrequencyEncoder(BaseEstimator, TransformerMixin):
    """
    Production-ready frequency encoder with proper train/test handling.
    """
    def __init__(self, columns=None, normalize=True, handle_unknown='global_mean'):
        """
        Parameters:
        -----------
        columns : list of str
            Columns to encode. If None, encode all object/category columns.
        normalize : bool
            If True, use frequencies (0-1). If False, use raw counts.
        handle_unknown : str
            Strategy for unseen categories: 'global_mean', 'zero', 'min', 'error'
        """
        self.columns = columns
        self.normalize = normalize
        self.handle_unknown = handle_unknown
        self.encoding_maps_ = {}
        self.global_defaults_ = {}
        
    def fit(self, X, y=None):
        """Learn frequency/count mappings from training data."""
        X = X.copy()
        if self.columns is None:
            self.columns = X.select_dtypes(include=['object', 'category']).columns.tolist()
        
        for col in self.columns:
            counts = X[col].value_counts(normalize=self.normalize)
            self.encoding_maps_[col] = counts.to_dict()
            
            # Set default for unseen categories
            if self.handle_unknown == 'global_mean':
                self.global_defaults_[col] = counts.mean()
            elif self.handle_unknown == 'zero':
                self.global_defaults_[col] = 0
            elif self.handle_unknown == 'min':
                self.global_defaults_[col] = counts.min()
            else:
                self.global_defaults_[col] = None
                
        return self
    
    def transform(self, X):
        """Apply learned frequency mappings."""
        X = X.copy()
        for col in self.columns:
            X[col] = X[col].map(self.encoding_maps_[col])
            
            # Handle unseen categories
            if self.handle_unknown == 'error':
                if X[col].isna().any():
                    raise ValueError(f"Unseen categories found in column '{col}'")
            else:
                X[col] = X[col].fillna(self.global_defaults_[col])
                
        return X
 
# Example usage
np.random.seed(42)
train_df = pd.DataFrame({
    'product_id': np.random.choice(['P001', 'P002', 'P003', 'P004'], 1000, 
                                    p=[0.5, 0.3, 0.15, 0.05]),
    'merchant': np.random.choice(['M1', 'M2', 'M3'], 1000, p=[0.6, 0.3, 0.1]),
    'target': np.random.randint(0, 2, 1000)
})
 
# Fit encoder on training data
encoder = FrequencyEncoder(normalize=True, handle_unknown='global_mean')
train_encoded = encoder.fit_transform(train_df)
 
print("Training data frequencies:")
print(train_encoded.head(10))
print(f"
Learned mappings:")
for col, mapping in encoder.encoding_maps_.items():
    print(f"  {col}: {mapping}")
 
# Apply to test data (including unseen category 'P999')
test_df = pd.DataFrame({
    'product_id': ['P001', 'P002', 'P999', 'P003'],  # P999 is unseen
    'merchant': ['M1', 'M2', 'M4', 'M3'],  # M4 is unseen
    'target': [0, 1, 1, 0]
})
 
test_encoded = encoder.transform(test_df)
print("
Test data with unseen categories:")
print(test_encoded)

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 joblib
import pandas as pd
import numpy as np
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import OneHotEncoder, StandardScaler
from sklearn.impute import SimpleImputer
import category_encoders as ce
from datetime import datetime
 
class EncodingPipeline:
    """
    Production-ready encoding pipeline with proper versioning and persistence.
    """
    def __init__(self, version: str = None):
        self.version = version or datetime.now().strftime("%Y%m%d_%H%M%S")
        self.pipeline = None
        self.metadata = {
            "version": self.version,
            "created_at": datetime.now().isoformat(),
            "feature_columns": None,
            "category_mappings": {}
        }
    
    def build_pipeline(self, 
                       low_cardinality_cols: list,
                       high_cardinality_cols: list,
                       numeric_cols: list,
                       target_encode_cols: list = None):
        """
        Construct sklearn ColumnTransformer with appropriate encoders.
        """
        transformers = []
        
        # Low cardinality: One-Hot Encoding
        if low_cardinality_cols:
            transformers.append((
                'low_card_ohe',
                OneHotEncoder(
                    sparse_output=True,
                    handle_unknown='ignore',
                    drop='if_binary',
                    min_frequency=0.01  # Group rare categories
                ),
                low_cardinality_cols
            ))
        
        # High cardinality: Target Encoding (must fit with target!)
        if high_cardinality_cols:
            transformers.append((
                'high_card_target',
                ce.TargetEncoder(
                    cols=high_cardinality_cols,
                    smoothing=100,  # Regularization
                    handle_unknown='value',  # Use global mean
                    handle_missing='value'
                ),
                high_cardinality_cols
            ))
        
        # Numeric: Standard scaling with imputation
        if numeric_cols:
            transformers.append((
                'numeric',
                Pipeline([
                    ('impute', SimpleImputer(strategy='median')),
                    ('scale', StandardScaler())
                ]),
                numeric_cols
            ))
        
        self.pipeline = ColumnTransformer(
            transformers=transformers,
            remainder='drop',  # Explicitly drop unused columns
            verbose_feature_names_out=True
        )
        
        self.metadata["feature_columns"] = {
            "low_cardinality": low_cardinality_cols,
            "high_cardinality": high_cardinality_cols,
            "numeric": numeric_cols
        }
        
        return self
    
    def fit(self, X: pd.DataFrame, y: pd.Series = None):
        """Fit all encoders. Target y required for target encoding."""
        if self.pipeline is None:
            raise ValueError("Build pipeline before fitting.")
        
        # Note: ColumnTransformer passes y to each transformer that needs it
        self.pipeline.fit(X, y)
        
        # Store category mappings for monitoring
        for name, transformer, cols in self.pipeline.transformers_:
            if isinstance(transformer, OneHotEncoder):
                for col, categories in zip(cols, transformer.categories_):
                    self.metadata["category_mappings"][col] = list(categories)
        
        self.metadata["n_output_features"] = len(self.get_feature_names())
        return self
    
    def transform(self, X: pd.DataFrame) -> np.ndarray:
        """Transform data using fitted encoders."""
        return self.pipeline.transform(X)
    
    def fit_transform(self, X: pd.DataFrame, y: pd.Series = None) -> np.ndarray:
        """Fit and transform in one step."""
        self.fit(X, y)
        return self.transform(X)
    
    def get_feature_names(self) -> list:
        """Get output feature names."""
        return list(self.pipeline.get_feature_names_out())
    
    def save(self, path: str):
        """Persist pipeline and metadata."""
        artifact = {
            "pipeline": self.pipeline,
            "metadata": self.metadata
        }
        joblib.dump(artifact, path)
        print(f"Saved encoding pipeline v{self.version} to {path}")
    
    @classmethod
    def load(cls, path: str) -> 'EncodingPipeline':
        """Load persisted pipeline."""
        artifact = joblib.load(path)
        instance = cls(version=artifact["metadata"]["version"])
        instance.pipeline = artifact["pipeline"]
        instance.metadata = artifact["metadata"]
        return instance
 
# Usage example
np.random.seed(42)
df = pd.DataFrame({
    'category_low': np.random.choice(['A', 'B', 'C'], 1000),
    'category_high': [f'id_{i}' for i in np.random.randint(0, 500, 1000)],
    'numeric_1': np.random.randn(1000),
    'numeric_2': np.random.randn(1000) * 10,
    'target': np.random.randint(0, 2, 1000)
})
 
# Build and fit pipeline
enc_pipeline = EncodingPipeline(version="v1.0.0")
enc_pipeline.build_pipeline(
    low_cardinality_cols=['category_low'],
    high_cardinality_cols=['category_high'],
    numeric_cols=['numeric_1', 'numeric_2']
)
 
X = df.drop('target', axis=1)
y = df['target']
X_encoded = enc_pipeline.fit_transform(X, y)
 
print(f"Pipeline version: {enc_pipeline.version}")
print(f"Output shape: {X_encoded.shape}")
print(f"Feature names: {enc_pipeline.get_feature_names()[:10]}...")
 
# Save for production
enc_pipeline.save("encoding_pipeline.joblib")
 
# Load in production
loaded_pipeline = EncodingPipeline.load("encoding_pipeline.joblib")
print(f"Loaded version: {loaded_pipeline.version}")