src/features.py

"""Feature engineering for Banco Ripley credit-scoring.

Builds derived features on top of the cleaned dataset, including:

* Temporal trend ratios comparing short vs. long windows (3 / 6 / 12 months).
* Behavioural volatility indicators (good-month vs bad-month deltas).
* Interaction terms between high-importance financial ratios.
* Aggregate risk scores for flag bundles.
"""

from pathlib import Path

import numpy as np
import pandas as pd
from loguru import logger
from tqdm import tqdm

from src.config import PROJ_ROOT


# ── Individual feature groups ─────────────────────────────────────────────────


def add_trend_ratios(df: pd.DataFrame) -> pd.DataFrame:
    """Add short-to-long window trend ratios for key financial metrics.

    For each metric available in both 3-month and 12-month windows the
    function computes ``metric_3 / (metric_12 + ε)`` as a trend signal.
    A ratio > 1 indicates recent growth; < 1 indicates deterioration.

    Args:
        df: Processed ``DataFrame`` containing raw window features.

    Returns:
        ``DataFrame`` augmented with ``TREND_<metric>_3_12`` columns.
    """
    df = df.copy()
    eps = 1e-6

    pairs = [
        ("PROM_UTIL_TARJ3", "PROM_UTIL_TARJ12", "TREND_UTIL_TARJ_3_12"),
        ("PROM_UTIL_COMP3", "PROM_UTIL_COMP12", "TREND_UTIL_COMP_3_12"),
        ("PROM_UTIL_EFEC3", "PROM_UTIL_EFEC12", "TREND_UTIL_EFEC_3_12"),
        ("PROM_RATIO_PRES3", "PROM_RATIO_PRES12", "TREND_RATIO_PRES_3_12"),
        ("PROM_LINEA3", "PROM_LINEA12", "TREND_LINEA_3_12"),
        ("PROM_DMES_DTOTAL3", "PROM_DMES_DTOTAL12", "TREND_DMES_DTOTAL_3_12"),
        ("PROM_DOTROS_DTOTAL3", "PROM_DOTROS_DTOTAL12", "TREND_DOTROS_DTOTAL_3_12"),
    ]

    for col_3, col_12, new_col in pairs:
        if col_3 in df.columns and col_12 in df.columns:
            df[new_col] = df[col_3] / (df[col_12].abs() + eps)

    return df


def add_volatility_scores(df: pd.DataFrame) -> pd.DataFrame:
    """Add composite behavioural volatility scores across windows.

    Combines ``DIF_BUE_MAL*`` fields into a single risk trajectory score.
    A high score means the client's payment behaviour has been consistently
    good; a low (negative) score signals recent deterioration.

    Args:
        df: Processed ``DataFrame``.

    Returns:
        ``DataFrame`` with ``VOLATILITY_SCORE_3``, ``VOLATILITY_SCORE_6``
        and ``VOLATILITY_SCORE_12`` columns.
    """
    df = df.copy()

    if "DIF_BUE_MAL3" in df.columns and "DIF_BUE_MAL100_3" in df.columns:
        df["VOLATILITY_SCORE_3"] = df["DIF_BUE_MAL3"] + df["DIF_BUE_MAL100_3"]

    if "DIF_BUE_MAL6" in df.columns and "DIF_BUE_MAL100_6" in df.columns:
        df["VOLATILITY_SCORE_6"] = df["DIF_BUE_MAL6"] + df["DIF_BUE_MAL100_6"]

    if "DIF_BUE_MAL12" in df.columns and "DIF_BUE_MAL100_12" in df.columns:
        df["VOLATILITY_SCORE_12"] = df["DIF_BUE_MAL12"] + df["DIF_BUE_MAL100_12"]

    return df


