Del dato al modelo: pipelines personalizados utilizando python

3er Encuentro Interdisciplinar: Gestión de datos en organizaciones

Karina A. Bartolomé

Especialista en Métodos Cuantitativos para la Gestión y Análisis de Datos en Organizaciones (FCE, UBA). Lic. en Economía (FCE, UNLP). Líder técnica de Ciencia de Datos (Ualá).

2025-09-05

0.1 Consideraciones previas

Este taller está enfocado en cuestiones avanzadas sobre el flujo de modelado de datos en python, se recomienda algún conocimiento previo para un mejor entendimiento del código.
Durante el seminario se utilizarán los siguientes paquetes (python):

📦 pandas==2.3.2

📦 numpy==2.2.0

📦 scikit-learn==1.7.1

📦 catboost==1.2.8

📦 matplotlib==3.10.6

📦 seaborn==0.13.2

📦 great_tables==0.18.0

Todo el código se encuentra disponible en Repositorio y en Google Colab

1 Planteo del caso

1.1 Prevención del fraude transaccional

Lectura de datos

import kagglehub
path = kagglehub.dataset_download("kartik2112/fraud-detection")
df_train = pd.read_csv(f"{path}/fraudTrain.csv")
df_test = pd.read_csv(f"{path}/fraudTest.csv")
df = pd.concat([df_train, df_test], axis=0, ignore_index=True).reset_index(drop=True)

target = 'is_fraud'
cols_selected = [
    "trans_date_trans_time",
    "merchant",
    "category",
    "amt",
    "city_pop",
    "job",
    "dob",
    "lat",
    "long",
    "merch_lat",
    "merch_long",
    "is_fraud",
]
df = df[cols_selected]

Se cuenta con un dataset de 1852394 transacciones (compras). Son 12 variables generadas sintéticamente por lo que el foco está sobre cómo procesar los datos y no sobre la performance del modelo ¹.

No se cuenta con valores faltantes, por lo que se genera “ruido” en el dataset, añadiendo valores faltantes en distintas variables para el ejemplo.

Generador de ruido en el dataset

## Valores faltantes
def add_random_nans(values, fraction=0.2):
    """
    Generador de valores faltantes
    """
    np.random.seed(42)
    mask = np.random.rand(len(values)) < fraction
    new_values = values.copy()
    new_values[mask] = np.nan
    return new_values

df = df.assign(
    merchant = lambda x: [i.replace('fraud_','') for i in x['merchant']],
    dob = lambda x: add_random_nans(x['dob'], fraction=0.05),
    job = lambda x: add_random_nans(x['job'], fraction=0.1),
    city_pop = lambda x: add_random_nans(x['city_pop'], fraction=0.03),
    merch_lat = lambda x: add_random_nans(x['merch_lat'], fraction=0.02),
    merch_long = lambda x: add_random_nans(x['merch_long'], fraction=0.02),
)

Ante una nueva transacción, ¿cuál es la probabilidad de que sea fraudulenta? ¿Debería bloquarse?

La variable objetivo (target) es is_fraud , donde el porcentaje de observaciones de clase 1 (fraudulentos) es 0.52%.

