Feature Engineer

from sklearn.preprocessing import StandardScaler, MinMaxScaler, RobustScaler

# StandardScaler: zero mean, unit variance
# Best for: Normally distributed features, SVMs, logistic regression
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)

# MinMaxScaler: scale to [0, 1]
# Best for: Neural networks, features with bounded ranges
scaler = MinMaxScaler(feature_range=(0, 1))
X_scaled = scaler.fit_transform(X)

# RobustScaler: uses median and IQR (robust to outliers)
# Best for: Data with significant outliers
scaler = RobustScaler()
X_scaled = scaler.fit_transform(X)

Does your model require scaling?
  Tree-based models (XGBoost, RF, LightGBM): NO (not needed)
  Linear models, SVMs, KNN: YES
  Neural networks: YES

Has outliers?
  YES -> RobustScaler
  NO  -> Continue

Need bounded output?
  YES -> MinMaxScaler
  NO  -> StandardScaler (default)

import numpy as np
from sklearn.preprocessing import PowerTransformer

def handle_skewed_features(df, columns, threshold=1.0):
    """Apply log transform to highly skewed features."""
    transforms = {}

    for col in columns:
        skewness = df[col].skew()

        if abs(skewness) > threshold:
            if (df[col] > 0).all():
                # Log transform for positive values
                df[f"{col}_log"] = np.log1p(df[col])
                # ... (condensed) ...
    return df, transforms

# Box-Cox (requires strictly positive data)
pt = PowerTransformer(method="box-cox")
X_transformed = pt.fit_transform(X_positive)

from sklearn.preprocessing import KBinsDiscretizer

# Equal-width binning
binner = KBinsDiscretizer(n_bins=10, encode="ordinal", strategy="uniform")
X_binned = binner.fit_transform(X)

# Equal-frequency (quantile) binning
binner = KBinsDiscretizer(n_bins=10, encode="ordinal", strategy="quantile")
X_binned = binner.fit_transform(X)

# Custom bins
def custom_bin_age(age: float) -> str:
    if age < 18: return "minor"
    elif age < 30: return "young_adult"
    elif age < 50: return "middle_age"
    elif age < 65: return "senior"
    else: return "elderly"

df["age_group"] = df["age"].apply(custom_bin_age)

from sklearn.preprocessing import PolynomialFeatures

# Polynomial interactions
poly = PolynomialFeatures(degree=2, interaction_only=True, include_bias=False)
X_interactions = poly.fit_transform(X[["feature_a", "feature_b", "feature_c"]])

# Manual domain-specific interactions
df["price_per_sqft"] = df["price"] / df["sqft"].clip(lower=1)
df["bmi"] = df["weight_kg"] / (df["height_m"] ** 2)
df["debt_to_income"] = df["total_debt"] / df["annual_income"].clip(lower=1)

from sklearn.preprocessing import OneHotEncoder
import pandas as pd

# Sklearn
encoder = OneHotEncoder(sparse_output=False, handle_unknown="ignore", drop="first")
X_encoded = encoder.fit_transform(df[["color", "size"]])

# Pandas (quick prototyping)
df_encoded = pd.get_dummies(df, columns=["color", "size"], drop_first=True)

from category_encoders import TargetEncoder
from sklearn.model_selection import KFold

