Advanced Feature Engineering Topics
Advanced feature engineering techniques, automation, and practices.
Table of Contents
- Automated Feature Engineering
- Feature Interactions and Polynomial Features
- Time Series Feature Engineering
- Text Feature Engineering
- Feature Engineering Pipelines
- Feature Importance and Selection
- Common Feature Engineering Pitfalls
Automated Feature Engineering
Featuretools
Automated feature engineering library.
try:
import featuretools as ft
# Create entity set
es = ft.EntitySet(id='data')
# Add entities
es = es.entity_from_dataframe(
entity_id='customers',
dataframe=df_customers,
index='customer_id'
)
es = es.entity_from_dataframe(
entity_id='orders',
dataframe=df_orders,
index='order_id',
time_index='order_date'
)
# Define relationships
es = es.add_relationship(
ft.Relationship(es['customers']['customer_id'],
es['orders']['customer_id'])
)
# Generate features
feature_matrix, feature_defs = ft.dfs(
entityset=es,
target_entity='customers',
max_depth=2,
verbose=1
)
print(f"Generated {len(feature_defs)} features")
except ImportError:
print("Install featuretools: pip install featuretools")
AutoFeat
Automatic feature engineering and selection.
try:
from autofeat import AutoFeatClassifier
# Automatic feature engineering
autofeat = AutoFeatClassifier(verbose=1)
X_new = autofeat.fit_transform(X, y)
print(f"Original features: {X.shape[1]}")
print(f"New features: {X_new.shape[1]}")
except ImportError:
print("Install autofeat: pip install autofeat")
Feature Interactions and Polynomial Features
Manual Interaction Features
# Create interaction features manually
df['feature1_x_feature2'] = df['feature1'] * df['feature2']
df['feature1_div_feature2'] = df['feature1'] / (df['feature2'] + 1e-6)
df['feature1_plus_feature2'] = df['feature1'] + df['feature2']
df['feature1_minus_feature2'] = df['feature1'] - df['feature2']
df['feature1_power_feature2'] = df['feature1'] ** df['feature2']
# Ratio features
df['ratio_1_2'] = df['feature1'] / (df['feature2'] + 1e-6)
df['ratio_2_1'] = df['feature2'] / (df['feature1'] + 1e-6)
# Difference features
df['diff_1_2'] = df['feature1'] - df['feature2']
df['abs_diff_1_2'] = np.abs(df['feature1'] - df['feature2'])
Polynomial Features
from sklearn.preprocessing import PolynomialFeatures
# Full polynomial features (degree 2)
poly_full = PolynomialFeatures(degree=2, include_bias=False)
X_poly_full = poly_full.fit_transform(X)
print(f"Original: {X.shape[1]} features")
print(f"Polynomial: {X_poly_full.shape[1]} features")
# Interaction-only (no squared terms)
poly_interaction = PolynomialFeatures(
degree=2,
interaction_only=True,
include_bias=False
)
X_poly_interaction = poly_interaction.fit_transform(X)
print(f"Interaction-only: {X_poly_interaction.shape[1]} features")
# Compare performance
from sklearn.linear_model import LogisticRegression
models = {
'Original': LogisticRegression(random_state=42, max_iter=1000),
'Polynomial': LogisticRegression(random_state=42, max_iter=1000),
'Interaction': LogisticRegression(random_state=42, max_iter=1000)
}
results = {}
results['Original'] = cross_val_score(
models['Original'], X, y, cv=5
).mean()
results['Polynomial'] = cross_val_score(
models['Polynomial'], X_poly_full, y, cv=5
).mean()
results['Interaction'] = cross_val_score(
models['Interaction'], X_poly_interaction, y, cv=5
).mean()
for name, score in results.items():
print(f"{name:15s}: {score:.3f}")
Feature Crosses
# Create feature crosses (combinations)
# Example: Combine categorical features
df['category_region_cross'] = df['category'].astype(str) + '_' + df['region'].astype(str)
# One-hot encode the cross
df_cross_encoded = pd.get_dummies(df, columns=['category_region_cross'], prefix='cross')
Time Series Feature Engineering
Temporal Features
# Extract time components
df['year'] = df['timestamp'].dt.year
df['month'] = df['timestamp'].dt.month
df['day'] = df['timestamp'].dt.day
df['day_of_week'] = df['timestamp'].dt.dayofweek
df['day_of_year'] = df['timestamp'].dt.dayofyear
df['week'] = df['timestamp'].dt.isocalendar().week
df['quarter'] = df['timestamp'].dt.quarter
df['hour'] = df['timestamp'].dt.hour
df['minute'] = df['timestamp'].dt.minute
# Cyclical encoding (for periodic features)
df['month_sin'] = np.sin(2 * np.pi * df['month'] / 12)
df['month_cos'] = np.cos(2 * np.pi * df['month'] / 12)
df['day_of_week_sin'] = np.sin(2 * np.pi * df['day_of_week'] / 7)
df['day_of_week_cos'] = np.cos(2 * np.pi * df['day_of_week'] / 7)
df['hour_sin'] = np.sin(2 * np.pi * df['hour'] / 24)
df['hour_cos'] = np.cos(2 * np.pi * df['hour'] / 24)
# Boolean features
df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)
df['is_month_start'] = df['day'] <= 3
df['is_month_end'] = df['day'] >= 28
df['is_holiday'] = df['date'].isin(holidays).astype(int)
Lag Features
# Lag features (previous values)
df['value_lag_1'] = df['value'].shift(1)
df['value_lag_7'] = df['value'].shift(7) # Weekly lag
df['value_lag_30'] = df['value'].shift(30) # Monthly lag
# Difference features
df['value_diff_1'] = df['value'].diff(1)
df['value_diff_7'] = df['value'].diff(7)
# Percentage change
df['value_pct_change'] = df['value'].pct_change()
df['value_pct_change_7'] = df['value'].pct_change(7)
Rolling Statistics
# Rolling window statistics
window_sizes = [3, 7, 14, 30]
for window in window_sizes:
df[f'value_rolling_mean_{window}'] = df['value'].rolling(window=window).mean()
df[f'value_rolling_std_{window}'] = df['value'].rolling(window=window).std()
df[f'value_rolling_min_{window}'] = df['value'].rolling(window=window).min()
df[f'value_rolling_max_{window}'] = df['value'].rolling(window=window).max()
df[f'value_rolling_median_{window}'] = df['value'].rolling(window=window).median()
# Expanding window statistics
df['value_expanding_mean'] = df['value'].expanding().mean()
df['value_expanding_std'] = df['value'].expanding().std()
Time-Based Aggregations
# Group by time periods
df['date'] = pd.to_datetime(df['timestamp'])
# Daily aggregations
daily_stats = df.groupby(df['date'].dt.date).agg({
'value': ['mean', 'std', 'min', 'max', 'count']
})
# Weekly aggregations
df['week'] = df['date'].dt.to_period('W')
weekly_stats = df.groupby('week')['value'].agg(['mean', 'std', 'min', 'max'])
# Monthly aggregations
df['month'] = df['date'].dt.to_period('M')
m>'month')['value'].agg(['mean', 'std', 'min', 'max'])
# Merge back
df = df.merge(weekly_stats, left_on='week', right_index=True, suffixes=('', '_weekly'))
Text Feature Engineering
Basic Text Features
# Length features
df['text_length'] = df['text'].str.len()
df['word_count'] = df['text'].str.split().str.len()
df['sentence_count'] = df['text'].str.count(r'[.!?]+')
df['paragraph_count'] = df['text'].str.count('\n\n') + 1
# Average features
df['avg_word_length'] = df['text_length'] / (df['word_count'] + 1)
df['avg_sentence_length'] = df['word_count'] / (df['sentence_count'] + 1)
# Character features
df['uppercase_count'] = df['text'].str.findall(r'[A-Z]').str.len()
df['lowercase_count'] = df['text'].str.findall(r'[a-z]').str.len()
df['digit_count'] = df['text'].str.findall(r'[0-9]').str.len()
df['special_char_count'] = df['text'].str.findall(r'[^a-zA-Z0-9\s]').str.len()
N-gram Features
from sklearn.feature_extraction.text import CountVectorizer, TfidfVectorizer
# Unigrams and bigrams
vectorizer = CountVectorizer(
ngram_range=(1, 2), # Unigrams and bigrams
max_features=1000
)
X_text = vectorizer.fit_transform(df['text'])
# TF-IDF
tfidf = TfidfVectorizer(
ngram_range=(1, 2),
max_features=1000
)
X_tfidf = tfidf.fit_transform(df['text'])
Text Statistics
# Readability features
df['avg_syllables_per_word'] = df['text'].apply(count_syllables) / df['word_count']
# Sentiment features (if using textblob)
try:
from textblob import TextBlob
df['sentiment_polarity'] = df['text'].apply(lambda x: TextBlob(x).sentiment.polarity)
df['sentiment_subjectivity'] = df['text'].apply(lambda x: TextBlob(x).sentiment.subjectivity)
except ImportError:
print("Install textblob: pip install textblob")
Comprehensive sklearn Pipeline and ColumnTransformer Guide
Introduction to Pipelines
Pipelines automate the ML workflow and prevent data leakage by ensuring transformations are applied consistently.
Benefits:
- Prevents data leakage
- Ensures consistent transformations
- Simplifies code
- Makes models reproducible
- Easy to deploy
Basic Pipeline
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.linear_model import LogisticRegression
# Simple pipeline
pipeline = Pipeline([
('scaler', StandardScaler()),
('classifier', LogisticRegression(random_state=42))
])
# Fit and predict
pipeline.fit(X_train, y_train)
predicti>
score = pipeline.score(X_test, y_test)
ColumnTransformer Basics
ColumnTransformer applies different transformations to different columns.
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
# Define column types
numeric_cols = ['age', 'income', 'score']
categorical_cols = ['city', 'category']
# Create ColumnTransformer
preprocessor = ColumnTransformer(
transformers=[
('num', StandardScaler(), numeric_cols),
('cat', OneHotEncoder(drop='first', sparse=False), categorical_cols)
],
remainder='drop' # What to do with remaining columns
)
# Use in pipeline
pipeline = Pipeline([
('preprocessor', preprocessor),
('classifier', LogisticRegression(random_state=42))
])
Advanced ColumnTransformer
Multiple Transformations per Column Type:
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, RobustScaler, OneHotEncoder
from sklearn.impute import SimpleImputer
# Complex preprocessing
preprocessor = ColumnTransformer(
transformers=[
# Numeric: impute then scale
('numeric', Pipeline([
('imputer', SimpleImputer(strategy='median')),
('scaler', StandardScaler())
]), ['age', 'income']),
# Skewed numeric: impute, transform, scale
('skewed', Pipeline([
('imputer', SimpleImputer(strategy='median')),
('transformer', PowerTransformer(method='yeo-johnson')),
('scaler', RobustScaler())
]), ['price', 'revenue']),
# Categorical: impute then encode
('categorical', Pipeline([
('imputer', SimpleImputer(strategy='most_frequent')),
('encoder', OneHotEncoder(drop='first', sparse=False, handle_unknown='ignore'))
]), ['city', 'category']),
# Binary: just encode
('binary', OneHotEncoder(drop='if_binary', sparse=False), ['is_active'])
],
remainder='passthrough' # Keep other columns
)
Custom Transformers
Creating Custom Transformers:
from sklearn.base import BaseEstimator, TransformerMixin
import numpy as np
import pandas as pd
class LogTransformer(BaseEstimator, TransformerMixin):
"""Log transformation with error handling"""
def fit(self, X, y=None):
return self
def transform(self, X):
return np.log1p(np.maximum(X, 0)) # Handle negative values
def fit_transform(self, X, y=None):
return self.fit(X, y).transform(X)
class FeatureInteraction(BaseEstimator, TransformerMixin):
"""Create interaction features"""
def __init__(self, interactions):
self.interacti # List of (i, j) tuples
def fit(self, X, y=None):
return self
def transform(self, X):
X_new = X.copy()
for i, j in self.interactions:
X_new = np.hstack([X_new, (X[:, i] * X[:, j]).reshape(-1, 1)])
return X_new
class TargetEncoder(BaseEstimator, TransformerMixin):
"""Target encoding with cross-validation"""
def __init__(self, cv=5):
self.cv = cv
self.encodings = {}
def fit(self, X, y):
from sklearn.model_selection import KFold
kf = KFold(n_splits=self.cv, shuffle=True, random_state=42)
# Calculate encodings for each fold
for train_idx, val_idx in kf.split(X):
X_train_fold, y_train_fold = X[train_idx], y[train_idx]
for category in np.unique(X_train_fold):
mask = X_train_fold == category
if category not in self.encodings:
self.encodings[category] = []
self.encodings[category].append(y_train_fold[mask].mean())
# Average across folds
for cat in self.encodings:
self.encodings[cat] = np.mean(self.encodings[cat])
return self
def transform(self, X):
return np.array([self.encodings.get(cat, 0) for cat in X])
Complex Pipeline Example
from sklearn.pipeline import Pipeline, FeatureUnion
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, PolynomialFeatures
from sklearn.feature_selection import SelectKBest, f_classif
from sklearn.impute import SimpleImputer
# Step 1: Preprocessing
preprocessing = ColumnTransformer(
transformers=[
('numeric', Pipeline([
('imputer', SimpleImputer(strategy='median')),
('scaler', StandardScaler())
]), numeric_features),
('categorical', Pipeline([
('imputer', SimpleImputer(strategy='most_frequent')),
('encoder', OneHotEncoder(drop='first', sparse=False))
]), categorical_features)
]
)
# Step 2: Feature Engineering
feature_engineering = FeatureUnion([
('polynomial', PolynomialFeatures(degree=2, include_bias=False)),
('interactions', FeatureInteraction([(0, 1), (0, 2)]))
])
# Step 3: Feature Selection
feature_selection = SelectKBest(score_func=f_classif, k=20)
# Step 4: Model
model = RandomForestClassifier(n_estimators=100, random_state=42)
# Complete pipeline
full_pipeline = Pipeline([
('preprocessing', preprocessing),
('feature_engineering', feature_engineering),
('feature_selection', feature_selection),
('model', model)
])
# Fit and evaluate
full_pipeline.fit(X_train, y_train)
score = full_pipeline.score(X_test, y_test)
Pipeline Best Practices
1. Always Use Pipelines:
# Bad: Manual transformations
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test) # Easy to forget!
model.fit(X_train_scaled, y_train)
# Good: Pipeline
pipeline = Pipeline([('scaler', StandardScaler()), ('model', model)])
pipeline.fit(X_train, y_train) # Automatically handles train/test
2. Get Feature Names:
# After fitting, get feature names
feature_names = pipeline.named_steps['preprocessor'].get_feature_names_out()
print(f"Feature names: {feature_names}")
3. Access Intermediate Steps:
# Access fitted transformers
scaler = pipeline.named_steps['scaler']
print(f"Mean: {scaler.mean_}")
print(f"Scale: {scaler.scale_}")
4. Grid Search with Pipelines:
from sklearn.model_selection import GridSearchCV
# Parameter grid
param_grid = {
'preprocessor__num__scaler': [StandardScaler(), RobustScaler()],
'model__n_estimators': [50, 100, 200],
'model__max_depth': [5, 10, 15]
}
# Grid search
grid_search = GridSearchCV(
pipeline,
param_grid,
cv=5,
scoring='accuracy',
n_jobs=-1
)
grid_search.fit(X_train, y_train)
print(f"Best score: {grid_search.best_score_:.3f}")
print(f"Best params: {grid_search.best_params_}")
Common Pipeline Patterns
Pattern 1: Simple Preprocessing + Model
pipeline = Pipeline([
('scaler', StandardScaler()),
('model', LogisticRegression())
])
Pattern 2: Mixed Data Types
preprocessor = ColumnTransformer(
transformers=[
('num', StandardScaler(), numeric_cols),
('cat', OneHotEncoder(), categorical_cols)
]
)
pipeline = Pipeline([
('preprocessor', preprocessor),
('model', RandomForestClassifier())
])
Pattern 3: Feature Engineering + Selection + Model
pipeline = Pipeline([
('preprocessor', preprocessor),
('poly', PolynomialFeatures(degree=2)),
('selector', SelectKBest(k=10)),
('model', LogisticRegression())
])
Debugging Pipelines
# Check pipeline steps
print("Pipeline steps:", pipeline.named_steps.keys())
# Check intermediate outputs
X_transformed = pipeline.named_steps['preprocessor'].transform(X_train)
print(f"Transformed shape: {X_transformed.shape}")
# Visualize pipeline
from sklearn.utils import estimator_html_repr
print(estimator_html_repr(pipeline))
Feature Engineering Pipelines
Complete Pipeline
from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import (StandardScaler, OneHotEncoder,
PowerTransformer, FunctionTransformer)
from sklearn.feature_selection import SelectKBest, f_classif
from sklearn.impute import SimpleImputer
# Define feature types
numeric_features = ['feature1', 'feature2', 'feature3']
categorical_features = ['category', 'region']
skewed_features = ['price', 'income']
# Create preprocessing pipeline
preprocessor = ColumnTransformer(
transformers=[
('numeric', Pipeline([
('imputer', SimpleImputer(strategy='median')),
('scaler', StandardScaler())
]), numeric_features),
('skewed', Pipeline([
('imputer', SimpleImputer(strategy='median')),
('transformer', PowerTransformer(method='yeo-johnson')),
('scaler', StandardScaler())
]), skewed_features),
('categorical', Pipeline([
('imputer', SimpleImputer(strategy='most_frequent')),
('encoder', OneHotEncoder(drop='first', sparse=False, handle_unknown='ignore'))
]), categorical_features)
],
remainder='drop'
)
# Feature selection
selector = SelectKBest(score_func=f_classif, k=10)
# Complete pipeline
pipeline = Pipeline([
('preprocessor', preprocessor),
('feature_selection', selector),
('model', LogisticRegression(random_state=42, max_iter=1000))
])
# Fit and evaluate
pipeline.fit(X_train, y_train)
score = pipeline.score(X_test, y_test)
print(f"Pipeline accuracy: {score:.3f}")
# Get feature names
feature_names_out = pipeline.named_steps['preprocessor'].get_feature_names_out()
selected_features = feature_names_out[pipeline.named_steps['feature_selection'].get_support()]
print(f"Selected features: {selected_features}")
Custom Transformers
from sklearn.base import BaseEstimator, TransformerMixin
class LogTransformer(BaseEstimator, TransformerMixin):
"""Custom log transformer"""
def fit(self, X, y=None):
return self
def transform(self, X):
return np.log1p(X)
class InteractionTransformer(BaseEstimator, TransformerMixin):
"""Create interaction features"""
def __init__(self, feature_indices):
self.feature_indices = feature_indices
def fit(self, X, y=None):
return self
def transform(self, X):
interacti>
for i, j in self.feature_indices:
interactions.append((X[:, i] * X[:, j]).reshape(-1, 1))
return np.hstack([X] + interactions)
# Use in pipeline
interacti>
('scaler', StandardScaler()),
('interactions', InteractionTransformer([(0, 1), (0, 2)])),
('model', LogisticRegression(random_state=42))
])
sklearn Deep Dive: Estimators, Mixins, and Composite Transformers
Understanding sklearn Architecture
sklearn uses a consistent interface for all estimators and transformers, built on base classes and mixins.
Base Classes
1. BaseEstimator:
Base class for all sklearn objects. Provides get_params() and set_params().
from sklearn.base import BaseEstimator
class CustomEstimator(BaseEstimator):
def __init__(self, param1=1, param2=2):
self.param1 = param1
self.param2 = param2
def get_params(self, deep=True):
"""Get parameters for this estimator"""
return {'param1': self.param1, 'param2': self.param2}
def set_params(self, **params):
"""Set parameters for this estimator"""
for key, value in params.items():
setattr(self, key, value)
return self
# Usage
estimator = CustomEstimator(param1=10, param2=20)
print(estimator.get_params()) # {'param1': 10, 'param2': 20}
estimator.set_params(param1=100)
print(estimator.param1) # 100
2. TransformerMixin:
Mixin for transformers. Provides fit_transform() method.
from sklearn.base import BaseEstimator, TransformerMixin
class CustomTransformer(BaseEstimator, TransformerMixin):
def __init__(self, multiplier=1.0):
self.multiplier = multiplier
def fit(self, X, y=None):
"""Fit transformer (learn parameters)"""
# Store any necessary statistics
self.mean_ = X.mean(axis=0)
return self
def transform(self, X):
"""Transform data"""
return (X - self.mean_) * self.multiplier
# fit_transform() is automatically provided by TransformerMixin
# It calls fit() then transform()
# Usage
transformer = CustomTransformer(multiplier=2.0)
X_transformed = transformer.fit_transform(X)
3. ClassifierMixin:
Mixin for classifiers. Provides score() method for classification.
from sklearn.base import BaseEstimator, ClassifierMixin
class CustomClassifier(BaseEstimator, ClassifierMixin):
def __init__(self, threshold=0.5):
self.threshold = threshold
def fit(self, X, y):
"""Train classifier"""
# Store training data (simple example)
self.X_train_ = X
self.y_train_ = y
return self
def predict(self, X):
"""Make predictions"""
# Simple nearest neighbor example
from sklearn.metrics.pairwise import euclidean_distances
distances = euclidean_distances(X, self.X_train_)
nearest_indices = distances.argmin(axis=1)
return self.y_train_[nearest_indices]
def predict_proba(self, X):
"""Predict probabilities"""
predicti>self.predict(X)
# Convert to probabilities (simplified)
proba = np.zeros((len(X), len(np.unique(self.y_train_))))
for i, pred in enumerate(predictions):
proba[i, pred] = 1.0
return proba
# score() is automatically provided by ClassifierMixin
# It uses accuracy_score by default
# Usage
classifier = CustomClassifier(threshold=0.5)
classifier.fit(X_train, y_train)
score = classifier.score(X_test, y_test) # Accuracy
4. RegressorMixin:
Mixin for regressors. Provides score() method for regression (R²).
from sklearn.base import BaseEstimator, RegressorMixin
class CustomRegressor(BaseEstimator, RegressorMixin):
def __init__(self, learning_rate=0.1):
self.learning_rate = learning_rate
def fit(self, X, y):
"""Train regressor"""
# Simple linear regression example
from sklearn.linear_model import LinearRegression
self.model_ = LinearRegression()
self.model_.fit(X, y)
return self
def predict(self, X):
"""Make predictions"""
return self.model_.predict(X)
# score() is automatically provided by RegressorMixin
# It uses r2_score by default
# Usage
regressor = CustomRegressor(learning_rate=0.1)
regressor.fit(X_train, y_train)
score = regressor.score(X_test, y_test) # R² score
Complete Custom Estimator Example
Custom Classifier with Full sklearn Interface:
from sklearn.base import BaseEstimator, ClassifierMixin
from sklearn.utils.validation import check_X_y, check_array
from sklearn.utils.multiclass import unique_labels
import numpy as np
class SimpleKNNClassifier(BaseEstimator, ClassifierMixin):
"""
Simple k-Nearest Neighbors Classifier
Implements full sklearn interface
"""
def __init__(self, n_neighbors=5, metric='euclidean'):
self.n_neighbors = n_neighbors
self.metric = metric
def fit(self, X, y):
"""
Fit the model
Parameters
----------
X : array-like of shape (n_samples, n_features)
Training data
y : array-like of shape (n_samples,)
Target values
Returns
-------
self : object
Returns self
"""
# Validate inputs
X, y = check_X_y(X, y)
# Store training data
self.X_train_ = X
self.y_train_ = y
# Store classes
self.classes_ = unique_labels(y)
self.n_classes_ = len(self.classes_)
return self
def predict(self, X):
"""
Predict class labels
Parameters
----------
X : array-like of shape (n_samples, n_features)
Test data
Returns
-------
y_pred : array of shape (n_samples,)
Predicted class labels
"""
# Validate that fit has been called
check_is_fitted(self)
# Validate input
X = check_array(X)
# Compute distances
from sklearn.metrics.pairwise import euclidean_distances
distances = euclidean_distances(X, self.X_train_)
# Find k nearest neighbors
nearest_indices = distances.argsort(axis=1)[:, :self.n_neighbors]
nearest_labels = self.y_train_[nearest_indices]
# Majority vote
predicti>
for labels in nearest_labels:
unique, counts = np.unique(labels, return_counts=True)
predictions.append(unique[np.argmax(counts)])
return np.array(predictions)
def predict_proba(self, X):
"""
Predict class probabilities
Parameters
----------
X : array-like of shape (n_samples, n_features)
Test data
Returns
-------
proba : array of shape (n_samples, n_classes_)
Class probabilities
"""
check_is_fitted(self)
X = check_array(X)
from sklearn.metrics.pairwise import euclidean_distances
distances = euclidean_distances(X, self.X_train_)
nearest_indices = distances.argsort(axis=1)[:, :self.n_neighbors]
nearest_labels = self.y_train_[nearest_indices]
# Calculate probabilities
proba = np.zeros((len(X), self.n_classes_))
for i, labels in enumerate(nearest_labels):
unique, counts = np.unique(labels, return_counts=True)
for j, label in enumerate(self.classes_):
if label in unique:
idx = np.where(unique == label)[0][0]
proba[i, j] = counts[idx] / self.n_neighbors
return proba
# Usage
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
X, y = make_classification(n_samples=1000, n_features=4, random_state=42)
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)
knn = SimpleKNNClassifier(n_neighbors=5)
knn.fit(X_train, y_train)
predicti>
probabilities = knn.predict_proba(X_test)
score = knn.score(X_test, y_test)
print(f"Accuracy: {score:.3f}")
Composite Transformers
1. FeatureUnion: Combine multiple transformers in parallel (all applied, results concatenated).
from sklearn.pipeline import FeatureUnion
from sklearn.decomposition import PCA
from sklearn.decomposition import TruncatedSVD
from sklearn.preprocessing import PolynomialFeatures
# Create FeatureUnion
feature_union = FeatureUnion([
('pca', PCA(n_components=5)),
('svd', TruncatedSVD(n_components=3)),
('poly', PolynomialFeatures(degree=2, include_bias=False))
])
# Fit and transform
X_combined = feature_union.fit_transform(X)
print(f"Original shape: {X.shape}")
print(f"Combined shape: {X_combined.shape}") # (n_samples, 5+3+n_poly_features)
# Get feature names
feature_names = feature_union.get_feature_names_out()
print(f"Feature names: {feature_names}")
2. Custom Composite Transformer:
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.pipeline import FeatureUnion
class CustomFeatureUnion(BaseEstimator, TransformerMixin):
"""Custom composite transformer"""
def __init__(self, transformers):
self.transformers = transformers
def fit(self, X, y=None):
"""Fit all transformers"""
for name, transformer in self.transformers:
transformer.fit(X, y)
return self
def transform(self, X):
"""Transform with all transformers and concatenate"""
results = []
for name, transformer in self.transformers:
transformed = transformer.transform(X)
results.append(transformed)
return np.hstack(results)
def get_feature_names_out(self, input_features=None):
"""Get feature names"""
feature_names = []
for name, transformer in self.transformers:
if hasattr(transformer, 'get_feature_names_out'):
names = transformer.get_feature_names_out(input_features)
feature_names.extend([f"{name}__{n}" for n in names])
else:
n_features = transformer.n_features_out_ if hasattr(transformer, 'n_features_out_') else X.shape[1]
feature_names.extend([f"{name}__feature_{i}" for i in range(n_features)])
return np.array(feature_names)
# Usage
custom_union = CustomFeatureUnion([
('scaler', StandardScaler()),
('pca', PCA(n_components=2))
])
X_transformed = custom_union.fit_transform(X)
3. Advanced FeatureUnion with Conditional Logic:
class ConditionalFeatureUnion(BaseEstimator, TransformerMixin):
"""FeatureUnion with conditional transformers"""
def __init__(self, transformers, selector=None):
self.transformers = transformers
self.selector = selector # Function to select which transformers to use
def fit(self, X, y=None):
"""Fit selected transformers"""
selected = self._select_transformers(X, y)
for name, transformer in selected:
transformer.fit(X, y)
return self
def transform(self, X):
"""Transform with selected transformers"""
selected = self._select_transformers(X, y=None)
results = []
for name, transformer in selected:
results.append(transformer.transform(X))
return np.hstack(results) if results else X
def _select_transformers(self, X, y):
"""Select transformers based on selector function"""
if self.selector is None:
return self.transformers
return [t for t in self.transformers if self.selector(t[0], X, y)]
# Usage with selector
def select_numeric_only(name, X, y):
"""Only use numeric transformers"""
return name in ['scaler', 'pca']
c>
transformers=[
('scaler', StandardScaler()),
('pca', PCA(n_components=2)),
('text_tfidf', TfidfVectorizer()) # Would be skipped
],
selector=select_numeric_only
)
Best Practices for Custom Estimators
1. Input Validation:
from sklearn.utils.validation import check_X_y, check_array, check_is_fitted
class ValidatedEstimator(BaseEstimator, ClassifierMixin):
def fit(self, X, y):
# Validate inputs
X, y = check_X_y(X, y)
# Store fitted state
self.is_fitted_ = True
return self
def predict(self, X):
# Check if fitted
check_is_fitted(self, 'is_fitted_')
# Validate input
X = check_array(X)
# Make predictions
return np.zeros(len(X))
2. Feature Names Support:
class NamedFeatureTransformer(BaseEstimator, TransformerMixin):
def fit(self, X, y=None):
if hasattr(X, 'columns'):
self.feature_names_in_ = X.columns.tolist()
return self
def transform(self, X):
return X
def get_feature_names_out(self, input_features=None):
if input_features is None:
input_features = self.feature_names_in_
return np.array([f"transformed_{name}" for name in input_features])
3. Metadata Routing:
from sklearn.utils.metadata_routing import get_routing_for_object
class MetadataAwareEstimator(BaseEstimator):
def fit(self, X, y, sample_weight=None):
# Handle sample_weight if provided
if sample_weight is not None:
# Use sample weights in training
pass
return self
Key Takeaways
- BaseEstimator: Provides parameter management
- Mixins: Add functionality (TransformerMixin, ClassifierMixin, RegressorMixin)
- FeatureUnion: Combine transformers in parallel
- Validation: Always use sklearn validation utilities
- Interface: Follow sklearn conventions for compatibility
Feature Importance and Selection
Multiple Selection Methods
from sklearn.feature_selection import (SelectKBest, RFE, RFECV,
SelectFromModel, mutual_info_classif)
from sklearn.ensemble import RandomForestClassifier
# Method 1: Univariate selection
selector_kbest = SelectKBest(score_func=f_classif, k=10)
X_kbest = selector_kbest.fit_transform(X, y)
# Method 2: Mutual information
selector_mi = SelectKBest(score_func=mutual_info_classif, k=10)
X_mi = selector_mi.fit_transform(X, y)
# Method 3: RFE
selector_rfe = RFE(
estimator=LogisticRegression(random_state=42, max_iter=1000),
n_features_to_select=10
)
X_rfe = selector_rfe.fit_transform(X, y)
# Method 4: Model-based selection
rf = RandomForestClassifier(n_estimators=100, random_state=42)
rf.fit(X, y)
selector_model = SelectFromModel(rf, threshold='median')
X_model = selector_model.fit_transform(X, y)
# Compare methods
methods = {
'KBest (F-test)': (X_kbest, selector_kbest),
'Mutual Information': (X_mi, selector_mi),
'RFE': (X_rfe, selector_rfe),
'Model-based': (X_model, selector_model)
}
for name, (X_selected, selector) in methods.items():
model = LogisticRegression(random_state=42, max_iter=1000)
scores = cross_val_score(model, X_selected, y, cv=5)
print(f"{name:25s}: {scores.mean():.3f} (+/- {scores.std():.3f})")
Feature Importance Analysis
# Multiple importance methods
importance_methods = {}
# Random Forest importance
rf = RandomForestClassifier(n_estimators=100, random_state=42)
rf.fit(X, y)
importance_methods['Random Forest'] = rf.feature_importances_
# XGBoost importance
try:
import xgboost as xgb
xgb_model = xgb.XGBClassifier(random_state=42)
xgb_model.fit(X, y)
importance_methods['XGBoost'] = xgb_model.feature_importances_
except ImportError:
pass
# Permutation importance
from sklearn.inspection import permutation_importance
perm_importance = permutation_importance(rf, X_test, y_test, n_repeats=10, random_state=42)
importance_methods['Permutation'] = perm_importance.importances_mean
# Compare
importance_df = pd.DataFrame(importance_methods, index=feature_names)
importance_df = importance_df.sort_values('Random Forest', ascending=False)
print("Feature Importance Comparison:")
print(importance_df.head(10))
Common Feature Engineering Pitfalls
Pitfall 1: Data Leakage
Problem: Using future information or target information in features
Solution:
# WRONG: Using target to create features
df['target_mean_by_category'] = df.groupby('category')['target'].transform('mean')
# CORRECT: Use cross-validation for target encoding
def target_encode_cv(df, cat_col, target_col, cv=5):
"""Target encoding with cross-validation"""
df_encoded = df.copy()
kf = KFold(n_splits=cv, shuffle=True, random_state=42)
for train_idx, val_idx in kf.split(df):
train_mean = df.iloc[train_idx].groupby(cat_col)[target_col].mean()
df_encoded.loc[val_idx, f'{cat_col}_target_encoded'] = \
df.loc[val_idx, cat_col].map(train_mean)
global_mean = df[target_col].mean()
df_encoded[f'{cat_col}_target_encoded'].fillna(global_mean, inplace=True)
return df_encoded
Pitfall 2: Scaling Before Split
Problem: Fitting scaler on entire dataset including test set
Solution:
# WRONG: Scale before split
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X) # Uses test data!
X_train, X_test = train_test_split(X_scaled, y)
# CORRECT: Scale after split
X_train, X_test, y_train, y_test = train_test_split(X, y)
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test) # Only transform!
Pitfall 3: Too Many Features
Problem: Creating too many features causes overfitting
Solution:
# Use feature selection
selector = SelectKBest(score_func=f_classif, k=20) # Limit features
X_selected = selector.fit_transform(X, y)
# Or use regularization
model = LogisticRegression(C=0.1, random_state=42) # Lower C = more regularization
Pitfall 4: Ignoring Feature Interactions
Problem: Missing important feature relationships
Solution:
# Create interaction features
poly = PolynomialFeatures(degree=2, interaction_only=True)
X_interaction = poly.fit_transform(X)
# Or manually create important interactions
df['important_interaction'] = df['feature1'] * df['feature2']
Key Takeaways
- Automation: Use tools like Featuretools for automated feature engineering
- Interactions: Create interaction and polynomial features
- Time Series: Extract temporal features, lags, and rolling statistics
- Text Features: Extract length, n-grams, and sentiment features
- Pipelines: Use pipelines to automate and prevent data leakage
- Feature Selection: Use multiple methods to find best features
- Avoid Pitfalls: Watch for data leakage, scaling issues, and overfitting
Next Steps
- Practice with domain-specific feature engineering
- Learn automated feature engineering tools
- Experiment with feature interactions
- Move to unsupervised learning module
Try next: Invent two features from domain knowledge and measure lift against a baseline.