"""Deterministic, stateless cleaning applied identically at train and serve time. Handles the 8 semantic categorical columns discovered in the data: F2230 - observation month (Oct25..Dec25) -> kept categorical F3886 - account type (17 levels) -> kept categorical F3888 - account open date (M-D-YYYY, 4292 vals) -> parsed to numeric age in days F3889 - activity-recency bucket (7 ordered) -> ordinal encoded F3890 - segment code (R/SU/M/U) -> kept categorical F3891 - occupation (7 levels) -> kept categorical F3892 - gender (M/F/O) -> kept categorical F3893 - RETAIL / CORPORATE -> kept categorical """ from __future__ import annotations import pandas as pd from src import config # Activity-recency bucket: smaller window = more recent activity. RECENCY_ORDER = {"L7D": 0, "L14D": 1, "L31D": 2, "L90D": 3, "L180D": 4, "L365D": 5, "G365D": 6} DATE_COL = "F3888" RECENCY_COL = "F3889" # Fixed reference date for reproducible account-age computation. REFERENCE_DATE = pd.Timestamp("2026-01-01") # Categorical columns left for one-hot encoding (after the special cases above). # Note: F2230 (month) is dropped as leakage (see config.LEAKAGE_EXCLUDE). CATEGORICAL_COLS = ["F3886", "F3890", "F3891", "F3892", "F3893"] def clean_frame(df: pd.DataFrame) -> pd.DataFrame: """Apply stateless transforms. Returns a new frame; does not mutate input.""" df = df.copy() # Drop label-adjacent leakage columns (see config.LEAKAGE_EXCLUDE). leak = [c for c in config.LEAKAGE_EXCLUDE if c in df.columns] if leak: df = df.drop(columns=leak) # F3888: parse open date -> account age in days (numeric). Drop original. if DATE_COL in df.columns: dt = pd.to_datetime(df[DATE_COL], format="%m-%d-%Y", errors="coerce") df["F3888_age_days"] = (REFERENCE_DATE - dt).dt.days df = df.drop(columns=[DATE_COL]) # F3889: ordinal-encode recency bucket. Drop original string. if RECENCY_COL in df.columns: df["F3889_recency_ord"] = df[RECENCY_COL].map(RECENCY_ORDER) df = df.drop(columns=[RECENCY_COL]) # Ensure remaining categorical columns are strings (for the one-hot encoder). for c in CATEGORICAL_COLS: if c in df.columns: df[c] = df[c].astype("object") return df def split_column_types(df: pd.DataFrame): """Return (numeric_cols, categorical_cols) present in a cleaned frame.""" cats = [c for c in CATEGORICAL_COLS if c in df.columns] nums = [c for c in df.columns if c not in cats] return nums, cats