Serie MLOps y Producción · de la ingeniería de datos al monitoreo.
Módulo II — Feature engineering y pipelines
En la sesión 4 se diseñó \(T\): logaritmo, encoding, escalamiento, selección. El F1 subía o bajaba según cómo se representaba \(x\). Aquí el objeto de estudio cambia. Ya no preguntamos si una transformación es “buena”. Preguntamos si el flujo completo —preprocesamiento más modelo— es un contrato de ejecución.
Un contrato, en ingeniería, es una promesa verificable: las mismas entradas producen el mismo procedimiento, y el procedimiento se puede apagar, guardar y reabrir. En machine learning esa promesa tiene tres cláusulas. Se entrena con una frontera de información explícita (el test no ayuda a aprender), se aplica la misma ruta en inferencia, y el resultado vive en un artefacto que alguien más puede recargar. Si alguna falta, lo que queda es un notebook que “dio buen F1 el martes”.
Este artículo enseña esas tres cláusulas en orden. Primero el concepto, después un ejemplo que se puede leer a ojo. El código no sustituye la teoría: la pone a prueba.
Objetivos de aprendizaje
Al finalizar esta sesión, el estudiante estará en capacidad de:
Explicar por qué el preprocesamiento en celdas sueltas rompe la validez de la evaluación (data leakage) y la consistencia en producción.
Integrar transformaciones y modelo en un Pipeline de scikit-learn, con ColumnTransformer para ramas heterogéneas.
Implementar un transformador propio (fit/transform) y declarar el manejo de niveles no vistos.
Fijar una partición train/test reproducible (semilla y estratificación) y justificar por qué sin semilla el experimento no es auditable.
Serializar el pipeline completo con joblib, recargarlo y reproducir la misma predicción sobre un registro nuevo.
Validar el esquema de entrada y registrar lo mínimo de una corrida en JSON.
Qué no cubre esta sesión
Cómo inventar o justificar features (sesión 4).
Validación cruzada, GridSearchCV y selección automática de modelos (sesión 6).
Un model registry con varias corridas (sesión 8).
El servicio HTTP que carga el artefacto (sesión 9).
Ruta de trabajo
El pipeline como contrato y la API fit / transform / predict.
El fallo del preprocesamiento suelto (fuga de scaler, de encoder y de selector).
Integración con ColumnTransformer y transformadores propios.
Partición reproducible y estratificación.
Serialización, esquema y contraejemplo de persistir solo el clasificador.
En una frase: un pipeline de machine learning es el objeto que obliga a que el dato crudo recorra los mismos pasos, en el mismo orden y con los mismos parámetros aprendidos, tanto cuando se entrena como cuando se predice.
Eso suena operacional, pero el concepto de fondo es estadístico. Un modelo no ve el mundo: ve una representación. Si esa representación se construye de una forma en el laboratorio y de otra en producción, el algoritmo está resolviendo un problema distinto al que se evaluó. El pipeline existe para que esa representación no quede suelta en celdas, funciones olvidadas o “pasos que yo me acuerdo”.
De la receta informal al objeto ejecutable
En un notebook es habitual escribir, en voz alta:
leo los datos,
relleno nulos,
escalo,
codifico categorías,
entreno,
miro el F1.
Esa lista es una receta. Una receta vive en la cabeza de quien la escribió. Un pipeline es la receta convertida en un grafo ejecutable: cada paso tiene nombre, cada paso sabe hacer fit (aprender parámetros) y transform (aplicarlos), y el equipo completo responde a predict. La diferencia no es estética. La receta se puede aplicar mal la segunda vez; el objeto, si está bien construido, no tiene segunda versión informal.
Piense en una cadena de montaje. No basta con saber que “primero se pinta y después se ensambla”. Hay que fijar con qué pintura, en qué estación y con la misma plantilla para cada unidad. Si una unidad se pinta a mano al final, el control de calidad mide otra cosa.
Tres obligaciones del contrato
Un pipeline de ML no es “usar la clase Pipeline”. Es cumplir tres obligaciones a la vez:
El test ayuda a entrenar: la métrica queda optimista
Misma ruta de inferencia
¿Un registro nuevo recorre los mismos pasos?
El orden y los parámetros ya ajustados
En producción el modelo recibe otra distribución
Artefacto persistente
¿Puedo apagar el notebook y repetir la predicción?
El flujo completo en disco
Falta el scaler o el encoder: las predicciones no coinciden
Las tres son necesarias. Un código que escala “bien” en el notebook pero no se puede recargar no es reproducible. Un archivo que se recarga pero aprendió el scaler con el test no es válido. Un flujo válido que en producción se reimplementa a mano en FastAPI suele divergir en una semana.
La ecuación que hay que conservar
Si \(T\) es el preprocesamiento aprendido y \(f\) el modelo,
$[ $hat{y} = f(T(x)), $]
el contrato exige que \(T\) y \(f\)viajen juntos. \(T\) no es un script de limpieza: es una función con estado. StandardScaler guarda \(\mu\) y \(\sigma\). OneHotEncoder guarda el vocabulario de categorías. SelectKBest guarda qué columnas sobrevivieron. Ese estado se obtiene con fit y solo entonces transform tiene sentido.
Guardar únicamente \(f\) deja \(T\) implícito en la RAM del notebook. Al reiniciar el kernel, \(T\) desaparece. El clasificador sigue existiendo, pero ya no sabe cómo se veía \(x\) cuando lo entrenaron.
fit, transform y predict: tres verbos, no tres scripts
En scikit-learn la distinción es conceptual, no cosmética:
fit: mira datos y escribe parámetros internos. Es el único verbo que aprende. Debe ejecutarse sobre train.
transform:lee esos parámetros y convierte \(x\) en \(T(x)\). No debería inventar una media nueva.
predict: aplica \(f\) sobre \(T(x)\).
fit_transform sobre el dataset completo es la forma más frecuente de romper la frontera: en una sola llamada se aprende y se aplica usando filas que luego se usarán para “evaluar”. El pipeline existe, entre otras cosas, para que fit ocurra una vez, en el lugar correcto, y transform se reutilice.
clone copia la especificación (pasos e hiperparámetros) sin copiar el estado aprendido. Sirve para repetir un experimento sin heredar un fit previo. Si dos corridas “salen distintas” y no hubo clone ni semilla, no se sabe si falló el modelo o la higiene del objeto.
Relación con las sesiones vecinas
Sesión 4 responde qué es \(T\): logaritmo, one-hot, scaler, selección. Aquí \(T\) se da por justificada y se encapsula.
Sesión 6 compara varios\(f\) con un protocolo de validación. Aquí usamos un modelo, sin búsqueda de hiperparámetros. Si se busca el mejor modelo antes de tener un contrato, se optimiza un procedimiento inválido.
El resto del artículo enseña cada obligación con un ejemplo que se puede leer a ojo: primero la fuga, luego la integración, luego la semilla, luego el disco y al final el registro de la corrida.
# Inspección de la API: un Pipeline vacío aún no es un contrato; es una especificación.pipe_demo = Pipeline( [ ("imputer", SimpleImputer(strategy="median")), ("scaler", StandardScaler()), ("clf", LogisticRegression(max_iter=200)), ])print("Pasos declarados:", list(pipe_demo.named_steps))print("¿Tiene predict? ", hasattr(pipe_demo, "predict"))print("¿Está fitted? ", hasattr(pipe_demo, "classes_"))print("clone no copia estado: el original y el clon son especificaciones equivalentes.")pipe_clon = clone(pipe_demo)print("Mismo diseño, distinto objeto:", pipe_demo isnot pipe_clon)print("Hiperparámetros visibles para auditoría:")print({k: v for k, v in pipe_demo.get_params().items() if k.count("__") <=1andnot k.endswith("_")})
La celda anterior no predice nada y, aun así, ya enseña el contrato: hay nombres de paso, hay verbos, y el estado aún no existe. Un objeto “fitted” tendrá mean_, categories_, coef_. Distinguir especificación de estado es lo que permite serializar después del fit y no antes.
2) Por qué falla el preprocesamiento suelto
En una frase: el error no es olvidar escalar; es aprender el escalamiento con información que el modelo no debería ver.
Qué es data leakage en este contexto
Data leakage (fuga de información) ocurre cuando el procedimiento de entrenamiento usa, de forma directa o indirecta, datos que en un escenario real no estarían disponibles en el momento de aprender. El caso clásico es incluir la etiqueta dentro de una feature. El caso silencioso —y el de esta sesión— es más sutil: el test no aparece como columna, pero sí aparece dentro de un estadístico.
StandardScaler no “limpia” el dato. Estima dos parámetros:
[ z = . ]
\(\mu\) y \(\sigma\) se calculan con las filas que el scaler vio en fit. Si el fit se hace sobre \(X\) completo (train + test):
El test influye en cómo se transforma el train: una fila extrema del test mueve \(\mu\) y \(\sigma\).
La métrica en test deja de ser una estimación honesta fuera de muestra.
En producción no existirá ese test. Laboratorio y operación usarán dos \(T\) distintas.
La fuga no necesita ser grande para ser inválida. El problema no es el tamaño del sesgo; es que el protocolo de evaluación quedó contaminado.
Por qué el notebook invita a este error
El preprocesamiento suelto es cómodo: una celda escala, otra codifica, otra entrena. Cada celda “funciona”. El fallo aparece cuando se reordena el trabajo:
se escala antes de partir,
se imputa la media global y después se hace train_test_split,
se ajusta un encoder con todas las categorías del archivo, incluidas las que solo están en test,
se eligen las \(k\) mejores variables mirando \(X\) completo.
Ninguna de esas líneas lanza una excepción. scikit-learn no puede saber que usted considera algunas filas “sagradas”. La frontera de información no es una validación automática: es una decisión de diseño que el pipeline vuelve difícil de violar.
La misma lógica aplica a imputación, encoding, selección, PCA y cualquier paso con fit. Si el paso aprende un parámetro, ese fit pertenece al train.
Dataset didáctico
El laboratorio es deliberadamente pequeño y legible: ingreso (numérica de gran magnitud), visitas y segmento. La etiqueta alta_prioridad depende de las tres, con ruido. No buscamos un F1 alto. Comparamos procedimientos sobre la misma partición.
# Dataset didáctico persistente de la sesión 5.n =450rng = np.random.default_rng(42)ingreso = rng.normal(loc=4_500_000, scale=1_200_000, size=n).clip(800_000, 12_000_000)visitas = rng.poisson(lam=5, size=n).astype(float)segmento = rng.choice(["retail", "pyme", "corporativo"], size=n, p=[0.55, 0.30, 0.15])# Señal: ingreso alto, más visitas y segmento corporativo aumentan P(y=1).logit = (-2.3+1.55* ((ingreso - ingreso.mean()) / ingreso.std())+0.12* visitas+ np.where(segmento =="corporativo", 1.05, 0.0))prob =1/ (1+ np.exp(-logit))y = (rng.random(n) < prob).astype(int)df = pd.DataFrame( {"ingreso": ingreso, "visitas": visitas, "segmento": segmento, "alta_prioridad": y})print("Primeras filas:")print(df.head().to_string(index=False))print("\nDistribución de la etiqueta:")print(df["alta_prioridad"].value_counts(normalize=True).round(3).to_string())print("\nMedia de ingreso (todo el archivo):", f"{df['ingreso'].mean():,.0f}")
Primeras filas:
ingreso visitas segmento alta_prioridad
4,865,660.4957 4.0000 retail 1
3,252,019.0725 5.0000 pyme 0
5,400,541.4350 4.0000 corporativo 1
5,628,677.6597 6.0000 retail 0
2,158,757.7736 7.0000 retail 0
Distribución de la etiqueta:
alta_prioridad
0 0.7490
1 0.2510
Media de ingreso (todo el archivo): 4,477,565
Demostración A: la misma partición, dos scalers
Mantenemos fijos el random_state y la estratificación. Train y test son idénticos en A y en B. La única diferencia es con qué filas se estima \(\mu\) y \(\sigma\). Si el procedimiento A (fit sobre \(X\) completo) aprende otra media, esa diferencia es exactamente la contaminación del test.
# Misma partición; dos formas de ajustar el scaler.X = df[["ingreso", "visitas", "segmento"]]y = df["alta_prioridad"]X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.25, random_state=42, stratify=y)scaler_fuga = StandardScaler().fit(X[["ingreso"]])scaler_honesto = StandardScaler().fit(X_train[["ingreso"]])comparacion_scaler = pd.DataFrame( {"procedimiento": ["A. fit sobre X completo (fuga)", "B. fit solo sobre train"],"media_aprendida": [float(scaler_fuga.mean_[0]), float(scaler_honesto.mean_[0])],"std_aprendida": [float(scaler_fuga.scale_[0]), float(scaler_honesto.scale_[0])], })comparacion_scaler["delta_media_vs_B"] = ( comparacion_scaler["media_aprendida"] - comparacion_scaler.loc[1, "media_aprendida"])print(comparacion_scaler.to_string(index=False))print(f"\nMedia real del test (información que A usó): {X_test['ingreso'].mean():,.0f}")
procedimiento media_aprendida std_aprendida delta_media_vs_B
A. fit sobre X completo (fuga) 4,477,564.9132 1,142,128.7816 -14,679.2032
B. fit solo sobre train 4,492,244.1165 1,143,389.4645 0.0000
Media real del test (información que A usó): 4,433,787
El número importa menos que el principio. Una diferencia de unos miles de pesos en la media no “arruina” el modelo a simple vista; sí arruina la pretensión de que el test es un examen ciego.
Regla operativa:
Cualquier transformación que aprenda parámetros (fit) debe declararse después del split y ajustarse solo con train.
Demostración B: el encoder también fugas
OneHotEncoder aprende un vocabulario. Si se ajusta con el archivo completo, un nivel que solo existe en test entra al diccionario. El modelo ve una columna que, en un escenario honesto, habría sido handle_unknown. La métrica se calcula sobre una representación que no existirá cuando el nivel aparezca por primera vez en producción —porque en producción ese nivel será “desconocido”, no una dummy ya contemplada.
# Un nivel exclusivo de test: el encoder global lo memoriza; el de train no.X_enc = X.copy()X_tr_e, X_te_e = X_train.copy(), X_test.copy()X_te_e = X_te_e.copy()X_te_e.iloc[:5, X_te_e.columns.get_loc("segmento")] ="gobierno"enc_global = OneHotEncoder(handle_unknown="ignore", sparse_output=False).fit(X_enc[["segmento"]])enc_train = OneHotEncoder(handle_unknown="ignore", sparse_output=False).fit(X_tr_e[["segmento"]])print("Vocabulario con fit global :", list(enc_global.categories_[0]))print("Vocabulario con fit train :", list(enc_train.categories_[0]))print("\n¿'gobierno' está en train?:", "gobierno"inset(X_tr_e["segmento"]))print("Columnas al transformar el test (global):", enc_global.transform(X_te_e[["segmento"]]).shape[1])print("Columnas al transformar el test (train) :", enc_train.transform(X_te_e[["segmento"]]).shape[1])print("El encoder honesto no inventa la dummy de gobierno: la ignora (vector de ceros en esas filas).")
Vocabulario con fit global : ['corporativo', 'pyme', 'retail']
Vocabulario con fit train : ['corporativo', 'pyme', 'retail']
¿'gobierno' está en train?: False
Columnas al transformar el test (global): 3
Columnas al transformar el test (train) : 3
El encoder honesto no inventa la dummy de gobierno: la ignora (vector de ceros en esas filas).
Demostración C: el selector también fugas
SelectKBest mira \(X\) e \(y\) y se queda con \(k\) columnas. Si se ajusta con todo el archivo, el test ayuda a decidir qué variables “existen” para el modelo. Es una fuga de diseño, más discreta que la del scaler: las métricas pueden ni moverse y, aun así, el protocolo es inválido.
# SelectKBest: el ranking no debe verse influido por el test.X_num = X[["ingreso", "visitas"]]sel_global = SelectKBest(f_classif, k=1).fit(X_num, y)sel_train = SelectKBest(f_classif, k=1).fit(X_train[["ingreso", "visitas"]], y_train)def nombre_ganador(selector, columnas):return columnas[int(np.argmax(selector.scores_))]print("Ganadora (fit global):", nombre_ganador(sel_global, X_num.columns))print("Scores globales :", dict(zip(X_num.columns, np.round(sel_global.scores_, 3))))print("Ganadora (fit train) :", nombre_ganador(sel_train, X_num.columns))print("Scores train :", dict(zip(X_num.columns, np.round(sel_train.scores_, 3))))print("Aunque gane la misma columna, los scores no tienen por qué coincidir: el test ya votó en A.")
Ganadora (fit global): ingreso
Scores globales : {'ingreso': np.float64(174.339), 'visitas': np.float64(0.247)}
Ganadora (fit train) : ingreso
Scores train : {'ingreso': np.float64(142.15), 'visitas': np.float64(0.037)}
Aunque gane la misma columna, los scores no tienen por qué coincidir: el test ya votó en A.
Las tres demostraciones dicen lo mismo con distintos objetos: el fit es el momento en que la frontera se respeta o se rompe. El pipeline no hace mágica; hace que ese fit ocurra una vez, sobre X_train, y que transform reutilice el estado.
Demostración D: imputar y después partir
Es el orden más común en tutoriales rápidos: “relleno nulos, ya tengo un CSV limpio, ahora sí hago el split”. La mediana del archivo incluye el test. El hueco del examen se rellenó con información del examen.
# Imputación global frente a imputación de train, con nulos sintéticos.X_imp = X.copy()rng_i = np.random.default_rng(11)huecos = rng_i.random(len(X_imp)) <0.15X_imp.loc[huecos, "ingreso"] = np.nanXtr_i, Xte_i, ytr_i, yte_i = train_test_split( X_imp, y, test_size=0.25, random_state=42, stratify=y)mediana_global = X_imp["ingreso"].median()mediana_train = Xtr_i["ingreso"].median()print(f"Mediana con todo X : {mediana_global:,.0f}")print(f"Mediana solo train : {mediana_train:,.0f}")print(f"Δ : {mediana_global - mediana_train:,.0f}")print("El signo del delta es irrelevante. El hecho de que no sea idéntico prueba la fuga.")
Mediana con todo X : 4,539,339
Mediana solo train : 4,572,231
Δ : -32,892
El signo del delta es irrelevante. El hecho de que no sea idéntico prueba la fuga.
3) Integración de modelo y preprocesamiento
En una frase:ColumnTransformer declara ramas (numéricas, categóricas) y Pipeline las encadena con el estimador final, de modo que fit y predict son una sola llamada.
El error de implementación habitual es transformar a mano, convertir a numpy, perder los nombres y entrenar el clasificador “en la celda de al lado”. Funciona hasta que hay que aplicar \(T\) a una fila nueva. Entonces reaparece la receta.
Rama categórica: imputación de moda + one-hot con handle_unknown. Aprende vocabulario.
Estimador: recibe ya \(T(x)\), no el crudo.
remainder="drop" es una decisión explícita: lo que no se listó no entra. remainder="passthrough" es la decisión contraria. Dejar el default sin leerlo es una fuente clásica de columnas fantasma.
# Pipeline integrado sobre la partición honesta de la sección 2.num_cols = ["ingreso", "visitas"]cat_cols = ["segmento"]preprocesador = ColumnTransformer( transformers=[ ("num", Pipeline( [ ("imputer", SimpleImputer(strategy="median")), ("scaler", StandardScaler()), ] ), num_cols, ), ("cat", Pipeline( [ ("imputer", SimpleImputer(strategy="most_frequent")), ("ohe", OneHotEncoder(handle_unknown="ignore", sparse_output=False), ), ] ), cat_cols, ), ], remainder="drop",)pipe = Pipeline( [ ("pre", preprocesador), ("clf", LogisticRegression(max_iter=400, random_state=42)), ])pipe.fit(X_train, y_train)pred = pipe.predict(X_test)print("F1 en test (contrato honesto):", f"{f1_score(y_test, pred):.3f}")print("Accuracy :", f"{accuracy_score(y_test, pred):.3f}")print("El objeto pipe ya contiene T y f. Un solo predict aplica ambos.")
F1 en test (contrato honesto): 0.582
Accuracy : 0.796
El objeto pipe ya contiene T y f. Un solo predict aplica ambos.
# Nombres de features después de T: auditoría de la representación, no cosmética.nombres = pipe.named_steps["pre"].get_feature_names_out()print("Columnas que realmente ve el clasificador:")print(list(nombres))print("\nMedia y escala aprendidas en la rama numérica:")rama_num = pipe.named_steps["pre"].named_transformers_["num"]print(" mean_ :", np.round(rama_num.named_steps["scaler"].mean_, 2))print(" scale_:", np.round(rama_num.named_steps["scaler"].scale_, 2))print("\nVocabulario de segmento:")rama_cat = pipe.named_steps["pre"].named_transformers_["cat"]print(" ", list(rama_cat.named_steps["ohe"].categories_[0]))
Columnas que realmente ve el clasificador:
['num__ingreso', 'num__visitas', 'cat__segmento_corporativo', 'cat__segmento_pyme', 'cat__segmento_retail']
Media y escala aprendidas en la rama numérica:
mean_ : [4.49224412e+06 5.04000000e+00]
scale_: [1.14338946e+06 2.12000000e+00]
Vocabulario de segmento:
['corporativo', 'pyme', 'retail']
Esos nombres y esos mean_ son el estado del contrato. Si mañana un revisor pregunta “¿con qué media se estandarizó ingreso?”, la respuesta no puede ser “la del CSV de abril”. Debe ser este objeto, o su serialización.
FunctionTransformer y transformadores propios
La sesión 4 justificó log1p. Aquí no se reabre esa discusión: se encapsula. FunctionTransformer envuelve una función pura (sin estado). Si la transformación aprende un parámetro, la API correcta es BaseEstimator + TransformerMixin, con fit que escribe atributos con sufijo _.
# log1p como paso nombrado: función pura, sin estado que fugar.def log1p_positivas(X):return np.log1p(np.asarray(X, dtype=float))log_step = FunctionTransformer(log1p_positivas, validate=True)X_log = log_step.fit_transform(X_train[["ingreso"]])print("ingreso crudo (train), primer valor:", f"{X_train['ingreso'].iloc[0]:,.0f}")print("ingreso log1p, primer valor :", f"{X_log[0, 0]:.3f}")print("FunctionTransformer no guarda μ: es T sin fit estadístico.")
ingreso crudo (train), primer valor: 2,751,413
ingreso log1p, primer valor : 14.828
FunctionTransformer no guarda μ: es T sin fit estadístico.
# Transformador con estado: recorta extremos usando cuantiles de TRAIN.class RecortePorCuantiles(BaseEstimator, TransformerMixin):'''Gana un piso y un techo en fit; en transform recorta.'''def__init__(self, q_inf=0.01, q_sup=0.99):self.q_inf = q_infself.q_sup = q_supdef fit(self, X, y=None): X = np.asarray(X, dtype=float)self.piso_ = np.quantile(X, self.q_inf, axis=0)self.techo_ = np.quantile(X, self.q_sup, axis=0)returnselfdef transform(self, X): X = np.asarray(X, dtype=float)return np.clip(X, self.piso_, self.techo_)recorte = RecortePorCuantiles().fit(X_train[["ingreso"]])print("Piso aprendido (q=0.01):", f"{recorte.piso_[0]:,.0f}")print("Techo aprendido (q=0.99):", f"{recorte.techo_[0]:,.0f}")print("Mínimo de test crudo :", f"{X_test['ingreso'].min():,.0f}")print("Si el mínimo de test < piso, transform lo recorta con el umbral de train, no con el de test.")
Piso aprendido (q=0.01): 1,929,838
Techo aprendido (q=0.99): 7,212,829
Mínimo de test crudo : 1,699,661
Si el mínimo de test < piso, transform lo recorta con el umbral de train, no con el de test.
Un transformador propio mal escrito es una fuente privilegiada de fugas: si fit calcula cuantiles de \(X\) “por comodidad”, el test entra otra vez. La convención del sufijo _ (piso_, techo_) no es estilo: señala estado aprendido, que debe serializarse con el resto del pipeline.
Inferencia de una fila: el contrato se juega en el registro único
En producción casi nunca llega un DataFrame de 450 filas. Llega un JSON con un cliente. Si \(T\) exige “el batch completo para calcular la media”, el diseño está roto. El pipeline entrenado ya tiene la media: predict sobre una fila debe bastar.
# Una fila cruda, mismo esquema que el entrenamiento.registro = pd.DataFrame( [{"ingreso": 7_200_000, "visitas": 9, "segmento": "corporativo"}])print("Entrada cruda:")print(registro.to_string(index=False))print("\nProbabilidad predicha:", pipe.predict_proba(registro)[0, 1].round(3))print("Clase predicha :", int(pipe.predict(registro)[0]))print("No recalculamos μ: usamos la que está dentro de pipe.")
Entrada cruda:
ingreso visitas segmento
7200000 9 corporativo
Probabilidad predicha: 0.992
Clase predicha : 1
No recalculamos μ: usamos la que está dentro de pipe.
Orden de columnas y set_output
El contrato incluye el esquema, no solo los valores. Si producción envía (segmento, ingreso, visitas) y el ColumnTransformer fue ajustado con (ingreso, visitas, segmento), scikit-learn reciente selecciona por nombre cuando el input es un DataFrame. Un numpy anónimo no tiene nombres: el orden se vuelve parte del contrato. set_output(transform="pandas") conserva columnas nominadas después de \(T\) y facilita la auditoría.
# El DataFrame tolera un reorder; la matriz anónima no.pipe.set_output(transform="pandas")T_df = pipe.named_steps["pre"].transform(X_test)print("Salida de T como DataFrame:", list(T_df.columns)[:6], "...")X_reordenado = X_test[["segmento", "visitas", "ingreso"]]pred_orden_ok = pipe.predict(X_reordenado)print("¿Predict idéntico tras reordenar columnas del DataFrame?:", np.array_equal(pred, pred_orden_ok))X_anonimo = X_test.to_numpy()[:, ::-1] # orden perverso, sin nombrestry: pipe.predict(pd.DataFrame(X_anonimo))print("Predict sobre anónimo sin nombres: no debería ser confiable.")exceptExceptionas exc:print("Fallo o degradación esperable sin esquema:")print(" ", type(exc).__name__, "-", str(exc).split(chr(10))[0][:160])
Salida de T como DataFrame: ['num__ingreso', 'num__visitas', 'cat__segmento_corporativo', 'cat__segmento_pyme', 'cat__segmento_retail'] ...
¿Predict idéntico tras reordenar columnas del DataFrame?: True
Fallo o degradación esperable sin esquema:
TypeError - 'NoneType' object is not iterable
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but StandardScaler was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but OneHotEncoder was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but StandardScaler was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but OneHotEncoder was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but LogisticRegression was fitted without feature names
warnings.warn(
4) Partición reproducible
En una frase: sin random_state fijo (y, cuando la clase es desbalanceada, sin stratify), el experimento no es una medición: es una anécdota.
La frontera de información exige un corte. Ese corte, si es aleatorio y no se fija, produce otro train y otro test en cada ejecución. Entonces no se puede responder “¿el F1 cambió porque mejoramos \(T\) o porque cambió la suerte del split?”.
Semilla, estratificación y lo que no resuelven
random_state hace determinista el generador. Dos llamadas con la misma semilla y los mismos datos producen el mismo corte.
stratify=y obliga a que la proporción de clases se preserve, aproximadamente, en train y test. En etiquetas al 19 % / 81 %, un split no estratificado puede dejar el test casi sin clase 1: el F1 se vuelve ruido.
Ni la semilla ni la estratificación sustituyen un split temporal cuando hay filtración de futuro (un cliente de 2026 en train y el mismo patrón de 2025 en test, o al revés). Ese caso se menciona aquí y se trata con rigor en orquestación y monitoreo.
Dos splits con la misma semilla deben coincidir en los índices. Un tercero con otra semilla debe diferir. Eso se verifica; no se asume.
# Tres particiones: dos gemelas y una distinta.def indices(seed): tr, te, _, _ = train_test_split( X, y, test_size=0.25, random_state=seed, stratify=y )returnset(tr.index), set(te.index)tr_a, te_a = indices(42)tr_b, te_b = indices(42)tr_c, te_c = indices(99)print("¿Train 42 == train 42?:", tr_a == tr_b)print("¿Test 42 == test 42?:", te_a == te_b)print("¿Train 42 == train 99?:", tr_a == tr_c)print("Solape test 42 ∩ test 99:", len(te_a & te_c), "filas")
¿Train 42 == train 42?: True
¿Test 42 == test 42?: True
¿Train 42 == train 99?: False
Solape test 42 ∩ test 99: 25 filas
# Estratificar o no: el test puede quedarse sin clase rara._, _, _, y_te_estr = train_test_split(X, y, test_size=0.25, random_state=0, stratify=y)_, _, _, y_te_libre = train_test_split(X, y, test_size=0.25, random_state=0, stratify=None)comp_split = pd.DataFrame( {"estratificado": y_te_estr.value_counts(normalize=True).sort_index(),"sin_stratify": y_te_libre.value_counts(normalize=True).sort_index(),"poblacion": y.value_counts(normalize=True).sort_index(), })comp_split.index.name ="clase"print("Proporción de clases en el test frente a la población:")print(comp_split.round(3).to_string())fig, ax = plt.subplots(figsize=(6.8, 3.6))comp_split.plot(kind="bar", ax=ax, rot=0)ax.set_ylim(0, 1)ax.set_title("Balance de clases según el modo de split")plt.tight_layout()plt.show()
Proporción de clases en el test frente a la población:
estratificado sin_stratify poblacion
clase
0 0.7520 0.7430 0.7490
1 0.2480 0.2570 0.2510
Si la barra de la clase 1 en sin_stratify se aleja de la población, el F1 de esa corrida no es comparable con el de otra semilla. Estratificar no “mejora el modelo”: hace comparable la medición.
Cuando el azar no es el corte correcto
Clientes, facturas y sensores tienen tiempo. Un train_test_split aleatorio puede poner el futuro en train y el pasado en test, o mezclar al mismo individuo en ambos lados (group leakage). El contrato de esta sesión sigue valiendo: \(T\) se ajusta solo con el conjunto declarado como train. Lo que cambia es cómo se define ese conjunto. Un corte por fecha (fecha < 2025-01-01) o por GroupShuffleSplit es, conceptualmente, el mismo fit honesto sobre otro índice.
El ejemplo siguiente no modela series de tiempo. Solo muestra que otro criterio de corte produce otro train, y que el pipeline no se entera: hay que decírselo con los índices.
# Corte temporal sintético: la 'fecha' ordena; el azar ya no decide el examen.df_tmp = df.copy()df_tmp["fecha"] = pd.date_range("2024-01-01", periods=len(df_tmp), freq="D")corte = df_tmp["fecha"].quantile(0.75)train_tmp = df_tmp[df_tmp["fecha"] <= corte]test_tmp = df_tmp[df_tmp["fecha"] > corte]print("Train temporal:", train_tmp["fecha"].min().date(), "→", train_tmp["fecha"].max().date(),f"({len(train_tmp)} filas)")print("Test temporal :", test_tmp["fecha"].min().date(), "→", test_tmp["fecha"].max().date(),f"({len(test_tmp)} filas)")print("Solape de índices:", len(set(train_tmp.index) &set(test_tmp.index)))print("El pipeline se ajustaría con train_tmp y se evaluaría en test_tmp: misma API, otro índice.")
Train temporal: 2024-01-01 → 2024-12-02 (337 filas)
Test temporal : 2024-12-03 → 2025-03-25 (113 filas)
Solape de índices: 0
El pipeline se ajustaría con train_tmp y se evaluaría en test_tmp: misma API, otro índice.
5) Serialización: guardar el flujo, no el clasificador
En una frase: el artefacto de producción es \(f \circ T\), no \(f\).
joblib (o pickle) escribe el grafo de objetos de Python. Si ese grafo es el Pipeline ya ajustado, al recargar se recuperan mean_, categories_ y coef_ juntos. Si solo se guarda el clasificador, se recupera un hiperplano que espera \(T(x)\) y se le entregará \(x\) crudo —o un \(T\) reimplementado “parecido”.
joblib, pickle y el riesgo de versión
joblib está optimizado para arrays de NumPy grandes; es la recomendación habitual de scikit-learn.
El archivo no es portable entre versiones mayores de sklearn/numpy con garantía. Un modelo guardado en 1.4 y cargado en 1.6 puede advertir o fallar. Por eso la trazabilidad (sección 6) registra sklearn.__version__.
El archivo es un programa serializado: cárguelo solo de fuentes confiables.
# Persistir el pipeline de la sección 3 y verificar identidad de predicción.ruta_pipe = RUTA_ARTEFACTOS /"pipeline_prioridades.joblib"joblib.dump(pipe, ruta_pipe)pipe_recargado = joblib.load(ruta_pipe)pred_memoria = pipe.predict(X_test)pred_disco = pipe_recargado.predict(X_test)proba_memoria = pipe.predict_proba(X_test)proba_disco = pipe_recargado.predict_proba(X_test)print("Archivo:", ruta_pipe.resolve())print("¿Clases idénticas? ", np.array_equal(pred_memoria, pred_disco))print("Máx. |Δ probabilidad| :", np.max(np.abs(proba_memoria - proba_disco)))print("Una fila nueva, objeto recargado:", int(pipe_recargado.predict(registro)[0]))
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but StandardScaler was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but OneHotEncoder was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but LogisticRegression was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but StandardScaler was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but OneHotEncoder was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but LogisticRegression was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but StandardScaler was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but OneHotEncoder was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but LogisticRegression was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but StandardScaler was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but OneHotEncoder was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but LogisticRegression was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but StandardScaler was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but OneHotEncoder was fitted without feature names
warnings.warn(
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but LogisticRegression was fitted without feature names
warnings.warn(
La identidad bit a bit (o un \(\Delta\) numérico nulo) es la prueba del contrato persistente. Si no coincide, no se discute el F1: se discute el artefacto.
# Contraejemplo pedagógico: persistir solo el clasificador.solo_clf = pipe.named_steps["clf"]ruta_clf = RUTA_ARTEFACTOS /"solo_clasificador.joblib"joblib.dump(solo_clf, ruta_clf)clf_huerfano = joblib.load(ruta_clf)print("El clasificador espera", clf_huerfano.n_features_in_, "columnas (ya transformadas).")print("El crudo tiene ", X_test.shape[1], "columnas (ingreso, visitas, segmento).")try: clf_huerfano.predict(X_test)print("ERROR DIDÁCTICO: no debería haber podido predecir el crudo.")exceptExceptionas exc:print("Fallo esperado al predecir el crudo:")print(" ", type(exc).__name__, "-", str(exc).split("\n")[0][:180])
El clasificador espera 5 columnas (ya transformadas).
El crudo tiene 3 columnas (ingreso, visitas, segmento).
Fallo esperado al predecir el crudo:
ValueError - could not convert string to float: 'pyme'
d:\Users\wilde\anaconda3\Lib\site-packages\sklearn\utils\validation.py:2684: UserWarning: X has feature names, but LogisticRegression was fitted without feature names
warnings.warn(
Ese ValueError (o el silencio peor: una predicción numérica incorrecta si las dimensiones coinciden por accidente) es el argumento contra “guardo el .pkl del modelo”. En un servicio, el síntoma es peor: el endpoint “funciona” y las decisiones están mal.
Validación de esquema: fallar antes de predecir
El contrato asume que \(x\) tiene las mismas columnas, tipos y dominios razonables que el train. Un JSON sin segmento, o con ingreso como texto, no debe llegar al predict para morir en una traza opaca. Se valida el esquema antes.
# Esquema mínimo: columnas obligatorias, tipos coercibles, dominio de segmento.ESQUEMA = {"columnas": ["ingreso", "visitas", "segmento"],"segmentos_conocidos": ["retail", "pyme", "corporativo"],}def validar_entrada(frame: pd.DataFrame, esquema: dict) ->list[str]: errores = [] faltan = [c for c in esquema["columnas"] if c notin frame.columns]if faltan: errores.append(f"faltan columnas: {faltan}")return erroresifnot np.issubdtype(frame["ingreso"].dtype, np.number): errores.append("ingreso no es numérico")ifnot np.issubdtype(frame["visitas"].dtype, np.number): errores.append("visitas no es numérico") extraños =sorted(set(frame["segmento"]) -set(esquema["segmentos_conocidos"]))if extraños: errores.append(f"segmentos no catalogados: {extraños} (el OHE los ignorará)")return erroresok = pd.DataFrame([{"ingreso": 3.1e6, "visitas": 4, "segmento": "pyme"}])mal = pd.DataFrame([{"ingreso": 3.1e6, "visitas": 4}]) # falta segmentonuevo_nivel = pd.DataFrame([{"ingreso": 3.1e6, "visitas": 4, "segmento": "gobierno"}])print("Caso válido :", validar_entrada(ok, ESQUEMA) or"sin errores")print("Caso incompleto:", validar_entrada(mal, ESQUEMA))print("Nivel nuevo :", validar_entrada(nuevo_nivel, ESQUEMA))print("El nivel nuevo no bloquea el predict (handle_unknown=ignore); sí debe quedar en el log.")
Caso válido : sin errores
Caso incompleto: ["faltan columnas: ['segmento']"]
Nivel nuevo : ["segmentos no catalogados: ['gobierno'] (el OHE los ignorará)"]
El nivel nuevo no bloquea el predict (handle_unknown=ignore); sí debe quedar en el log.
6) Trazabilidad mínima de una corrida
En una frase: si no se puede reconstruir qué se entrenó, el F1 no es un resultado: es un rumor.
Un registro mínimo no es MLflow (sesión 8). Es un JSON al lado del joblib con aquello sin lo cual un par no puede repetir la corrida:
El JSON no sustituye al artefacto: lo nombra. Si alguien mueve el joblib y deja el JSON, o al revés, el contrato documental se rompe. En un repositorio serio ambos viajan juntos y con el mismo identificador de corrida (sesión 8).
Métricas del contrato final:
accuracy precision recall f1
0.7960 0.5930 0.5710 0.5820
Inferencia recargada: 0
Artefacto: C:\Users\wilde\Dropbox\bit-blog\mlops-produccion\05-pipelines-reproducibles\artefactos_sesion5\pipeline_integrador.joblib
Lectura. El assert de predicciones idénticas no es pedantería: es el criterio de aceptación del artefacto. El assert del esquema evita el falso positivo de “el servicio respondió 200”. Juntos, sustituyen la frase “en mi máquina corrió”.
8) Errores frecuentes
fit_transform sobre \(X\) completo y después el split.
Guardar model.pkl cuando model es solo el clasificador.
Reimplementar \(T\) en el servicio “porque es más liviano”.
Olvidar handle_unknown y que un nivel nuevo tumbe el endpoint.
Semilla distinta en cada celda y discutir deltas de F1 como si fueran ciencia.
No registrar la versión de sklearn: el joblib se vuelve arqueología.
Imputar, luego partir: la mediana del test ya entró.
Confundir este contrato con un registry (sesión 8) o con un deploy (sesión 9).
9) Checklist del contrato
¿El fit de cada paso ocurre solo en train?
¿predict sobre una fila usa el mismo objeto que el fit?
¿El archivo en disco, al recargarse, reproduce la predicción?
¿El esquema se valida antes de transformar?
¿El JSON nombra semilla, columnas, pasos, métricas y versión?
¿Un compañero podría repetir la corrida sin preguntarle a usted?
Si (6) falla, no hay pipeline: hay un notebook con una clase Pipeline adentro.
10) Conclusiones
Un pipeline reproducible no es un atajo de código. Es el mecanismo que hace que \(f(T(x))\) sea la misma función en el laboratorio, en el archivo y en la primera petición de producción. En esta sesión se argumentó y se midió que:
el preprocesamiento suelto fugas por scaler, por encoder y por selector;
fit / transform / predict son verbos con semántica estadística;
la semilla y la estratificación hacen comparable la medición;
serializar \(f\) sin \(T\) rompe el contrato aunque el F1 del notebook sea alto;
un JSON mínimo convierte la corrida en un hecho auditable.
La sesión 6 usará este contrato como unidad experimental: se compararán modelos sin volver a improvisar \(T\) en celdas sueltas.
11) Taller práctico
Use df, X_train, X_test y la carpeta artefactos_sesion5.
A. Fuga de imputación. Introduzca nulos en ingreso (10 % al azar). Impute la mediana con fit sobre \(X\) completo y, por separado, solo sobre train. Reporte ambas medianas y el \(\Delta\).
B. Transformador propio. Incorpore RecortePorCuantiles en la rama numérica del pipeline. Serialice. Verifique que el objeto recargado tiene piso_ y techo_.
C. Nivel nuevo. Prediga un registro con segmento="gobierno" usando el pipeline de la sección 3. ¿El predict falla o ignora? Contraste con handle_unknown="error".
D. Solo clasificador. Intente predecir X_test crudo con solo_clasificador.joblib. Documente la excepción. Luego predígalo correctamente con pipeline_prioridades.joblib.
E. Trazabilidad. Agregue al JSON el hash o el tamaño en bytes del joblib. ¿Por qué eso detecta un artefacto sustituido?
# Espacio de taller. Los artefactos de las secciones 5 y 7 ya están en disco.print("Archivos en", RUTA_ARTEFACTOS.resolve())for p insorted(RUTA_ARTEFACTOS.iterdir()):print(f" {p.name:40s}{p.stat().st_size:8d} bytes")print("\nResuelva A–E. El criterio de éxito es un contrato verificable, no un F1 más alto.")
Archivos en C:\Users\wilde\Dropbox\bit-blog\mlops-produccion\05-pipelines-reproducibles\artefactos_sesion5
pipeline_integrador.joblib 3642 bytes
pipeline_prioridad.joblib 3137 bytes
pipeline_prioridades.joblib 4170 bytes
registro_corrida.json 654 bytes
solo_clasificador.joblib 911 bytes
Resuelva A–E. El criterio de éxito es un contrato verificable, no un F1 más alto.
💬 ¿Te sirvió?
Deja en los comentarios una duda o un caso donde aplicarías esto — respondo todos. Sígueme para no perderte el próximo artículo de la serie y comparte con alguien que esté aprendiendo arquitectura de IA.
👉 El código completo está disponible para ejecutar directamente.
