eda/alerts.py

"""Motor de alertas de calidad de datos.

Cada regla examina el DataFrame y emite alertas con tres niveles:
  - critico: probablemente rompe cualquier análisis o modelo
  - aviso:   requiere una decisión consciente antes de seguir
  - info:    conviene saberlo, pero no bloquea

Cada alerta lleva una recomendación concreta, no solo el diagnóstico:
decir "columna X tiene 80% de faltantes" sin decir qué hacer con ella
es dejar el trabajo a medias.
"""

from __future__ import annotations

import pandas as pd

from .profiler import (
    categorical_columns,
    iqr_outlier_bounds,
    numeric_columns,
)

MISSING_WARNING = 20.0   # % de faltantes que merece aviso
MISSING_CRITICAL = 60.0  # % de faltantes probablemente irrecuperable
HIGH_CARDINALITY = 0.5   # ratio únicos/filas sospechoso en categóricas
OUTLIER_WARNING = 5.0    # % de outliers IQR que merece revisión
SKEW_WARNING = 2.0       # |asimetría| a partir de la cual avisar
CORRELATION_WARNING = 0.95  # |r| que sugiere variables redundantes
IMBALANCE_WARNING = 10.0    # % de la clase minoritaria


def _alert(nivel: str, columna: str, mensaje: str, recomendacion: str) -> dict:
    return {
        "nivel": nivel,
        "columna": columna,
        "mensaje": mensaje,
        "recomendacion": recomendacion,
    }


def check_missing(df: pd.DataFrame) -> list[dict]:
    alerts = []
    for col in df.columns:
        pct = 100 * df[col].isna().mean()
        if pct >= MISSING_CRITICAL:
            alerts.append(_alert(
                "critico", col,
                f"{pct:.0f}% de valores faltantes",
                "Con más del 60% ausente, imputar suele inventar datos: valora descartar la columna o tratarla como indicador binario de presencia.",
            ))
        elif pct >= MISSING_WARNING:
            alerts.append(_alert(
                "aviso", col,
                f"{pct:.0f}% de valores faltantes",
                "Decide una estrategia explícita: imputación (mediana/moda), categoría 'desconocido', o descarte justificado.",
            ))
    return alerts


def check_constant(df: pd.DataFrame) -> list[dict]:
    alerts = []
    for col in df.columns:
        if df[col].nunique(dropna=True) <= 1:
            alerts.append(_alert(
                "critico", col,
                "Columna constante (un único valor)",
                "No aporta información: elimínala antes de modelar.",
            ))
    return alerts


def check_id_like(df: pd.DataFrame) -> list[dict]:
    alerts = []
    for col in df.columns:
        n = len(df)
        if n == 0:
            continue
        if df[col].nunique() == n and str(df[col].dtype) != "float64":
            alerts.append(_alert(
                "aviso", col,
                "Todos los valores son únicos — parece un identificador",
                "Los IDs no deben entrar a un modelo (memorizarlos es sobreajuste garantizado). Úsala solo como índice.",
            ))
    return alerts


def check_high_cardinality(df: pd.DataFrame) -> list[dict]:
    alerts = []
    n = len(df)
    for col in categorical_columns(df):
        uniques = df[col].nunique()
        if n > 0 and uniques > 50 and uniques / n >= HIGH_CARDINALITY:
            alerts.append(_alert(
                "aviso", col,
                f"Cardinalidad muy alta ({uniques} valores distintos)",
                "One-hot dispararía la dimensionalidad: considera agrupar categorías raras, hashing o target encoding.",
            ))
    return alerts


def check_duplicates(df: pd.DataFrame) -> list[dict]:
    dup = int(df.duplicated().sum())
    if dup == 0:
        return []
    pct = 100 * dup / len(df)
    nivel = "aviso" if pct >= 5 else "info"
    return [_alert(
        nivel, "(dataset)",
        f"{dup} filas duplicadas ({pct:.1f}%)",
        "Comprueba si son medidas repetidas legítimas o errores de captura; si son errores, elimínalas antes de todo lo demás.",
    )]


def check_outliers(df: pd.DataFrame) -> list[dict]:
    alerts = []
    for col in numeric_columns(df):
        series = df[col].dropna()
        if len(series) < 10:
            continue
        low, high = iqr_outlier_bounds(series)
        pct = 100 * ((series < low) | (series > high)).mean()
        if pct >= OUTLIER_WARNING:
            alerts.append(_alert(
                "aviso", col,
                f"{pct:.1f}% de outliers según IQR",
                "Investiga si son errores de medida o valores reales extremos: la respuesta cambia el tratamiento (corregir vs. transformar/robustecer).",
            ))
    return alerts


def check_skewness(df: pd.DataFrame) -> list[dict]:
    alerts = []
    for col in numeric_columns(df):
        series = df[col].dropna()
        if len(series) < 10:
            continue
        skew = float(series.skew())
        if abs(skew) >= SKEW_WARNING:
            alerts.append(_alert(
                "info", col,
                f"Distribución muy asimétrica (asimetría {skew:.1f})",
                "Si vas a usar modelos lineales, una transformación log o Box-Cox puede ayudar; a los de árboles les da igual.",
            ))
    return alerts


def check_redundant_pairs(df: pd.DataFrame) -> list[dict]:
    numeric = df[numeric_columns(df)]
    if numeric.shape[1] < 2:
        return []
    corr = numeric.corr()
    alerts = []
    cols = corr.columns
    for i in range(len(cols)):
        for j in range(i + 1, len(cols)):
            value = corr.iloc[i, j]
            if pd.notna(value) and abs(value) >= CORRELATION_WARNING:
                alerts.append(_alert(
                    "aviso", f"{cols[i]} ~ {cols[j]}",
                    f"Correlación {value:.2f}: prácticamente redundantes",
                    "Mantén una de las dos o combínalas; la multicolinealidad desestabiliza los modelos lineales y estorba en la interpretación.",
                ))
    return alerts


def check_imbalance(df: pd.DataFrame, target: str | None) -> list[dict]:
    if not target or target not in df.columns:
        return []
    counts = df[target].value_counts(dropna=True)
    if not (2 <= len(counts) <= 20):
        return []
    minority_pct = 100 * counts.iloc[-1] / counts.sum()
    if minority_pct < IMBALANCE_WARNING:
        return [_alert(
            "aviso", target,
            f"Clases desbalanceadas: la minoritaria ('{counts.index[-1]}') es solo el {minority_pct:.1f}%",
            "La accuracy engañará: usa F1/AUC-PR, estratifica los splits y valora sobremuestreo o pesos de clase.",
        )]
    return []


def run_all_checks(df: pd.DataFrame, target: str | None = None) -> list[dict]:
    """Ejecuta todas las reglas y devuelve las alertas ordenadas por gravedad."""
    alerts = (
        check_constant(df)
        + check_missing(df)
        + check_id_like(df)
        + check_high_cardinality(df)
        + check_duplicates(df)
        + check_outliers(df)
        + check_skewness(df)
        + check_redundant_pairs(df)
        + check_imbalance(df, target)
    )
    order = {"critico": 0, "aviso": 1, "info": 2}
    alerts.sort(key=lambda a: order[a["nivel"]])
    return alerts