¡Hola comunidad de Data Science! 
El feature engineering sigue siendo el factor más determinante en el éxito de proyectos de ML. Mientras que las arquitecturas de modelos se estandarizan, la creatividad en la construcción de features es lo que realmente diferencia proyectos promedio de soluciones excepcionales.
Feature Engineering Temporal Avanzado
Características Cíclicas para Variables Temporales
Las variables temporales requieren tratamiento especial para capturar patrones cíclicos:
import numpy as np
import pandas as pd
def crear_features_ciclicos(df, columna_tiempo):
"""Convierte timestamps en features cíclicos"""
# Extraer componentes temporales
df['hora'] = df[columna_tiempo].dt.hour
df['dia_semana'] = df[columna_tiempo].dt.dayofweek
df['mes'] = df[columna_tiempo].dt.month
df['dia_año'] = df[columna_tiempo].dt.dayofyear
# Transformaciones cíclicas (sin y cos para preservar continuidad)
df['hora_sin'] = np.sin(2 * np.pi * df['hora'] / 24)
df['hora_cos'] = np.cos(2 * np.pi * df['hora'] / 24)
df['dia_semana_sin'] = np.sin(2 * np.pi * df['dia_semana'] / 7)
df['dia_semana_cos'] = np.cos(2 * np.pi * df['dia_semana'] / 7)
df['mes_sin'] = np.sin(2 * np.pi * df['mes'] / 12)
df['mes_cos'] = np.cos(2 * np.pi * df['mes'] / 12)
# Features de estacionalidad
df['es_fin_semana'] = (df['dia_semana'] >= 5).astype(int)
df['es_hora_pico'] = ((df['hora'] >= 7) & (df['hora'] <= 9) |
(df['hora'] >= 17) & (df['hora'] <= 19)).astype(int)
return df
# Ejemplo de uso
df['timestamp'] = pd.to_datetime(df['timestamp'])
df = crear_features_ciclicos(df, 'timestamp')
Rolling Features y Lag Variables
def crear_features_temporales(df, valor_col, ventana=[7, 30, 90]):
"""Crea features basadas en ventanas temporales"""
df = df.sort_values('timestamp')
for v in ventana:
# Rolling statistics
df[f'{valor_col}_rolling_mean_{v}d'] = df[valor_col].rolling(window=v).mean()
df[f'{valor_col}_rolling_std_{v}d'] = df[valor_col].rolling(window=v).std()
df[f'{valor_col}_rolling_min_{v}d'] = df[valor_col].rolling(window=v).min()
df[f'{valor_col}_rolling_max_{v}d'] = df[valor_col].rolling(window=v).max()
# Lag features
df[f'{valor_col}_lag_{v}d'] = df[valor_col].shift(v)
# Rate of change
df[f'{valor_col}_pct_change_{v}d'] = df[valor_col].pct_change(periods=v)
# Diferencia con media móvil
df[f'{valor_col}_diff_from_rolling_{v}d'] = (
df[valor_col] - df[f'{valor_col}_rolling_mean_{v}d']
)
return df
Feature Engineering para Variables Categóricas
Target Encoding Robusto
from sklearn.model_selection import KFold
def target_encoding_robusto(X, y, columna_cat, n_folds=5, smoothing=10):
"""Target encoding con validación cruzada para evitar overfitting"""
kf = KFold(n_splits=n_folds, shuffle=True, random_state=42)
target_encoded = np.zeros(len(X))
# Media global para regularización
global_mean = y.mean()
for train_idx, val_idx in kf.split(X):
# Calcular medias por categoría en conjunto de entrenamiento
category_means = y.iloc[train_idx].groupby(X[columna_cat].iloc[train_idx]).mean()
category_counts = X[columna_cat].iloc[train_idx].value_counts()
# Aplicar smoothing
smoothed_means = (
(category_means * category_counts + global_mean * smoothing) /
(category_counts + smoothing)
)
# Mapear valores de validación
target_encoded[val_idx] = X[columna_cat].iloc[val_idx].map(smoothed_means).fillna(global_mean)
return target_encoded
# Ejemplo de uso
X['categoria_target_encoded'] = target_encoding_robusto(X, y, 'categoria')
Frequency Encoding y Rank Features
def crear_features_categoricos_avanzados(df, columna_cat):
"""Crea múltiples tipos de encoding para variables categóricas"""
# Frequency encoding
freq_map = df[columna_cat].value_counts().to_dict()
df[f'{columna_cat}_frequency'] = df[columna_cat].map(freq_map)
# Rank encoding (por frecuencia)
rank_map = df[columna_cat].value_counts().rank(method='dense', ascending=False).to_dict()
df[f'{columna_cat}_rank'] = df[columna_cat].map(rank_map)
# Rareza (inverso de frecuencia)
df[f'{columna_cat}_rarity'] = 1 / df[f'{columna_cat}_frequency']
# Encoding por percentiles de frecuencia
freq_percentiles = df[f'{columna_cat}_frequency'].rank(pct=True)
df[f'{columna_cat}_freq_percentile'] = freq_percentiles
return df
Feature Engineering Numérico Avanzado
Transformaciones Matemáticas Inteligentes
def transformaciones_numericas_avanzadas(df, columnas_numericas):
"""Aplica transformaciones matemáticas que capturan relaciones no lineales"""
for col in columnas_numericas:
# Transformaciones básicas
df[f'{col}_log'] = np.log1p(df[col].clip(lower=0))
df[f'{col}_sqrt'] = np.sqrt(df[col].clip(lower=0))
df[f'{col}_squared'] = df[col] ** 2
df[f'{col}_cubed'] = df[col] ** 3
# Transformaciones de normalización
df[f'{col}_standardized'] = (df[col] - df[col].mean()) / df[col].std()
df[f'{col}_min_max_scaled'] = (df[col] - df[col].min()) / (df[col].max() - df[col].min())
# Binning inteligente
df[f'{col}_quantile_bin'] = pd.qcut(df[col], q=10, labels=False, duplicates='drop')
# Outlier indicators
Q1 = df[col].quantile(0.25)
Q3 = df[col].quantile(0.75)
IQR = Q3 - Q1
df[f'{col}_is_outlier'] = (
(df[col] < Q1 - 1.5 * IQR) | (df[col] > Q3 + 1.5 * IQR)
).astype(int)
return df
Features de Interacción Automática
from itertools import combinations
def crear_features_interaccion(df, columnas_numericas, max_interacciones=20):
"""Crea features de interacción entre variables numéricas"""
interaction_features = []
# Combinaciones de 2 variables
for col1, col2 in combinations(columnas_numericas, 2):
# Multiplicación
feature_name = f'{col1}_x_{col2}'
df[feature_name] = df[col1] * df[col2]
interaction_features.append(feature_name)
# División (evitando división por cero)
feature_name = f'{col1}_div_{col2}'
df[feature_name] = df[col1] / (df[col2] + 1e-8)
interaction_features.append(feature_name)
# Diferencia
feature_name = f'{col1}_minus_{col2}'
df[feature_name] = df[col1] - df[col2]
interaction_features.append(feature_name)
# Suma
feature_name = f'{col1}_plus_{col2}'
df[feature_name] = df[col1] + df[col2]
interaction_features.append(feature_name)
if len(interaction_features) >= max_interacciones:
break
return df, interaction_features
Feature Selection Inteligente
Selección por Importancia y Correlación
import seaborn as sns
import matplotlib.pyplot as plt
from sklearn.feature_selection import mutual_info_regression
from sklearn.ensemble import RandomForestRegressor
def seleccion_features_avanzada(X, y, umbral_correlacion=0.95):
"""Selecciona features usando múltiples criterios"""
# 1. Eliminar features con baja varianza
from sklearn.feature_selection import VarianceThreshold
selector_var = VarianceThreshold(threshold=0.01)
X_var_filtered = pd.DataFrame(
selector_var.fit_transform(X),
columns=X.columns[selector_var.get_support()]
)
# 2. Eliminar features altamente correlacionadas
corr_matrix = X_var_filtered.corr().abs()
upper_triangle = corr_matrix.where(
np.triu(np.ones_like(corr_matrix, dtype=bool), k=1)
)
features_to_drop = [
column for column in upper_triangle.columns
if any(upper_triangle[column] > umbral_correlacion)
]
X_corr_filtered = X_var_filtered.drop(columns=features_to_drop)
# 3. Selección por mutual information
mi_scores = mutual_info_regression(X_corr_filtered, y)
mi_scores = pd.Series(mi_scores, index=X_corr_filtered.columns)
# 4. Importancia con Random Forest
rf = RandomForestRegressor(n_estimators=100, random_state=42)
rf.fit(X_corr_filtered, y)
rf_importance = pd.Series(rf.feature_importances_, index=X_corr_filtered.columns)
# 5. Combinar scores
feature_scores = pd.DataFrame({
'mutual_info': mi_scores,
'rf_importance': rf_importance,
'combined_score': mi_scores * rf_importance
}).sort_values('combined_score', ascending=False)
return feature_scores, X_corr_filtered
def plot_feature_importance(feature_scores, top_n=20):
"""Visualiza importancia de features"""
top_features = feature_scores.head(top_n)
fig, (ax1, ax2) = plt.subplots(1, 2, figsize=(15, 6))
# Mutual Information
top_features['mutual_info'].plot(kind='barh', ax=ax1, color='skyblue')
ax1.set_title('Top Features - Mutual Information')
ax1.set_xlabel('Mutual Information Score')
# Random Forest Importance
top_features['rf_importance'].plot(kind='barh', ax=ax2, color='lightcoral')
ax2.set_title('Top Features - Random Forest Importance')
ax2.set_xlabel('Feature Importance')
plt.tight_layout()
plt.show()
Feature Engineering con Embeddings
Embeddings para Variables Categóricas de Alta Cardinalidad
from sklearn.decomposition import TruncatedSVD
from sklearn.feature_extraction.text import TfidfVectorizer
def crear_embeddings_categoricos(df, columna_cat, dim_embedding=50):
"""Crea embeddings para variables categóricas usando co-ocurrencia"""
# Crear documentos basados en co-ocurrencia
documentos = df.groupby('id')[columna_cat].apply(lambda x: ' '.join(x.astype(str))).values
# TF-IDF Vectorization
vectorizer = TfidfVectorizer(max_features=1000, ngram_range=(1, 2))
tfidf_matrix = vectorizer.fit_transform(documentos)
# Reducción dimensional con SVD
svd = TruncatedSVD(n_components=dim_embedding, random_state=42)
embeddings = svd.fit_transform(tfidf_matrix)
# Crear mapeo de categoría a embedding
categorias_unicas = df[columna_cat].unique()
categoria_to_embedding = {}
for categoria in categorias_unicas:
# Promedio de embeddings donde aparece la categoría
indices = df[df[columna_cat] == categoria]['id'].unique()
embedding_promedio = embeddings[indices].mean(axis=0)
categoria_to_embedding[categoria] = embedding_promedio
# Agregar embeddings como features
for i in range(dim_embedding):
df[f'{columna_cat}_embed_{i}'] = df[columna_cat].map(
lambda x: categoria_to_embedding[x][i]
)
return df, categoria_to_embedding
Pipeline de Feature Engineering Automatizado
Clase para Automatizar el Proceso
class FeatureEngineeringPipeline:
def __init__(self):
self.target_encoders = {}
self.scalers = {}
self.feature_names = []
def fit_transform(self, X, y):
"""Ajusta y transforma features en un pipeline"""
X_processed = X.copy()
# Identificar tipos de columnas
numeric_cols = X_processed.select_dtypes(include=[np.number]).columns
categorical_cols = X_processed.select_dtypes(include=['object', 'category']).columns
datetime_cols = X_processed.select_dtypes(include=['datetime64']).columns
# Feature engineering temporal
for col in datetime_cols:
X_processed = crear_features_ciclicos(X_processed, col)
# Feature engineering numérico
if len(numeric_cols) > 0:
X_processed = transformaciones_numericas_avanzadas(X_processed, numeric_cols)
X_processed, _ = crear_features_interaccion(X_processed, numeric_cols[:5]) # Limitar interacciones
# Feature engineering categórico
for col in categorical_cols:
X_processed = crear_features_categoricos_avanzados(X_processed, col)
X_processed[f'{col}_target_encoded'] = target_encoding_robusto(X_processed, y, col)
# Selección de features
feature_scores, X_processed = seleccion_features_avanzada(X_processed, y)
top_features = feature_scores.head(100).index.tolist() # Top 100 features
X_processed = X_processed[top_features]
self.feature_names = X_processed.columns.tolist()
return X_processed
def transform(self, X):
"""Transforma nuevos datos usando parámetros ajustados"""
# Implementar transformación usando parámetros guardados
pass
Tips Prácticos para Implementación
1. Validación de Features
- Cross-validation: Siempre valida nuevos features con CV para evitar overfitting
- Feature importance: Usa SHAP values para entender el impacto real de cada feature
- Temporal validation: Para datos con componente temporal, usa validación temporal en lugar de aleatoria
2. Optimización de Performance
- Lazy evaluation: Calcula features costosos solo cuando sea necesario
- Caching: Guarda features intermedios para reutilizar en experimentos
- Paralelización: Usa
joblibomultiprocessingpara crear features en paralelo
3. Monitoreo en Producción
- Feature drift: Monitorea distribuciones de features en producción
- Data quality: Implementa checks automáticos para valores faltantes y outliers
- Performance tracking: Rastrea cómo nuevos features afectan métricas de negocio
Consideraciones Clave
El feature engineering efectivo requiere:
- Conocimiento del dominio: Las mejores features vienen de entender el problema
- Experimentación sistemática: Documenta qué funciona y qué no
- Balance: Más features no siempre es mejor; calidad sobre cantidad
- Reproducibilidad: Usa pipelines que puedan recrear features consistentemente
¿Qué técnicas de feature engineering han tenido mayor impacto en sus proyectos? ¿Algún approach innovador que hayan desarrollado para dominios específicos?
#FeatureEngineering machinelearning #DataScience mlops #DataPreprocessing