utils/function.py

from __future__ import annotations

import math
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any, Mapping


def resolve_schema_path(base_dir: Path) -> Path:
    candidates = [
        base_dir / "model" / "schema" / "runtime_schema_DE.json",
        base_dir / "model" / "resources-to-build" / "runtime_schema_DE.json",
    ]

    for candidate in candidates:
        if candidate.exists():
            return candidate

    searched = ", ".join(str(path.relative_to(base_dir)) for path in candidates)
    raise FileNotFoundError(f"Runtime schema not found. Checked: {searched}")


def to_float_or_none(value: Any) -> float | None:
    if value is None or value == "":
        return None
    try:
        number = float(value)
        return number if math.isfinite(number) else None
    except (TypeError, ValueError):
        return None


def to_categorical(value: Any) -> str:
    if value is None:
        return "MISSING"
    text = str(value).strip()
    return text if text else "MISSING"


def parse_date(value: Any) -> date | None:
    if value is None:
        return None
    if isinstance(value, date):
        return value

    for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%d-%m-%Y", "%d/%m/%Y"):
        try:
            return datetime.strptime(str(value).strip(), fmt).date()
        except ValueError:
            continue

    return None


def periodo_to_fecha_corte(periodo: int | str) -> date | None:
    value = str(periodo).strip()
    if len(value) != 6 or not value.isdigit():
        return None

    year, month = int(value[:4]), int(value[4:])
    if not 1 <= month <= 12:
        return None

    if month == 12:
        return date(year, 12, 31)

    return date(year, month + 1, 1) - timedelta(days=1)


def years_diff_floor(start: date | None, end: date | None) -> int | None:
    if start is None or end is None:
        return None

    years = end.year - start.year
    if (end.month, end.day) < (start.month, start.day):
        years -= 1
    return years


def months_diff_floor(start: date | None, end: date | None) -> int | None:
    if start is None or end is None:
        return None

    months = (end.year - start.year) * 12 + (end.month - start.month)
    if end.day < start.day:
        months -= 1
    return months


def safe_log1p(value: Any) -> float | None:
    number = to_float_or_none(value)
    if number is None:
        return None
    return math.log1p(max(number, 0.0))


def cap_dias_mora(dias_mora: Any) -> float | None:
    value = to_float_or_none(dias_mora)
    if value is None:
        return None
    return min(value, 120.0)


def bucket_mora(dias_mora: Any) -> str:
    value = to_float_or_none(dias_mora)
    if value is None:
        return "MISSING"
    if value <= 0:
        return "0"
    if value <= 8:
        return "1_8"
    if value <= 30:
        return "9_30"
    if value <= 60:
        return "31_60"
    if value <= 89:
        return "61_89"
    return "90_plus"


def horizonte_hasta_dic(fecha_snapshot: date | None) -> int | None:
    if fecha_snapshot is None:
        return None
    return 12 - fecha_snapshot.month


def safe_ratio(numerator: Any, denominator: Any) -> float | None:
    left = to_float_or_none(numerator)
    right = to_float_or_none(denominator)
    if left is None or right is None or right <= 0:
        return None
    return left / right


def build_feature_map(raw: Mapping[str, Any]) -> tuple[dict[str, Any], dict[str, Any]]:
    fecha_snapshot = parse_date(raw.get("fecha_snapshot")) or periodo_to_fecha_corte(raw["periodo"])
    fecha_nacimiento = parse_date(raw.get("fecha_nacimiento"))
    fecha_1er_desembolso = parse_date(raw.get("fecha_1er_desembolso"))
    fecha_vencimiento = parse_date(raw.get("fecha_vencimiento"))

    saldo_actual = to_float_or_none(raw.get("saldo_actual"))
    capital_desembolsado = to_float_or_none(raw.get("capital_desembolsado"))
    cuota_capital = to_float_or_none(raw.get("cuota_capital"))
    dias_mora = to_float_or_none(raw.get("dias_mora_capital"))
    dias_mora_capped = cap_dias_mora(dias_mora)

    derived = {
        "fecha_snapshot": str(fecha_snapshot) if fecha_snapshot else None,
        "horizonte_meses_hasta_dic": horizonte_hasta_dic(fecha_snapshot),
        "edad_cliente_t": years_diff_floor(fecha_nacimiento, fecha_snapshot),
        "edad_credito_meses_t": months_diff_floor(fecha_1er_desembolso, fecha_snapshot),
        "plazo_original_meses": months_diff_floor(fecha_1er_desembolso, fecha_vencimiento),
        "plazo_remanente_meses_t": months_diff_floor(fecha_snapshot, fecha_vencimiento),
        "dias_mora_capital_capped": dias_mora_capped,
        "mora_bucket": bucket_mora(dias_mora),
        "ratio_saldo_capital": safe_ratio(saldo_actual, capital_desembolsado),
        "ratio_cuota_saldo": safe_ratio(cuota_capital, saldo_actual),
        "ratio_cuota_capital_des": safe_ratio(cuota_capital, capital_desembolsado),
        "saldo_actual_log": safe_log1p(saldo_actual),
        "capital_desembolsado_log": safe_log1p(capital_desembolsado),
        "cuota_capital_log": safe_log1p(cuota_capital),
    }

    feature_map = {
        "id_cooperativa": to_categorical(raw.get("id_cooperativa")),
        "calificacion_credito": to_categorical(raw.get("calificacion_credito")),
        "saldo_actual": saldo_actual,
        "capital_desembolsado": capital_desembolsado,
        "cuota_capital": cuota_capital,
        "tasa_interes": to_float_or_none(raw.get("tasa_interes")),
        "tasa_pactada": to_float_or_none(raw.get("tasa_pactada")),
        "reestructurado": to_categorical(raw.get("reestructurado")),
        "status_credito": to_categorical(raw.get("status_credito")),
        "metodo_calculo": to_categorical(raw.get("metodo_calculo")),
        "tipo_garantia": to_categorical(raw.get("tipo_garantia")),
        "frecuencia_pago": to_categorical(raw.get("frecuencia_pago")),
        "id_agencia": to_categorical(raw.get("id_agencia")),
        "tipo_cliente": to_categorical(raw.get("tipo_cliente")),
        "categoria_cliente": to_categorical(raw.get("categoria_cliente")),
        "persona_pep": to_categorical(raw.get("persona_pep")),
        "persona_cpe": to_categorical(raw.get("persona_cpe")),
        **{key: value for key, value in derived.items() if key != "fecha_snapshot"},
    }

    return feature_map, derived


def get_categorical_feature_indices(schema: Mapping[str, Any]) -> list[int]:
    categorical = set(schema["categorical_features"])
    return [
        index
        for index, feature_name in enumerate(schema["feature_order"])
        if feature_name in categorical
    ]


def build_catboost_row(
    raw: Mapping[str, Any],
    schema: Mapping[str, Any],
) -> tuple[list[Any], dict[str, Any], dict[str, Any]]:
    feature_map, derived = build_feature_map(raw)
    categorical = set(schema["categorical_features"])
    row: list[Any] = []

    for feature_name in schema["feature_order"]:
        value = feature_map.get(feature_name)
        if feature_name in categorical:
            row.append(to_categorical(value))
        else:
            row.append(float("nan") if value is None else float(value))

    return row, feature_map, derived