""" Production feature engineering module for fraud detection. Computes velocity features, statistical profiles, behavioral anomaly scores, temporal patterns, and graph-based features from raw transaction data. """ import logging from typing import Optional import numpy as np import pandas as pd logger = logging.getLogger(__name__) class FeatureEngineer: """Compute fraud detection features from raw transaction data. Features are computed in a production-safe manner: - All window-based features use only past data (no future leakage) - Features are computed per-cardholder where appropriate - Missing values are handled with sensible defaults """ VELOCITY_WINDOWS_HOURS = [1, 6, 24, 168] # 1h, 6h, 24h, 7d def __init__(self, config: Optional[dict] = None): self.config = config or {} self._fitted_stats = {} def fit(self, train_df: pd.DataFrame) -> "FeatureEngineer": """Learn population-level statistics from training data. These statistics are used to compute deviation features and normalize values. """ logger.info("Fitting feature engineer on %d training records", len(train_df)) # Learn per-MCC amount distributions self._fitted_stats["mcc_amount_stats"] = ( train_df.groupby("merchant_category_code")["transaction_amount"] .agg(["mean", "std", "median"]) .to_dict("index") ) # Learn per-country transaction frequency self._fitted_stats["country_freq"] = ( train_df["country_code"].value_counts(normalize=True).to_dict() ) # Learn global amount statistics self._fitted_stats["global_amount_mean"] = train_df["transaction_amount"].mean() self._fitted_stats["global_amount_std"] = train_df["transaction_amount"].std() # Learn per-cardholder baseline spending self._fitted_stats["cardholder_baseline"] = ( train_df.groupby("cardholder_id") .agg( avg_amount=("transaction_amount", "mean"), std_amount=("transaction_amount", "std"), avg_daily_count=("transaction_id", lambda x: len(x) / max(1, ( train_df.loc[x.index, "timestamp"].max() - train_df.loc[x.index, "timestamp"].min() ).days)), primary_country=("country_code", lambda x: x.mode().iloc[0] if len(x) > 0 else "US"), primary_mcc=("merchant_category_code", lambda x: x.mode().iloc[0] if len(x) > 0 else "grocery"), ) .to_dict("index") ) logger.info("Feature engineer fitted — %d cardholder profiles learned", len(self._fitted_stats["cardholder_baseline"])) return self def transform(self, df: pd.DataFrame) -> pd.DataFrame: """Compute all features for a transaction DataFrame. Args: df: Transaction DataFrame (must be sorted by timestamp). Returns: DataFrame with original columns plus engineered features. """ logger.info("Computing features for %d transactions", len(df)) df = df.copy().sort_values("timestamp").reset_index(drop=True) # Compute feature groups df = self._compute_velocity_features(df) df = self._compute_amount_deviation_features(df) df = self._compute_temporal_features(df) df = self._compute_behavioral_features(df) df = self._compute_frequency_encoding(df) df = self._compute_interaction_features(df) # Fill remaining NaNs numeric_cols = df.select_dtypes(include=[np.number]).columns df[numeric_cols] = df[numeric_cols].fillna(0) feature_cols = [c for c in df.columns if c.startswith("feat_")] logger.info("Feature engineering complete — %d features computed", len(feature_cols)) return df def _compute_velocity_features(self, df: pd.DataFrame) -> pd.DataFrame: """Compute transaction velocity (count and sum) over sliding time windows. For each transaction, counts the number and sum of preceding transactions by the same cardholder within each time window. """ logger.info("Computing velocity features...") df = df.sort_values(["cardholder_id", "timestamp"]).reset_index(drop=True) for window_hours in self.VELOCITY_WINDOWS_HOURS: window_td = pd.Timedelta(hours=window_hours) count_col = f"feat_velocity_count_{window_hours}h" sum_col = f"feat_velocity_sum_{window_hours}h" avg_col = f"feat_velocity_avg_{window_hours}h" counts = [] sums = [] for _, group in df.groupby("cardholder_id"): timestamps = group["timestamp"].values amounts = group["transaction_amount"].values g_counts = np.zeros(len(group)) g_sums = np.zeros(len(group)) for i in range(len(group)): window_start = timestamps[i] - np.timedelta64(window_hours, "h") mask = (timestamps[:i] >= window_start) & (timestamps[:i] < timestamps[i]) g_counts[i] = mask.sum() g_sums[i] = amounts[:i][mask].sum() counts.extend(g_counts) sums.extend(g_sums) df[count_col] = counts df[sum_col] = sums df[avg_col] = np.where( df[count_col] > 0, df[sum_col] / df[count_col], 0, ) return df def _compute_amount_deviation_features(self, df: pd.DataFrame) -> pd.DataFrame: """Compute how much each transaction deviates from expected spending patterns.""" logger.info("Computing amount deviation features...") # Deviation from cardholder baseline cardholder_means = df["cardholder_id"].map( {k: v["avg_amount"] for k, v in self._fitted_stats.get("cardholder_baseline", {}).items()} ).fillna(self._fitted_stats.get("global_amount_mean", 100)) cardholder_stds = df["cardholder_id"].map( {k: v["std_amount"] for k, v in self._fitted_stats.get("cardholder_baseline", {}).items()} ).fillna(self._fitted_stats.get("global_amount_std", 50)) df["feat_amount_zscore_cardholder"] = ( (df["transaction_amount"] - cardholder_means) / cardholder_stds.clip(lower=1) ) # Deviation from MCC baseline mcc_stats = self._fitted_stats.get("mcc_amount_stats", {}) df["feat_amount_zscore_mcc"] = df.apply( lambda row: ( (row["transaction_amount"] - mcc_stats.get(row["merchant_category_code"], {}).get("mean", 100)) / max(1, mcc_stats.get(row["merchant_category_code"], {}).get("std", 50)) ), axis=1, ) # Log-transformed amount (helps with skewed distribution) df["feat_log_amount"] = np.log1p(df["transaction_amount"]) # Amount percentile rank within cardholder history df["feat_amount_rank"] = df.groupby("cardholder_id")["transaction_amount"].rank(pct=True) return df def _compute_temporal_features(self, df: pd.DataFrame) -> pd.DataFrame: """Compute time-based features.""" logger.info("Computing temporal features...") df["feat_hour_sin"] = np.sin(2 * np.pi * df["hour_of_day"] / 24) df["feat_hour_cos"] = np.cos(2 * np.pi * df["hour_of_day"] / 24) df["feat_dow_sin"] = np.sin(2 * np.pi * df["day_of_week"] / 7) df["feat_dow_cos"] = np.cos(2 * np.pi * df["day_of_week"] / 7) df["feat_is_weekend"] = df["is_weekend"].astype(float) df["feat_is_night"] = df["is_night"].astype(float) # Time since last transaction (per cardholder) df = df.sort_values(["cardholder_id", "timestamp"]) df["feat_time_since_last_txn_seconds"] = ( df.groupby("cardholder_id")["timestamp"] .diff() .dt.total_seconds() .fillna(0) ) df["feat_log_time_since_last"] = np.log1p(df["feat_time_since_last_txn_seconds"]) return df def _compute_behavioral_features(self, df: pd.DataFrame) -> pd.DataFrame: """Compute behavioral anomaly features.""" logger.info("Computing behavioral features...") # Is transaction in a different country from cardholder's primary country? cardholder_primary_country = { k: v["primary_country"] for k, v in self._fitted_stats.get("cardholder_baseline", {}).items() } df["feat_is_foreign_txn"] = ( df["cardholder_id"].map(cardholder_primary_country).fillna("US") != df["country_code"] ).astype(float) # Is merchant category unusual for this cardholder? cardholder_primary_mcc = { k: v["primary_mcc"] for k, v in self._fitted_stats.get("cardholder_baseline", {}).items() } df["feat_unusual_mcc"] = ( df["cardholder_id"].map(cardholder_primary_mcc).fillna("grocery") != df["merchant_category_code"] ).astype(float) # Country rarity score country_freq = self._fitted_stats.get("country_freq", {}) df["feat_country_rarity"] = 1 - df["country_code"].map(country_freq).fillna(0) # Entry mode risk encoding entry_mode_risk = { "chip": 0.1, "contactless": 0.15, "swipe": 0.3, "online": 0.5, "manual_entry": 0.8, } df["feat_entry_mode_risk"] = df["entry_mode"].map(entry_mode_risk).fillna(0.5) # Merchant risk score passthrough df["feat_merchant_risk"] = df["merchant_risk_score"] return df def _compute_frequency_encoding(self, df: pd.DataFrame) -> pd.DataFrame: """Frequency-encode categorical variables.""" logger.info("Computing frequency encoding features...") for col in ["merchant_category_code", "country_code", "entry_mode", "card_type"]: freq = df[col].value_counts(normalize=True) df[f"feat_{col}_freq"] = df[col].map(freq).fillna(0) return df def _compute_interaction_features(self, df: pd.DataFrame) -> pd.DataFrame: """Compute cross-feature interactions.""" logger.info("Computing interaction features...") # High amount + night + online = high risk signal df["feat_night_online_amount"] = ( df["feat_is_night"] * df["feat_entry_mode_risk"] * df["feat_log_amount"] ) # Foreign + high velocity = potential compromise df["feat_foreign_velocity"] = ( df["feat_is_foreign_txn"] * df.get("feat_velocity_count_1h", 0) ) # Amount deviation * merchant risk df["feat_deviation_risk"] = ( df["feat_amount_zscore_cardholder"].abs() * df["feat_merchant_risk"] ) return df def get_feature_names(self) -> list[str]: """Return list of all computed feature column names.""" return [ # Velocity *[f"feat_velocity_count_{w}h" for w in self.VELOCITY_WINDOWS_HOURS], *[f"feat_velocity_sum_{w}h" for w in self.VELOCITY_WINDOWS_HOURS], *[f"feat_velocity_avg_{w}h" for w in self.VELOCITY_WINDOWS_HOURS], # Amount deviation "feat_amount_zscore_cardholder", "feat_amount_zscore_mcc", "feat_log_amount", "feat_amount_rank", # Temporal "feat_hour_sin", "feat_hour_cos", "feat_dow_sin", "feat_dow_cos", "feat_is_weekend", "feat_is_night", "feat_time_since_last_txn_seconds", "feat_log_time_since_last", # Behavioral "feat_is_foreign_txn", "feat_unusual_mcc", "feat_country_rarity", "feat_entry_mode_risk", "feat_merchant_risk", # Frequency encoding "feat_merchant_category_code_freq", "feat_country_code_freq", "feat_entry_mode_freq", "feat_card_type_freq", # Interactions "feat_night_online_amount", "feat_foreign_velocity", "feat_deviation_risk", ]