def target_encode_with_smoothing(
    train_df: pd.DataFrame,
    test_df: pd.DataFrame,
    columns: list[str],
    target: str,
    smoothing: float = 10.0,
) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Target encoding with smoothing to prevent overfitting."""

    encoder = TargetEncoder(
        cols=columns,
        # ... (condensed) ...
        stats = train.groupby(col)[target].agg(["mean", "count"])
        smooth_mean = (stats["mean"] * stats["count"] + global_mean * smoothing) / (stats["count"] + smoothing)
        encoded.iloc[val_idx] = df.iloc[val_idx][col].map(smooth_mean).fillna(global_mean)

    return encoded

from sklearn.preprocessing import OrdinalEncoder

# When categories have natural order
size_order = ["XS", "S", "M", "L", "XL", "XXL"]
education_order = ["high_school", "bachelors", "masters", "phd"]

encoder = OrdinalEncoder(
    categories=[size_order, education_order],
    handle_unknown="use_encoded_value",
    unknown_value=-1,
)
X_encoded = encoder.fit_transform(df[["size", "education"]])

def frequency_encode(df: pd.DataFrame, columns: list[str]) -> pd.DataFrame:
    """Encode categorical by frequency of occurrence."""
    for col in columns:
        freq = df[col].value_counts(normalize=True)
        df[f"{col}_freq"] = df[col].map(freq)
    return df

from sklearn.feature_extraction.text import TfidfVectorizer

# Basic TF-IDF
vectorizer = TfidfVectorizer(
    max_features=5000,
    min_df=2,
    max_df=0.95,
    ngram_range=(1, 2),
    sublinear_tf=True,
    stop_words="english",
)
X_tfidf = vectorizer.fit_transform(texts)

import numpy as np

def extract_text_features(text: str) -> dict:
    """Extract statistical features from text."""
    words = text.split()
    sentences = text.split(".")

    return {
        "char_count": len(text),
        "word_count": len(words),
        "sentence_count": len(sentences),
        "avg_word_length": np.mean([len(w) for w in words]) if words else 0,
        "avg_sentence_length": np.mean([len(s.split()) for s in sentences if s.strip()]),
        "unique_word_ratio": len(set(words)) / len(words) if words else 0,
        "uppercase_ratio": sum(1 for c in text if c.isupper()) / len(text) if text else 0,
        "digit_ratio": sum(1 for c in text if c.isdigit()) / len(text) if text else 0,
        "punctuation_count": sum(1 for c in text if c in ".,;:!?"),
        "has_question_mark": int("?" in text),
        "has_exclamation": int("!" in text),
    }

from sentence_transformers import SentenceTransformer
import numpy as np

def text_to_embedding_features(
    texts: list[str],
    model_name: str = "all-MiniLM-L6-v2",
) -> np.ndarray:
    """Convert text to dense embedding features."""
    model = SentenceTransformer(model_name)
    embeddings = model.encode(texts, show_progress_bar=True, batch_size=64)
    return embeddings  # Shape: (n_texts, 384)

import pandas as pd

def extract_datetime_features(df: pd.DataFrame, date_col: str) -> pd.DataFrame:
    """Extract comprehensive temporal features."""
    dt = pd.to_datetime(df[date_col])

    df[f"{date_col}_year"] = dt.dt.year
    df[f"{date_col}_month"] = dt.dt.month
    df[f"{date_col}_day"] = dt.dt.day
    df[f"{date_col}_dayofweek"] = dt.dt.dayofweek
    df[f"{date_col}_hour"] = dt.dt.hour
    df[f"{date_col}_minute"] = dt.dt.minute
    df[f"{date_col}_quarter"] = dt.dt.quarter
    df[f"{date_col}_is_weekend"] = dt.dt.dayofweek.isin([5, 6]).astype(int)
    # ... (condensed) ...
    df[f"{date_col}_hour_cos"] = np.cos(2 * np.pi * dt.dt.hour / 24)
    df[f"{date_col}_dow_sin"] = np.sin(2 * np.pi * dt.dt.dayofweek / 7)
    df[f"{date_col}_dow_cos"] = np.cos(2 * np.pi * dt.dt.dayofweek / 7)

    return df

def create_lag_features(
    df: pd.DataFrame,
    target_col: str,
    lags: list[int] = [1, 7, 14, 28],
    rolling_windows: list[int] = [7, 14, 30],
) -> pd.DataFrame:
    """Create lag and rolling window features for time series."""

    # Lag features
    for lag in lags:
        df[f"{target_col}_lag_{lag}"] = df[target_col].shift(lag)

    # Rolling statistics
    for window in rolling_windows:
        # ... (condensed) ...
    # Differences
    df[f"{target_col}_diff_1"] = df[target_col].diff(1)
    df[f"{target_col}_pct_change"] = df[target_col].pct_change()

    return df

from sklearn.feature_selection import (
    SelectKBest, f_classif, mutual_info_classif,
    f_regression, mutual_info_regression,
)

def filter_features(
    X, y,
    task: str = "classification",
    k: int = 20,
    method: str = "mutual_info",
) -> list[str]:
    """Select top-k features using filter methods."""

    if task == "classification":
        # ... (condensed) ...
        "score": selector.scores_,
    }).sort_values("score", ascending=False)

    selected = scores.head(k)["feature"].tolist()
    return selected

from sklearn.feature_selection import RFECV
from sklearn.ensemble import RandomForestClassifier

def recursive_feature_selection(X, y, min_features: int = 5) -> list[str]:
    """Recursive feature elimination with cross-validation."""
    model = RandomForestClassifier(n_estimators=100, random_state=42)

    rfecv = RFECV(
        estimator=model,
        step=1,
        cv=5,
        scoring="f1",
        min_features_to_select=min_features,
        n_jobs=-1,
    )
    rfecv.fit(X, y)

    selected = X.columns[rfecv.support_].tolist()
    print(f"Optimal features: {rfecv.n_features_}")
    return selected

def get_feature_importance(model, feature_names: list[str], X_test=None, y_test=None) -> pd.DataFrame:
    """Extract feature importance from trained models."""

    # Native importance (tree-based models)
    if hasattr(model, "feature_importances_"):
        importances = model.feature_importances_
    elif X_test is not None and y_test is not None:
        # Permutation importance (model-agnostic)
        from sklearn.inspection import permutation_importance
        result = permutation_importance(model, X_test, y_test, n_repeats=10)
        importances = result.importances_mean
    else:
        raise ValueError("Model has no feature_importances_ and no test data provided")

    # ... (condensed) ...
        "feature": X.columns,
        "mean_abs_shap": np.abs(shap_values).mean(axis=0),
    }).sort_values("mean_abs_shap", ascending=False)

    return importance

from feast import FeatureStore, Entity, Feature, FeatureView, FileSource
from feast.types import Float32, Int64
from datetime import timedelta

# Define data source
driver_stats_source = FileSource(
    path="data/driver_stats.parquet",
    timestamp_field="event_timestamp",
    created_timestamp_column="created",
)

# Define entity
driver = Entity(
    name="driver_id",
    # ... (condensed) ...
# Get online features for inference
features = store.get_online_features(
    features=["driver_stats:conv_rate", "driver_stats:acc_rate"],
    entity_rows=[{"driver_id": 1001}],
).to_dict()

import featuretools as ft

def auto_feature_engineer(
    df: pd.DataFrame,
    entity_id: str,
    max_depth: int = 2,
) -> pd.DataFrame:
    """Automated feature engineering with Featuretools."""

    es = ft.EntitySet(id="dataset")
    es = es.add_dataframe(
        dataframe_name="main",
        dataframe=df,
        index=entity_id,
    # ... (condensed) ...
        trans_primitives=["day", "month", "year", "weekday", "is_weekend"],
    )

    print(f"Generated {len(feature_defs)} features")
    return feature_matrix

from sklearn.pipeline import Pipeline
from sklearn.compose import ColumnTransformer
from sklearn.preprocessing import StandardScaler, OneHotEncoder, OrdinalEncoder
from sklearn.impute import SimpleImputer

def build_feature_pipeline(config: dict) -> Pipeline:
    """Build production feature engineering pipeline."""

    numeric_transformer = Pipeline([
        ("imputer", SimpleImputer(strategy="median")),
        ("scaler", StandardScaler()),
    ])

    categorical_low_card = Pipeline([
        # ... (condensed) ...

    return Pipeline([
        ("preprocessor", preprocessor),
        ("model", config["model"]),
    ])