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)


# ---------------------------------------------------------------------------
# Detección de tipos mal guardados y columnas redundantes
# ---------------------------------------------------------------------------

def numeric_stored_as_text(df: pd.DataFrame) -> list[str]:
    """Columnas de texto donde casi todo se puede convertir a número.

    Típico de CSVs con decimales con coma ("3,14") o números con unidades.
    """
    suspects = []
    for col in categorical_columns(df):
        series = df[col].dropna().astype(str)
        if series.empty or df[col].dtype == bool:
            continue
        cleaned = series.str.replace(",", ".", regex=False).str.strip()
        converted = pd.to_numeric(cleaned, errors="coerce")
        if converted.notna().mean() >= 0.8 and series.nunique() > 2:
            suspects.append(col)
    return suspects


def dates_stored_as_text(df: pd.DataFrame) -> list[str]:
    """Columnas de texto donde casi todo parsea como fecha."""
    suspects = []
    for col in categorical_columns(df):
        series = df[col].dropna().astype(str)
        if series.empty or df[col].dtype == bool:
            continue
        # Solo intentarlo si los valores "parecen" fechas (dígitos + separador):
        # to_datetime es caro y de gatillo fácil con números sueltos.
        looks_like = series.str.contains(r"\d{1,4}[-/.]\d{1,2}", regex=True, na=False)
        if looks_like.mean() < 0.8:
            continue
        parsed = pd.to_datetime(series, errors="coerce", format="mixed", dayfirst=True)
        if parsed.notna().mean() >= 0.8:
            suspects.append(col)
    return suspects


def duplicate_columns(df: pd.DataFrame) -> list[tuple[str, str]]:
    """Pares de columnas con contenido idéntico (misma información dos veces)."""
    pairs = []
    cols = df.columns.tolist()
    hashes = {col: pd.util.hash_pandas_object(df[col], index=False).sum() for col in cols}
    for i in range(len(cols)):
        for j in range(i + 1, len(cols)):
            if hashes[cols[i]] == hashes[cols[j]] and df[cols[i]].equals(df[cols[j]]):
                pairs.append((cols[i], cols[j]))
    return pairs


# ---------------------------------------------------------------------------
# Asociación entre categóricas (V de Cramér) y con el objetivo (eta cuadrado)
# ---------------------------------------------------------------------------

MAX_CRAMER_CARDINALITY = 30  # con más categorías, la tabla de contingencia no dice nada


def cramers_v(x: pd.Series, y: pd.Series) -> float:
    """V de Cramér entre dos series categóricas (0 = independientes, 1 = asociación total).

    Chi-cuadrado calculado a mano con numpy para no depender de scipy.
    """
    table = pd.crosstab(x, y).to_numpy().astype(float)
    n = table.sum()
    if n == 0 or min(table.shape) < 2:
        return float("nan")
    row_sums = table.sum(axis=1, keepdims=True)
    col_sums = table.sum(axis=0, keepdims=True)
    expected = row_sums @ col_sums / n
    with np.errstate(divide="ignore", invalid="ignore"):
        chi2 = np.nansum(np.where(expected > 0, (table - expected) ** 2 / expected, 0.0))
    return float(np.sqrt(chi2 / (n * (min(table.shape) - 1))))


def cramers_v_matrix(df: pd.DataFrame) -> pd.DataFrame:
    cols = [
        c for c in categorical_columns(df)
        if 2 <= df[c].nunique(dropna=True) <= MAX_CRAMER_CARDINALITY
    ]
    if len(cols) < 2:
        return pd.DataFrame()
    matrix = pd.DataFrame(np.eye(len(cols)), index=cols, columns=cols)
    for i in range(len(cols)):
        for j in range(i + 1, len(cols)):
            value = cramers_v(df[cols[i]].astype(str), df[cols[j]].astype(str))
            matrix.iloc[i, j] = matrix.iloc[j, i] = round(value, 3)
    return matrix


def eta_squared_table(df: pd.DataFrame, target: str) -> pd.DataFrame:
    """Eta² de cada variable numérica respecto a un objetivo categórico.

    Mide qué proporción de la varianza de la numérica explican las clases
    del objetivo (0 = nada, 1 = toda). Complementa a la correlación, que
    solo funciona con objetivos numéricos o binarios.
    """
    if target not in df.columns:
        return pd.DataFrame()
    rows = []
    groups_all = df[target].astype(str)
    for col in numeric_columns(df):
        if col == target:
            continue
        data = pd.DataFrame({"valor": df[col], "grupo": groups_all}).dropna()
        if len(data) < 10 or data["grupo"].nunique() < 2:
            continue
        grand_mean = data["valor"].mean()
        ss_total = ((data["valor"] - grand_mean) ** 2).sum()
        if ss_total == 0:
            continue
        ss_between = sum(
            len(g) * (g["valor"].mean() - grand_mean) ** 2
            for _, g in data.groupby("grupo")
        )
        rows.append({"variable": col, "eta_cuadrado": round(float(ss_between / ss_total), 3)})
    result = pd.DataFrame(rows)
    if not result.empty:
        result = result.sort_values("eta_cuadrado", ascending=False).reset_index(drop=True)
    return result


# ---------------------------------------------------------------------------
# PCA a mano con numpy (sin scikit-learn: la dependencia no se justifica
# para dos componentes)
# ---------------------------------------------------------------------------

PCA_SAMPLE = 5_000


def pca_summary(df: pd.DataFrame, color_col: str | None = None) -> dict | None:
    """PCA sobre las numéricas estandarizadas.

    Devuelve la varianza explicada por componente y la proyección 2D
    (para el gráfico), o None si no hay suficientes datos.
    """
    cols = numeric_columns(df)
    if len(cols) < 3:
        return None
    data = df[cols].dropna()
    if len(data) < 10:
        return None
    if len(data) > PCA_SAMPLE:
        data = data.sample(PCA_SAMPLE, random_state=42)

    X = data.to_numpy(dtype=float)
    X = X - X.mean(axis=0)
    std = X.std(axis=0)
    keep = std > 0
    X = X[:, keep] / std[keep]
    if X.shape[1] < 2:
        return None

    _, s, vt = np.linalg.svd(X, full_matrices=False)
    variance = s**2
    ratio = variance / variance.sum()

    projection = pd.DataFrame(X @ vt[:2].T, columns=["PC1", "PC2"], index=data.index)
    if color_col and color_col in df.columns:
        projection[color_col] = df.loc[data.index, color_col].astype(str)

    loadings = pd.DataFrame(
        vt[:2].T,
        index=[c for c, k in zip(cols, keep) if k],
        columns=["PC1", "PC2"],
    ).round(3)

    return {
        "varianza_explicada": [round(float(r), 4) for r in ratio],
        "proyeccion": projection,
        "cargas": loadings,
        "n_muestra": len(data),
    }