$P(\color{#FF9933}{is\_fraud}=1) = f(\color{#3399FF}{X})$

$\color{#FF9933}{is\_fraud}$: variable que puede tomar 2 valores: 1 (transacción fraudulenta) o 0 (transacción legítima)

$\color{#3399FF}{X}$: matriz nxm, siendo n la cantidad de observaciones y m la cantidad de variables (o atributos)

Tabla 1: Datos transaccionales (muestra de 4 observaciones)

is_fraud	trans_date_trans_time	merchant	category	amt	city_pop	job	dob	lat	long	merch_lat	merch_long
0	2019-08-29 02:46:55	Harris Inc	gas_transport	$67.94	302.0	Magazine features editor	1973-05-04	32.68	-81.25	32.28	-81.21
1	2019-01-10 22:14:49	Kuhic, Bins and Pfeffer	shopping_net	$1,161.58	276002.0	Medical technical officer	1950-12-14	26.33	-81.59	25.92	-82.5
0	2019-12-25 09:45:55	Kutch, Hermiston and Farrell	gas_transport	$53.81	2408.0	Sales professional, IT	1997-07-05	32.55	-80.31	33.38	-81.0
0	2020-07-10 10:18:00	Kovacek, Dibbert and Ondricka	grocery_pos	$166.69	337.0	Occupational psychologist	1954-07-05	38.24	-122.91	37.45	-122.52

Fuente de los datos: Credit Card Transactions Fraud Detection Dataset.

2 Esquema de modelado

2.1 Esquema de modelado

La Figura Figura 2 muestra un posible esquema de trabajo para modelos de aprendizaje automático en donde se busca predecir sobre datos nuevos. Quemy et al. (2020) destaca la importancia de diseñar primero el flujo de datos y luego destinar tiempo al ajuste de hiperparámetros del modelo.

Figura 2: Diagrama caso

2.2 Particiones

El dataset se separa en un segmento para entrenamiento (train) y otro para la evaluación (test). En general se habla de particiones para el entrenamiento del modelo pero en este caso se trabajará con 2 particiones para el ajuste del procesamiento y modelado en conjunto.

Generación de particiones

y = df[target]
X = df.drop([target], axis=1)

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.3, shuffle=True, stratify=y, random_state=42
)

N observaciones para entrenamiento: 1296675 (0.52% de transacciones fraudulentas)

N observaciones para evaluación: 555719 (0.52% de transacciones fraudulentas)

3 Preprocesamiento

3.1 Tipos de transformaciones

Ciertos tipos de transformaciones requieren “aprender” algunos aspectos de los datos de entrenamiento mientras que otras no.

Ejemplos de transformaciones que dependen de los datos de entrenamiento :

Imputación de valores faltantes con la mediana → La mediana depende de los datos
Escalado → Se debe calcular la media y desvío estándar de los datos

Ejemplos de transformaciones que no dependen de los datos de entrenamiento:

Construcción de una nueva variable mediante un cálculo simple → x^2
Combinaciones de variables en una nueva variable → x/y

3.2 Transformaciones en python, mediante scikit-learn

Para implementar este tipo de aprendizajes de ciertos aspectos de los datos al generar transformaciones custom, en scikit-learn se utiliza una clase específica TransformerMixin.

Figura 3: CustomTransformers en scikit-learn

3.3 Transformaciones iniciales

Cálculo de la edad
Cálculo de la distancia entre el comercio y el usuario
Generación de variables vinculadas a la fecha y hora de la transacción

Transformaciones iniciales

class TransformacionesIniciales(BaseEstimator, TransformerMixin):

    def __init__(self, timestamp_features=True, distance_features=True):
        """
        Args:
            timestamp_features (bool): Generar variables basadas en la fecha/hora de la trx
            distance_features (bool): Generar variables basadas en la distancia al comercio
        """
        self.timestamp_features = distance_features
        self.distance_features = distance_features

    def fit(self, X, y=None):
        return self

    def transform(self, X, y=None):
        X_ = X.copy() # Copia para no afectar al df original

        # Casteo de variables
        X_ = X_.assign(
            dob = lambda x: pd.to_datetime(x['dob'], errors='coerce'),
            trans_date_trans_time = lambda x: pd.to_datetime(
                x["trans_date_trans_time"], errors="coerce"
            ),
        )

        # Cálculo de edad
        X_ = X_.assign(
            age = lambda x: round((x['trans_date_trans_time']-x['dob']).dt.days / 365.25,2)
        )

        # Features basadas en fecha y hora de la trx:
        if self.timestamp_features:
            X_ = X_.assign(
                trans_date__year = lambda x: x["trans_date_trans_time"].dt.year,
                trans_date__month = lambda x: x["trans_date_trans_time"].dt.month,
                trans_date__day = lambda x: x["trans_date_trans_time"].dt.day,
                trans_date__dow = lambda x: x["trans_date_trans_time"].dt.dayofweek,
                trans_date__hour = lambda x: x["trans_date_trans_time"].dt.hour,
                trans_date__partofday = lambda x: x['trans_date__hour'].apply(categorizar_hora)
            )

        if self.distance_features:
            # Distancia (en kilometros)
            X_["distance_to_merch"] = calcular_distancia_haversine(
                X_["lat"], X_["long"], X_["merch_lat"], X_["merch_long"]
            )

        X_ = X_.drop(['trans_date_trans_time','dob'], axis=1) 
        return X_

.fit_transform()

transformaciones = TransformacionesIniciales()
transformaciones.fit(X_train)
X_test_transformed = transformaciones.transform(X_test)

Tabla 2: Transformaciones iniciales

merchant	category	amt	city_pop	job	lat	long	merch_lat	merch_long	age	trans_date__year	trans_date__month	trans_date__day	trans_date__dow	trans_date__hour	trans_date__partofday	distance_to_merch
Harris Group	food_dining	$40.23	3451.0	Financial trader	33.92	-89.68	33.1	-89.25	35.3	2019	9	8	6	14	tarde	99.84
Ledner-Pfannerstill	gas_transport	$61.23	3807.0	Surgeon	43.97	-71.15	43.21	-71.55	19.23	2019	1	16	2	5	madrugada	91.2
Wilkinson LLC	personal_care	$64.80	2258.0	Building surveyor	41.46	-74.17	42.27	-74.9	82.96	2020	3	2	0	13	tarde	108.48
Fisher-Schowalter	shopping_net	$8.61	3096.0	Social research officer, government	44.86	-85.81	45.85	-85.69	44.15	2019	12	4	2	22	noche	110.39

3.4 Feature engineering

Scikit-learn cuenta con múltiples transformers ya definidos (SimpleImputer, MinMaxScaler, etc). Además, se definen 4 transformers custom adicionales (MeanEncoder, RareCategoryGrouper, ColumnCapper, IsolationForestTransformer):

MeanEncoder

class MeanEncoder(BaseEstimator, TransformerMixin):
    """
    Encoding de variables categóricas mediante el promedio de la target en esa categoría.
    """

    def __init__(self, variables=None):
        self.variables = variables
        self.encoding_dict_ = {}
        self.global_mean_ = None

    def fit(self, X, y):
        X_ = X.copy()
        y_ = pd.Series(y)

        if self.variables is None:
            self.variables = X_.select_dtypes(
                include=["object", "category"]
            ).columns.tolist()

        self.global_mean_ = y_.mean()

        for var in self.variables:
            self.encoding_dict_[var] = y_.groupby(X_[var]).mean().to_dict()

        return self

    def transform(self, X):
        X_ = X.copy()
        for var in self.variables:
            X_[var] = X_[var].map(self.encoding_dict_[var])
            # Para nuevas categorías asigna el valor promedio
            X_[var] = X_[var].fillna(self.global_mean_)
        return X_

RareCategoryGrouper

class RareCategoryGrouper(BaseEstimator, TransformerMixin):
    """
    Agrupar categorías poco frecuentes en "infrequent"
    """

    def __init__(self, variables=None, min_freq=0.05):
        """
        Args:
            variables (list): List of categorical variables to group.
            min_freq (float or int):
                Si float (0 < min_freq < 1), mínima proporción en el dataset.
                Si int (>=1), mínima cantidad de observaciones.
        """
        self.variables = variables
        self.min_freq = min_freq
        self.frequent_categories_ = {}

    def fit(self, X, y=None):
        X_ = X.copy()

        if self.variables is None:
            self.variables = X_.select_dtypes(include="object").columns.tolist()

        for var in self.variables:
            freqs = X_[var].value_counts(normalize=isinstance(self.min_freq, float))
            self.frequent_categories_[var] = freqs[
                freqs >= self.min_freq
            ].index.tolist()

        return self

    def transform(self, X):
        X_ = X.copy()
        for var in self.variables:
            X_[var] = X_[var].where(
                X_[var].isin(self.frequent_categories_[var]), "infrequent"
            )
        return X_

ColumnCapper

class ColumnCapper(BaseEstimator, TransformerMixin):
    """
    Cappear las variables mediante el método de IQR
    """

    def __init__(self, numeric_features=None, factor=1.5):
        """
        Args:
            numeric_features (list): lista de variables sobre las cuales remover valores atípicos
            factor (float): multiplicador del IQR
        """
        self.numeric_features = numeric_features
        self.factor = factor

    def fit(self, X, y=None):
        X_ = X.copy()
        if self.numeric_features is None:
            self.numeric_features = X_.select_dtypes(include="number").columns.tolist()
        # IQR
        Q1 = X_[self.numeric_features].quantile(0.25)
        Q3 = X_[self.numeric_features].quantile(0.75)
        IQR = Q3 - Q1
        self.lower_bound = Q1 - self.factor * IQR
        self.upper_bound = Q3 + self.factor * IQR
        return self

    def transform(self, X, y=None):
        X_ = X.copy()
        for col in self.numeric_features:
            X_[col] = X_[col].clip(
                lower=self.lower_bound[col], 
                upper=self.upper_bound[col]
            )
        return X_

Devarakonda (2023) propone un enfoque híbrido para la detección de fraude, combinando modelos supervisados y no supervisados (Issolation Forest). En este caso, se utilizará un score de cuán anómala es la transacción como variable explicativa del fraude.

IsolationForestTransformer

class IsolationForestTransformer(BaseEstimator, TransformerMixin):
    """
    Ajustar un modelo IssolationForest para detección de anomalías. 
    Retorna un anomaly_score. 
    """

    def __init__(self, **kwargs):
        self.iforest_kwargs = kwargs

    def fit(self, X, y=None):
        self.iforest_ = IsolationForest(**self.iforest_kwargs)
        self.iforest_.fit(X)
        return self

    def transform(self, X):
        scores = self.iforest_.decision_function(X)
        return pd.DataFrame({"anomaly_score": scores}, index=X.index)

Se crea un Pipeline de preprocesamiento general:

Transformaciones iniciales
Transformer que según el tipo de variable (numérica o categórica) aplica ciertas transformaciones:
- Pipeline para procesamiento de variables categóricas
- Pipeline para procesamiento de variables numéricas
Identificación de observaciones anómalas y cappeo de variables según el IQR

Preprocesamiento

preproc_categoricas = Pipeline(steps=[
    ('rare_labels', RareCategoryGrouper(min_freq=0.01)),
    ('imputar_nulos', SimpleImputer(missing_values=np.nan, strategy='most_frequent')),
    ('mean_encoder', MeanEncoder())
])

preproc_numericas = Pipeline(steps=[
    ('imputar_nulos', SimpleImputer(strategy='median'))
])

feature_eng = ColumnTransformer([
    ('cat', preproc_categoricas, make_column_selector(dtype_exclude=['float','int'])),
    ('num', preproc_numericas, make_column_selector(dtype_include=['float','int']))
], verbose_feature_names_out=False, remainder='drop', verbose=True)

features_preprocessing = Pipeline([
    ('data_cleaning', TransformacionesIniciales()),
    ('feature_eng', feature_eng),
    ('anomalies', FeatureUnion([
            ('outliers', ColumnCapper()),
            ('anomaly', IsolationForestTransformer())
        ])
    )
], verbose=True)

3.5 Datos transformados

.fit_transform()

features_preprocessing.fit(X_train, y_train)
X_test_transformed = features_preprocessing.transform(X_test)

[Pipeline] ..... (step 1 of 3) Processing data_cleaning, total=   1.5s
[ColumnTransformer] ........... (1 of 2) Processing cat, total=   1.7s
[ColumnTransformer] ........... (2 of 2) Processing num, total=   1.8s
[Pipeline] ....... (step 2 of 3) Processing feature_eng, total=   3.6s
[Pipeline] ......... (step 3 of 3) Processing anomalies, total=   1.3s

Tabla 3: Datos post procesamiento (muestra de 4 observaciones de la partición de evaluación)

merchant	category	job	trans_date__partofday	amt	city_pop	lat	long	merch_lat	merch_long	age	trans_date__year	trans_date__month	trans_date__day	trans_date__dow	trans_date__hour	distance_to_merch	anomaly_score
0.005	0.001	0.005	0.001	40.23	3451.0	33.922	-89.678	33.097	-89.252	35.3	2019.0	9.0	8.0	6.0	14.0	99.842	0.058
0.005	0.004	0.005	0.009	61.23	3807.0	43.974	-71.15	43.206	-71.548	19.23	2019.0	1.0	16.0	2.0	5.0	91.196	0.008
0.005	0.002	0.005	0.001	64.8	2258.0	41.458	-74.166	42.268	-74.895	82.96	2020.0	3.0	2.0	0.0	13.0	108.475	0.002
0.005	0.013	0.005	0.011	8.61	3096.0	44.86	-85.814	45.849	-85.687	44.15	2019.0	12.0	4.0	2.0	22.0	110.389	-0.032

4 Modelo

4.1 Pipeline de modelado

Se crea un nuevo Pipeline que incluye un primer paso de preprocesamiento y un segundo paso de modelado (un clasificador).

Pipeline

clf = CatBoostClassifier(
    iterations=500,
    loss_function="Logloss",
    class_weights=[1, 20],
    random_seed=42,
    verbose=100,
)

pipe = Pipeline([
    ("preproc", features_preprocessing), 
    ("model", clf)
], verbose=True)

4.2 Entrenamiento del pipeline completo

Durante el entrenamiento del pipeline (preprocesamiento + modelo), se visualizan los tiempos que tarda cada uno de los pasos:

pipe.fit(X_train, y_train)


[Pipeline] ..... (step 1 of 3) Processing data_cleaning, total=   1.9s
[ColumnTransformer] ........... (1 of 2) Processing cat, total=   2.4s
[ColumnTransformer] ........... (2 of 2) Processing num, total=   2.7s
[Pipeline] ....... (step 2 of 3) Processing feature_eng, total=   5.3s
[Pipeline] ......... (step 3 of 3) Processing anomalies, total=  10.8s
0:      learn: 0.4501382    total: 98.1ms   remaining: 48.9s
100:    learn: 0.0492939    total: 13.2s    remaining: 52.3s
200:    learn: 0.0390136    total: 26.7s    remaining: 39.8s
300:    learn: 0.0324107    total: 39.1s    remaining: 25.9s
400:    learn: 0.0280930    total: 52.5s    remaining: 13s
499:    learn: 0.0247809    total: 1m 5s    remaining: 0us

Almacenar el modelo para luego utilizarlo (despliegue):

pickle.dump(): Guardar el modelo

with open('artifacts/pipe_model_fraud.pkl', 'wb') as file:
    pickle.dump(pipe, file)

5 Predicciones

5.1 Predicciones sobre datos nuevos (despliegue de modelos)

Cargar el archivo .pkl para utilizarlo:

pickle.load(): Cargar el modelo

with open('artifacts/pipe_model_fraud.pkl', 'rb') as file:
    pipe = pickle.load(file)

🆕 Datos de una nueva transacción (datos en producción):

Nueva transacción

nueva_trx = pd.DataFrame({
    "trans_date_trans_time": "2019-10-09 20:38:49",
    "merchant": np.nan,
    "category": "gas_transport",
    "amt": 9.66,
    "city_pop": 10000,
    "job": np.nan,
    "dob": "1995-08-16",
    "zip": np.nan,
    "lat": 45.8433,
    "long": -113.1948,
    "merch_lat": 45.837213,
    "merch_long": -113.191425,
}, index=["nueva_trx"])

Tabla 4: Nueva observación

trans_date_trans_time	merchant	category	amt	city_pop	job	dob	zip	lat	long	merch_lat	merch_long
2019-10-09 20:38:49		gas_transport	$9.66	10000		1995-08-16		45.84	-113.19	45.84	-113.19

Utilizar el modelo para estimar la probabilidad de que la nueva transacción sea fraudulenta:

Predicción de probabilidad

y_pred = pipe.predict_proba(nueva_trx)[:,1]

La probabilidad de que la transacción sea fraudulenta es: 1.62% → ¿Aprobar o rechazar la transacción? Depende de la aversión al riesgo de la entidad en cuestión.

6 Comentarios finales

6.1 Comentarios finales

El uso de pipelines en python permite organizar el flujo de procesamiento de datos de manera clara, reproducible y escalable.
Esto permite reducir el riesgo durante el despliegue de modelos al eliminar código duplicado o transformaciones inconsistentes entre entrenamiento y predicción.
Todas las transformaciones mostradas son ejemplos ilustrativos, otras transformaciones podrían generar mejores resultados.

6.2 Referencias / Recursos

Bartolome, Karina. 2024. «Calibración de probabilidades: Estimación de riesgo crediticio mediante modelos de machine learning». https://karbartolome.github.io/workshops/20240513-uba-calibracion/slides.

Devarakonda, Rahul Roy. 2023. «Machine Learning Approach for Fraud Detection in a Financial Services Application». SSRN Electronic Journal. https://doi.org/10.2139/ssrn.5234670.

Quemy, Anthony, Jonathan Vayssiere, Paul De La Noue, y Antoine Sénéjoux. 2020. «Two-stage optimization for machine learning workflow». Expert Systems with Applications 140: 112876.

6.3 Contacto

karinabartolome

karbartolome

Blog