Cómo detectar drift de características en Machine Learning: paso a paso

Detectar drift de características en Machine Learning es útil para entender cuándo los datos en producción cambian respecto a los datos de entrenamiento, evitando pérdida de rendimiento. Este tutorial muestra, paso a paso, cómo calcular medidas simples de drift por característica y señalar cambios relevantes con Python.

Requisitos previos

• Python 3.8+ instalado
• Pandas y scipy instalados (pip install pandas scipy)
• Conjunto de entrenamiento y de producción como archivos CSV
• Conocimientos básicos de Python y estadística descriptiva

Paso 1: ¿Por qué detectar drift de características?

El drift de características (feature drift) ocurre cuando la distribución de una columna cambia a lo largo del tiempo. Identificar drift ayuda a decidir si es necesario retrain, ajustar el preprocesamiento o señalizar alertas. Vamos a usar pruebas estadísticas y medidas prácticas para cada característica.

Paso 2: Preparar datos y cargar librerías

Cargue los CSV del conjunto de entrenamiento y del conjunto de producción y seleccione las columnas que quiere monitorizar. Mantenemos solo columnas numéricas en este ejemplo; para categóricas usamos comparaciones por frecuencias.

import pandas as pd
from scipy.stats import ks_2samp, chi2_contingency

train = pd.read_csv('train.csv')
prod = pd.read_csv('prod.csv')

# Lista de columnas a monitorizar (numéricas y categóricas)
num_cols = ['age','salary']
cat_cols = ['region','product_type']

Paso 3: Calcular drift para características numéricas (Kolmogorov-Smirnov)

La prueba de Kolmogorov-Smirnov compara dos muestras para ver si provienen de la misma distribución. No asume normalidad. Usamos p-value y una medida de efecto (D) para señalar drift.

def detect_numeric_drift(train_ser, prod_ser, alpha=0.05):
    # Remove NaNs
    a = train_ser.dropna()
    b = prod_ser.dropna()
    if len(a) < 20 or len(b) < 20:
        return {'p_value': None, 'D': None, 'drift': 'insufficient_data'}
    stat, p = ks_2samp(a, b)
    drift = 'drift' if p < alpha else 'no_drift'
    return {'p_value': float(p), 'D': float(stat), 'drift': drift}

results_num = {}
for col in num_cols:
    results_num[col] = detect_numeric_drift(train[col], prod[col])

print(results_num)

Paso 4: Calcular drift para características categóricas (prueba chi-cuadrado)

Para categóricas con conteos, comparamos tablas de frecuencia. Si hay categorías nuevas o eliminadas, esto indica drift cualitativo.

def detect_categorical_drift(train_ser, prod_ser, alpha=0.05):
    a = train_ser.fillna('NULL')
    b = prod_ser.fillna('NULL')
    # Frecuencias alineadas por categoría
    freq = pd.concat([a.value_counts(), b.value_counts()], axis=1, sort=False).fillna(0)
    freq.columns = ['train', 'prod']
    chi2, p, _, _ = chi2_contingency(freq.values)
    different_categories = set(b.unique()) - set(a.unique())
    drift = 'drift' if p < alpha or len(different_categories) > 0 else 'no_drift'
    return {'p_value': float(p), 'different_categories': list(different_categories), 'drift': drift}

results_cat = {}
for col in cat_cols:
    results_cat[col] = detect_categorical_drift(train[col], prod[col])

print(results_cat)

Paso 5: Medidas complementarias y umbrales prácticos

Además de pruebas estadísticas, es útil calcular diferencias de media, mediana y proporciones para priorizar features. Defina umbrales prácticos (ej.: diferencia de media > 10% o D > 0.1).

def numeric_statistics(train_ser, prod_ser):
    t = train_ser.dropna()
    p = prod_ser.dropna()
    stats = {
        'train_mean': float(t.mean()),
        'prod_mean': float(p.mean()),
        'mean_rel_diff': abs(t.mean() - p.mean()) / (abs(t.mean()) + 1e-9)
    }
    return stats

summary = {}
for col in num_cols:
    summary[col] = {**results_num[col], **numeric_statistics(train[col], prod[col])}

print(summary)

Paso 6: Automatizar y generar informe

Junte todo en una función que itere por columnas, aplique pruebas y escriba un CSV de alerta con columnas señaladas por prioridad.

def drift_report(train_df, prod_df, num_cols, cat_cols):
    rows = []
    for col in num_cols:
        r = detect_numeric_drift(train_df[col], prod_df[col])
        stats = numeric_statistics(train_df[col], prod_df[col])
        priority = 'high' if r['drift']=='drift' and stats['mean_rel_diff'] > 0.1 else 'low'
        rows.append({'feature': col, 'type': 'numeric', **r, **stats, 'priority': priority})
    for col in cat_cols:
        r = detect_categorical_drift(train_df[col], prod_df[col])
        priority = 'high' if r['drift']=='drift' else 'low'
        rows.append({'feature': col, 'type': 'categorical', **r, 'priority': priority})
    return pd.DataFrame(rows)

report = drift_report(train, prod, num_cols, cat_cols)
report.to_csv('drift_report.csv', index=False)
print(report)

Verificar el resultado

Abra drift_report.csv: debe listar cada característica con p_value, indicador de drift y priority. Revise las columnas marcadas como high: inspeccione histogramas o boxplots para confirmar visualmente. Si varios features están en high, considere retrain, recalibrar umbrales o ajustar ETL.

Conclusión

Este método simple combina pruebas estadísticas (KS y chi-cuadrado) con medidas prácticas para detectar drift de características en Machine Learning. Siguientes pasos: integrar esto en un pipeline de monitorización en producción, añadir pruebas por ventana temporal y usar técnicas de explicabilidad para entender causas. Consejo: empiece a monitorizar gradualmente y ajuste umbrales según el coste de falso positivo/negativo en su escenario.