def add_mark_aggregates(df: pd.DataFrame) -> pd.DataFrame:
    """Add aggregate risk-flag counts per observation.

    Counts how many product-holding / risk marks are active at each
    window (current, 3m, 6m, 12m) and exposes them as a numeric signal.

    Args:
        df: Processed ``DataFrame``.

    Returns:
        ``DataFrame`` with ``N_MARKS_CURRENT``, ``N_MARKS_3M``,
        ``N_MARKS_6M`` and ``N_MARKS_12M`` columns.
    """
    df = df.copy()

    current_marks = ["MARCA_HP", "MARCA_SEG_VIDA", "MARCA_DIF", "MARCA_CONV", "MARCA_GAR"]
    marks_3m = ["MARCA_DIF3", "MARCA_SEG_VIDA3", "MARCA_GAR3", "MARCA_CONV3", "MARCA_HIP3"]
    marks_6m = ["MARCA_DIF6", "MARCA_SEG_VIDA6", "MARCA_GAR6", "MARCA_HIP6", "MARCA_CONV6"]
    marks_12m = ["MARCA_DIF12", "MARCA_SEG_VIDA12", "MARCA_GAR12", "MARCA_HIP12", "MARCA_CONV12"]

    def _sum_present(row: pd.Series, cols: list[str]) -> int:
        return sum(row[c] for c in cols if c in row.index)

    df["N_MARKS_CURRENT"] = df.apply(_sum_present, cols=current_marks, axis=1).astype("int8")
    df["N_MARKS_3M"] = df.apply(_sum_present, cols=marks_3m, axis=1).astype("int8")
    df["N_MARKS_6M"] = df.apply(_sum_present, cols=marks_6m, axis=1).astype("int8")
    df["N_MARKS_12M"] = df.apply(_sum_present, cols=marks_12m, axis=1).astype("int8")

    return df


def add_utilisation_interactions(df: pd.DataFrame) -> pd.DataFrame:
    """Add interaction terms between card-utilisation and debt-ratio features.

    Captures joint effects such as high card utilisation combined with a
    high proportion of micro-enterprise debt, which can indicate financial
    stress.

    Args:
        df: Processed ``DataFrame``.

    Returns:
        ``DataFrame`` with interaction columns prefixed by ``INT_``.
    """
    df = df.copy()

    if "UTIL_TARJ" in df.columns and "DMES_DTOTAL" in df.columns:
        df["INT_UTIL_TARJ_X_DMES"] = df["UTIL_TARJ"] * df["DMES_DTOTAL"]

    if "UTIL_EFEC" in df.columns and "RATIO_PRES" in df.columns:
        df["INT_UTIL_EFEC_X_RATIO_PRES"] = df["UTIL_EFEC"] * df["RATIO_PRES"]

    if "MAX_ATRASO3" in df.columns and "MAX_CALIF3" in df.columns:
        df["INT_ATRASO3_X_CALIF3"] = (
            df["MAX_ATRASO3"].astype(float) * df["MAX_CALIF3"].astype(float)
        )

    return df


def add_debt_evolution(df: pd.DataFrame) -> pd.DataFrame:
    """Add debt-evolution indicators across temporal windows.

    Uses precomputed ``D12_MAXD*`` and ``PROM*_PROM*`` ratios to signal
    whether total debt is growing, stable or contracting relative to
    historical maximums and averages.

    Args:
        df: Processed ``DataFrame``.

    Returns:
        ``DataFrame`` with ``DEBT_GROWTH_*`` flag columns (``int8``).
    """
    df = df.copy()

    if "D12_MAXD" in df.columns:
        df["DEBT_ABOVE_12M_MAX"] = (df["D12_MAXD"] >= 1.0).astype("int8")

    if "PROM3_PROM12" in df.columns:
        df["DEBT_ACCEL_3_12"] = (df["PROM3_PROM12"] > 1.0).astype("int8")

    if "PROM6_PROM12" in df.columns:
        df["DEBT_ACCEL_6_12"] = (df["PROM6_PROM12"] > 1.0).astype("int8")

    return df


# ── Pipeline ──────────────────────────────────────────────────────────────────


def build_features(df: pd.DataFrame) -> pd.DataFrame:
    """Run the full feature-engineering pipeline.

    Applies in sequence:

    1. :func:`add_trend_ratios`
    2. :func:`add_volatility_scores`
    3. :func:`add_mark_aggregates`
    4. :func:`add_utilisation_interactions`
    5. :func:`add_debt_evolution`

    Args:
        df: Cleaned and encoded ``DataFrame`` from the dataset module.

    Returns:
        ``DataFrame`` enriched with all derived features.
    """
    steps = [
        ("Trend ratios", add_trend_ratios),
        ("Volatility scores", add_volatility_scores),
        ("Mark aggregates", add_mark_aggregates),
        ("Utilisation interactions", add_utilisation_interactions),
        ("Debt evolution", add_debt_evolution),
    ]
    for name, fn in tqdm(steps, desc="Feature engineering"):
        logger.debug(f"Applying: {name}")
        df = fn(df)

    n_new = df.shape[1]
    logger.info(f"Feature engineering complete – total columns: {n_new}")
    return df


# ── CLI entry-point ───────────────────────────────────────────────────────────