Preparar datos: limpieza, variables y pipelines

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns

titanic = sns.load_dataset("titanic")
print(titanic.shape)
print(titanic[["survived", "pclass", "sex", "age", "fare",
               "embarked", "deck"]].head())

from matplotlib.colors import LinearSegmentedColormap

cols = ["survived", "pclass", "sex", "age", "sibsp", "parch",
        "fare", "embarked", "deck"]
faltan = titanic[cols].isna()
print((faltan.mean() * 100).round(1))

blanco_tinta = LinearSegmentedColormap.from_list("bt", ["white", "C0"])
plt.imshow(faltan, aspect="auto", cmap=blanco_tinta,
           interpolation="nearest")
plt.xticks(range(len(cols)), cols, rotation=90)
plt.grid(False)                       # sin rejilla encima del mapa
plt.ylabel("Pasajero (fila)")
plt.title("Casillas vacías (oscuras)")
plt.show()

columnas = ["pclass", "sex", "age", "sibsp", "parch", "fare",
            "embarked"]
X = titanic[columnas]
y = titanic["survived"]
print(X.shape, "  sobrevivió:", round(y.mean() * 100, 1), "%")

from sklearn.model_selection import train_test_split

# stratify=y: la misma proporción de sobrevivientes en las dos partes
X_ent, X_prueba, y_ent, y_prueba = train_test_split(
    X, y, test_size=0.25, stratify=y, random_state=0)
print("Entrenamiento:", X_ent.shape, "  prueba:", X_prueba.shape)
print("Edades faltantes:", X_ent["age"].isna().sum(), "y",
      X_prueba["age"].isna().sum())

for col in ["age", "fare"]:
    print(f"{col}:  media = {X_ent[col].mean():5.2f}"
          f"   mediana = {X_ent[col].median():5.2f}"
          f"   máximo = {X_ent[col].max():6.2f}")

from sklearn.impute import SimpleImputer

imputar_edad = SimpleImputer(strategy="median")
imputar_edad.fit(X_ent[["age"]])      # aprende con el entrenamiento
print("Mediana aprendida:", imputar_edad.statistics_)

edad_ent = imputar_edad.transform(X_ent[["age"]])
edad_prueba = imputar_edad.transform(X_prueba[["age"]])
print("Faltantes después:", np.isnan(edad_ent).sum(), "y",
      np.isnan(edad_prueba).sum())

imputar_puerto = SimpleImputer(strategy="most_frequent")
imputar_puerto.fit(X_ent[["embarked"]])
print("Puerto más frecuente:", imputar_puerto.statistics_)

from sklearn.preprocessing import OneHotEncoder

puertos = pd.DataFrame({"embarked": ["S", "C", "Q", "S"]})
onehot = OneHotEncoder(sparse_output=False, handle_unknown="ignore")
onehot.fit(puertos)
print(onehot.get_feature_names_out())
print(onehot.transform(puertos))
# Un puerto que no apareció en el entrenamiento queda todo en cero
nuevo = pd.DataFrame({"embarked": ["X"]})
print(onehot.transform(nuevo))

from sklearn.preprocessing import StandardScaler, MinMaxScaler

# Primero rellenamos la edad; la tarifa no tiene faltantes
edad_tarifa = pd.DataFrame(
    {"edad": edad_ent[:, 0], "tarifa": X_ent["fare"].to_numpy()})

estandarizar = StandardScaler().fit(edad_tarifa)
print("Medias:", estandarizar.mean_.round(2))
print("Desviaciones:", estandarizar.scale_.round(2))
est = estandarizar.transform(edad_tarifa)

normalizar = MinMaxScaler().fit(edad_tarifa)
norm = normalizar.transform(edad_tarifa)
print("Estandarizadas: de", est.min(axis=0).round(2),
      "a", est.max(axis=0).round(2))
print("Normalizadas:   de", norm.min(axis=0).round(2),
      "a", norm.max(axis=0).round(2))

fig, ejes = plt.subplots(2, 2, figsize=(4.8, 4.8))
ejes[0, 0].hist(edad_tarifa["edad"], bins=30)
ejes[0, 0].set_title("Edad (años)")
ejes[0, 1].hist(edad_tarifa["tarifa"], bins=30)
ejes[0, 1].set_title("Tarifa")
ejes[1, 0].hist(est[:, 0], bins=30, color="C2")
ejes[1, 0].set_title("Edad (puntaje z)")
ejes[1, 1].hist(est[:, 1], bins=30, color="C2")
ejes[1, 1].set_title("Tarifa (z)")
plt.show()

# Con fuga: las cuentas usan también a los pasajeros de prueba
print(f"Mediana de la edad:  todos = {X['age'].median()}"
      f"   solo entrenamiento = {X_ent['age'].median()}")
print(f"Media de la tarifa:  todos = {X['fare'].mean():.2f}"
      f"   solo entrenamiento = {X_ent['fare'].mean():.2f}")

