eda/profiler.py

"""Cálculo del perfil exploratorio de un DataFrame.

Este módulo no sabe nada de Streamlit: recibe DataFrames y devuelve
estructuras simples (dicts y DataFrames). Eso permite probarlo sin interfaz
y reutilizarlo desde un notebook o un script.
"""

from __future__ import annotations

import io

import numpy as np
import pandas as pd

# Por encima de este número de filas, los gráficos usan una muestra para
# mantener la interfaz fluida (las estadísticas se calculan siempre con todo).
SAMPLE_THRESHOLD = 50_000


# ---------------------------------------------------------------------------
# Carga
# ---------------------------------------------------------------------------

def load_dataframe(data: bytes, filename: str) -> pd.DataFrame:
    """Carga CSV (detectando el separador) o Excel a partir de bytes."""
    name = filename.lower()
    if name.endswith((".xlsx", ".xls")):
        return pd.read_excel(io.BytesIO(data))
    # sep=None + engine="python" detecta ; , \t automáticamente — habitual
    # en CSVs europeos exportados de Excel.
    return pd.read_csv(io.BytesIO(data), sep=None, engine="python")


def numeric_columns(df: pd.DataFrame) -> list[str]:
    return df.select_dtypes(include="number").columns.tolist()


def categorical_columns(df: pd.DataFrame) -> list[str]:
    """Todo lo que no es numérico ni fecha se trata como categórico.

    Detectar por exclusión es deliberado: pandas 3 cambió el dtype de texto
    de 'object' a 'str' (backend Arrow) y una lista blanca de dtypes se
    quedaría corta según la versión instalada.
    """
    numeric = set(numeric_columns(df))
    return [
        col
        for col in df.columns
        if col not in numeric and not pd.api.types.is_datetime64_any_dtype(df[col])
    ]


# ---------------------------------------------------------------------------
# Visión general
# ---------------------------------------------------------------------------

def overview(df: pd.DataFrame) -> dict:
    return {
        "filas": int(len(df)),
        "columnas": int(df.shape[1]),
        "celdas": int(df.size),
        "memoria_mb": round(df.memory_usage(deep=True).sum() / 1024**2, 2),
        "numericas": len(numeric_columns(df)),
        "categoricas": len(categorical_columns(df)),
        "otras": df.shape[1] - len(numeric_columns(df)) - len(categorical_columns(df)),
        "faltantes_total": int(df.isna().sum().sum()),
        "faltantes_pct": round(100 * df.isna().sum().sum() / max(df.size, 1), 2),
        "filas_duplicadas": int(df.duplicated().sum()),
        "duplicadas_pct": round(100 * df.duplicated().sum() / max(len(df), 1), 2),
    }


def column_table(df: pd.DataFrame) -> pd.DataFrame:
    """Tabla resumen por columna: tipo, faltantes, únicos y valores de ejemplo."""
    rows = []
    for col in df.columns:
        series = df[col]
        sample = series.dropna().unique()[:3]
        rows.append(
            {
                "columna": col,
                "tipo": str(series.dtype),
                "no_nulos": int(series.notna().sum()),
                "faltantes_%": round(100 * series.isna().mean(), 1),
                "unicos": int(series.nunique()),
                "ejemplos": ", ".join(str(v)[:25] for v in sample),
            }
        )
    return pd.DataFrame(rows)


# ---------------------------------------------------------------------------
# Faltantes
# ---------------------------------------------------------------------------

def missing_table(df: pd.DataFrame) -> pd.DataFrame:
    missing = df.isna().sum()
    missing = missing[missing > 0].sort_values(ascending=False)
    return pd.DataFrame(
        {
            "columna": missing.index,
            "faltantes": missing.values,
            "porcentaje": (100 * missing / len(df)).round(1).values,
        }
    )


# ---------------------------------------------------------------------------
# Numéricas
# ---------------------------------------------------------------------------

def iqr_outlier_bounds(series: pd.Series) -> tuple[float, float]:
    q1, q3 = series.quantile(0.25), series.quantile(0.75)
    iqr = q3 - q1
    return q1 - 1.5 * iqr, q3 + 1.5 * iqr


def numeric_table(df: pd.DataFrame) -> pd.DataFrame:
    """describe() ampliado con asimetría, curtosis y outliers por IQR."""
    rows = []
    for col in numeric_columns(df):
        series = df[col].dropna()
        if series.empty:
            continue
        low, high = iqr_outlier_bounds(series)
        n_outliers = int(((series < low) | (series > high)).sum())
        rows.append(
            {
                "columna": col,
                "media": round(float(series.mean()), 3),
                "mediana": round(float(series.median()), 3),
                "desv_tipica": round(float(series.std()), 3) if len(series) > 1 else 0.0,
                "minimo": round(float(series.min()), 3),
                "maximo": round(float(series.max()), 3),
                "asimetria": round(float(series.skew()), 2) if len(series) > 2 else 0.0,
                "curtosis": round(float(series.kurtosis()), 2) if len(series) > 3 else 0.0,
                "outliers_iqr": n_outliers,
                "outliers_%": round(100 * n_outliers / len(series), 1),
            }
        )
    return pd.DataFrame(rows)