from sklearn.feature_selection import SelectKBest
from sklearn.linear_model import LogisticRegression

rng = np.random.default_rng(0)
X_azar = rng.normal(size=(200, 5000))     # 5000 columnas de ruido
y_azar = rng.integers(0, 2, size=200)     # respuestas al azar
X_nuevo = rng.normal(size=(1000, 5000))   # datos que llegan después
y_nuevo = rng.integers(0, 2, size=1000)

Xa_ent, Xa_prueba, ya_ent, ya_prueba = train_test_split(
    X_azar, y_azar, test_size=0.25, random_state=0)

resultados = {}
for nombre, Xs, ys in [("Con fuga", X_azar, y_azar),
                       ("Sin fuga", Xa_ent, ya_ent)]:
    # La única diferencia: con qué filas se eligen las 20 columnas
    elegir = SelectKBest(k=20).fit(Xs, ys)
    modelo = LogisticRegression().fit(elegir.transform(Xa_ent), ya_ent)
    resultados[nombre] = [
        modelo.score(elegir.transform(Xa_prueba), ya_prueba),
        modelo.score(elegir.transform(X_nuevo), y_nuevo)]
    print(f"{nombre}:  prueba = {resultados[nombre][0]:.2f}"
          f"   datos nuevos = {resultados[nombre][1]:.3f}")

alturas = resultados["Con fuga"] + resultados["Sin fuga"]
nombres = ["Prueba\ncon fuga", "Nuevos\ncon fuga",
           "Prueba\nsin fuga", "Nuevos\nsin fuga"]
# Coral solo para el número engañoso
barras = plt.bar(nombres, alturas, color=["C1", "C2", "C0", "C2"])
# padding separa los números de la línea punteada del 0,5
plt.bar_label(barras, fmt="%.2f", padding=8)
plt.axhline(0.5, color="C2", linestyle="--", linewidth=1)
plt.ylim(0, 1)
plt.ylabel("Exactitud")
plt.title("La fuga engaña en la prueba")
plt.show()

from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.neighbors import KNeighborsClassifier

numericas = ["pclass", "age", "sibsp", "parch", "fare"]
categoricas = ["sex", "embarked"]

prep_numericas = Pipeline([
    ("imputar", SimpleImputer(strategy="median")),
    ("escalar", StandardScaler())])
prep_categoricas = Pipeline([
    ("imputar", SimpleImputer(strategy="most_frequent")),
    ("onehot", OneHotEncoder(handle_unknown="ignore"))])
preparar = ColumnTransformer([
    ("num", prep_numericas, numericas),
    ("cat", prep_categoricas, categoricas)])

pipe_knn = Pipeline([("preparar", preparar),
                     ("modelo", KNeighborsClassifier(n_neighbors=5))])
pipe_knn.fit(X_ent, y_ent)

print(pipe_knn.named_steps["preparar"].get_feature_names_out())
print("Forma lista para el modelo:",
      pipe_knn.named_steps["preparar"].transform(X_ent).shape)

from sklearn.base import clone
from sklearn.model_selection import StratifiedKFold, cross_val_score

pliegues = StratifiedKFold(n_splits=5, shuffle=True, random_state=0)

for escala in [StandardScaler(), "passthrough"]:
    pipe = clone(pipe_knn).set_params(preparar__num__escalar=escala)
    puntajes = cross_val_score(pipe, X_ent, y_ent, cv=pliegues)
    print(f"Escalar = {str(escala):16}  exactitud = "
          f"{puntajes.mean():.3f} ± {puntajes.std():.3f}")

# Con fuga: se prepara una vez con todas las filas del entrenamiento
X_preparado = clone(preparar).fit_transform(X_ent)
con_fuga = cross_val_score(KNeighborsClassifier(n_neighbors=5),
                           X_preparado, y_ent, cv=pliegues)
sin_fuga = cross_val_score(pipe_knn, X_ent, y_ent, cv=pliegues)
print(f"Con fuga: {con_fuga.mean():.3f}"
      f"   sin fuga: {sin_fuga.mean():.3f}")

def agregar_variables(df):
    """Agrega variables nuevas calculadas fila por fila."""
    df = df.copy()
    df["familia"] = df["sibsp"] + df["parch"] + 1
    df["tarifa_persona"] = df["fare"] / df["familia"]
    df["es_nino"] = (df["age"] < 12).astype(int)
    return df

X_ent_mas = agregar_variables(X_ent)
print(X_ent_mas[["sibsp", "parch", "familia", "fare",
                 "tarifa_persona", "age", "es_nino"]].head(4))

def pipe_logistica(numericas):
    """Pipeline de regresión logística para unas columnas numéricas."""
    prep = ColumnTransformer([
        ("num", prep_numericas, numericas),
        ("cat", prep_categoricas, categoricas)])
    return Pipeline([("preparar", prep),
                     ("modelo", LogisticRegression(max_iter=1000))])