# ---------------------------------------------------------------------------
# Categóricas
# ---------------------------------------------------------------------------

def categorical_table(df: pd.DataFrame) -> pd.DataFrame:
    rows = []
    for col in categorical_columns(df):
        series = df[col].dropna().astype(str)
        if series.empty:
            continue
        counts = series.value_counts()
        rows.append(
            {
                "columna": col,
                "unicos": int(series.nunique()),
                "moda": str(counts.index[0])[:40],
                "frecuencia_moda_%": round(100 * counts.iloc[0] / len(series), 1),
                "ratio_cardinalidad": round(series.nunique() / len(series), 3),
            }
        )
    return pd.DataFrame(rows)


def value_counts_for(df: pd.DataFrame, col: str, top: int = 15) -> pd.DataFrame:
    counts = df[col].astype(str).value_counts(dropna=False).head(top)
    return pd.DataFrame({"valor": counts.index, "frecuencia": counts.values})


# ---------------------------------------------------------------------------
# Correlaciones
# ---------------------------------------------------------------------------

def correlation_matrix(df: pd.DataFrame, method: str = "pearson") -> pd.DataFrame:
    numeric = df[numeric_columns(df)]
    if numeric.shape[1] < 2:
        return pd.DataFrame()
    return numeric.corr(method=method).round(3)


def top_correlations(corr: pd.DataFrame, n: int = 10) -> pd.DataFrame:
    """Pares de variables más correlacionados (en valor absoluto)."""
    if corr.empty:
        return pd.DataFrame()
    pairs = []
    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):
                pairs.append({"variable_1": cols[i], "variable_2": cols[j], "correlacion": value})
    pairs.sort(key=lambda p: abs(p["correlacion"]), reverse=True)
    return pd.DataFrame(pairs[:n])


# ---------------------------------------------------------------------------
# Análisis respecto a una columna objetivo
# ---------------------------------------------------------------------------

def target_kind(df: pd.DataFrame, target: str) -> str:
    """'categorico' si el objetivo tiene pocas clases; 'numerico' si es continuo."""
    if target in categorical_columns(df) or df[target].nunique() <= 10:
        return "categorico"
    return "numerico"


def class_balance(df: pd.DataFrame, target: str) -> pd.DataFrame:
    counts = df[target].astype(str).value_counts(dropna=False)
    return pd.DataFrame(
        {
            "clase": counts.index,
            "n": counts.values,
            "porcentaje": (100 * counts / counts.sum()).round(1).values,
        }
    )


def correlations_with_target(df: pd.DataFrame, target: str) -> pd.DataFrame:
    """Correlación de cada numérica con el objetivo (si el objetivo es numérico
    o binario codificable)."""
    numeric = df[numeric_columns(df)]
    target_series = df[target]
    if not pd.api.types.is_numeric_dtype(target_series):
        # Objetivo no numérico: solo es codificable si es binario.
        classes = target_series.dropna().unique()
        if len(classes) != 2:
            return pd.DataFrame()
        target_series = (target_series == classes[0]).astype(float)
    if target in numeric.columns:
        numeric = numeric.drop(columns=[target])
    if numeric.empty:
        return pd.DataFrame()
    corr = numeric.corrwith(target_series).dropna().sort_values(key=abs, ascending=False)
    return pd.DataFrame({"variable": corr.index, "correlacion_con_objetivo": corr.round(3).values})


# ---------------------------------------------------------------------------
# Resumen compacto para el LLM (solo agregados, nunca filas de datos)
# ---------------------------------------------------------------------------

def compact_summary(df: pd.DataFrame, alerts: list[dict] | None = None) -> dict:
    summary = {
        "vision_general": overview(df),
        "columnas": column_table(df).to_dict(orient="records"),
        "numericas": numeric_table(df).to_dict(orient="records"),
        "categoricas": categorical_table(df).to_dict(orient="records"),
        "top_correlaciones": top_correlations(correlation_matrix(df), 8).to_dict(orient="records"),
    }
    if alerts:
        summary["alertas"] = [
            {"nivel": a["nivel"], "columna": a["columna"], "mensaje": a["mensaje"]}
            for a in alerts
        ]
    return summary


def sample_for_plots(df: pd.DataFrame) -> pd.DataFrame:
    """Muestra aleatoria reproducible para no ahogar al navegador con gráficos."""
    if len(df) <= SAMPLE_THRESHOLD:
        return df
    return df.sample(SAMPLE_THRESHOLD, random_state=42)