pruebas = {"originales": [],
           "+ familia y tarifa_persona": ["familia", "tarifa_persona"],
           "+ es_nino": ["es_nino"]}
for nombre, extra in pruebas.items():
    puntajes = cross_val_score(pipe_logistica(numericas + extra),
                               X_ent_mas, y_ent, cv=pliegues)
    print(f"{nombre:27} {puntajes.mean():.3f} ± {puntajes.std():.3f}")

from sklearn.model_selection import GridSearchCV

rejilla = {
    "modelo__n_neighbors": [1, 3, 5, 9, 15, 25, 41],
    "preparar__num__escalar": [StandardScaler(), MinMaxScaler(),
                               "passthrough"]}
busqueda = GridSearchCV(pipe_knn, rejilla, cv=pliegues)
busqueda.fit(X_ent, y_ent)
for nombre, valor in busqueda.best_params_.items():
    print(f"Mejor {nombre}: {valor}")
print(f"Exactitud en validación: {busqueda.best_score_:.3f}")

# Los 21 puntajes, como tabla: una fila por escalador
tabla = busqueda.cv_results_["mean_test_score"].reshape(7, 3).T
escaladores = ["Estandarizar", "Normalizar", "Sin escalar"]
ks = rejilla["modelo__n_neighbors"]
fig, ax = plt.subplots(figsize=(4.8, 2.6))
sns.heatmap(tabla, annot=True, fmt=".2f", cmap=blanco_tinta,
            cbar=False, xticklabels=ks, yticklabels=escaladores,
            annot_kws={"fontsize": 11}, ax=ax)
ax.tick_params(axis="y", labelrotation=0)
ax.set_xlabel("Número de vecinos k")
ax.set_title("Exactitud en validación")
plt.show()

from scipy.stats import loguniform
from sklearn.model_selection import RandomizedSearchCV

pipe_log = pipe_logistica(numericas + ["es_nino"])
distribuciones = {"modelo__C": loguniform(1e-3, 1e2)}
azar = RandomizedSearchCV(pipe_log, distribuciones, n_iter=20,
                          cv=pliegues, random_state=0)
azar.fit(X_ent_mas, y_ent)
print(f"Mejor C: {azar.best_params_['modelo__C']:.3f}")
print(f"Exactitud en validación: {azar.best_score_:.3f}")

# El pipeline del tema 10: el Titanic, de la tabla cruda a la predicción
import numpy as np
import pandas as pd
import seaborn as sns
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import classification_report
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

NUMERICAS = ["pclass", "age", "sibsp", "parch", "fare",
             "familia", "tarifa_persona", "es_nino"]
CATEGORICAS = ["sex", "embarked"]

def agregar_variables(df):
    """Agrega variables nuevas calculadas fila por fila."""
    df = df.copy()
    df["familia"] = df["sibsp"] + df["parch"] + 1
    df["tarifa_persona"] = df["fare"] / df["familia"]
    df["es_nino"] = (df["age"] < 12).astype(int)
    return df

def crear_pipeline_titanic(modelo):
    """Prepara las columnas del Titanic y termina en el modelo dado."""
    numericas = Pipeline([
        ("imputar", SimpleImputer(strategy="median")),
        ("escalar", StandardScaler())])
    categoricas = Pipeline([
        ("imputar", SimpleImputer(strategy="most_frequent")),
        ("onehot", OneHotEncoder(handle_unknown="ignore"))])
    preparar = ColumnTransformer([
        ("num", numericas, NUMERICAS),
        ("cat", categoricas, CATEGORICAS)])
    return Pipeline([("preparar", preparar), ("modelo", modelo)])

titanic = agregar_variables(sns.load_dataset("titanic"))
X = titanic[NUMERICAS + CATEGORICAS]
y = titanic["survived"]
X_ent, X_prueba, y_ent, y_prueba = train_test_split(
    X, y, test_size=0.25, stratify=y, random_state=0)

pipeline_titanic = crear_pipeline_titanic(
    LogisticRegression(C=0.44, max_iter=1000))
pipeline_titanic.fit(X_ent, y_ent)

# La prueba se usa una sola vez, al final
pred = pipeline_titanic.predict(X_prueba)
exactitud = pipeline_titanic.score(X_prueba, y_prueba)
print(f"Exactitud en prueba: {exactitud:.3f}")
print(classification_report(y_prueba, pred,
                            target_names=["murió", "sobrevivió"]))

nuevo = agregar_variables(pd.DataFrame([{
    "pclass": 3, "sex": "female", "age": np.nan, "sibsp": 0,
    "parch": 2, "fare": 15.0, "embarked": "S"}]))
prob = pipeline_titanic.predict_proba(nuevo[NUMERICAS + CATEGORICAS])
print(f"Probabilidad de sobrevivir: {prob[0, 1]:.2f}")