Crea un modelo de predicción de abandono de producción utilizando MLOps.

En cualquier negocio basado en suscripciones o con ingresos recurrentes, la fuga de clientes es el factor silencioso que impide el crecimiento. Adquirir un nuevo cliente es mucho más costoso que retener a uno existente. Si bien los equipos de inteligencia empresarial pueden reportar datos históricos sobre la fuga, los líderes técnicos son responsables de construir los sistemas que predicen la futura fuga, lo que permite una intervención proactiva.

Esto no es un ejercicio para desarrollar un modelo de predicción de abandono. Un modelo de predicción de abandono de nivel de producción es un sistema de software complejo y en constante evolución que requiere una arquitectura sólida, operaciones MLOps disciplinadas y una estrategia de implementación clara.

Este artículo proporciona un esquema técnico y completo de ingeniería para construir, implementar y mantener un modelo de predicción de abandono (churn). Nos centraremos en las decisiones arquitectónicas, la implementación del flujo de trabajo y los desafíos operativos que enfrentará, más allá de la teoría del modelo.

El plano arquitectónico

El modo de fallo más común es un modelo que funciona bien en un entorno Jupyter Notebook pero no puede integrarse en el negocio. Un sistema de producción debe ser fiable, auditable y escalable.

Su arquitectura debe separar el procesamiento de datos, el entrenamiento del modelo, y la inferencia del modelo.

Servicios de Ingeniería de Productos

Trabaje con nuestros gestores de proyectos, ingenieros de software y testers de control de calidad para desarrollar su nuevo producto de software personalizado o para apoyar su flujo de trabajo actual, siguiendo metodologías Agile, DevOps y Lean.

Build with 4Geeks

Un plano arquitectónico bien definido se ve así:

1. Ingesta y transformación de datos: Los datos brutos (registros de aplicaciones, eventos de facturación, tickets de soporte) llegan a un lago de datos (p. ej., S3, GCS) o almacén (p. ej., Snowflake, BigQuery, Redshift).
2. Ingeniería de características: Un trabajo programado (utilizando dbt, Spark o un orquestador como Airflow/Prefect) se ejecuta diariamente para transformar los datos brutos en una tienda de características o una simple "tabla de entrada de modelo." Esta tabla es la única fuente de verdad tanto para el entrenamiento como para la inferencia.
3. Pipeline de entrenamiento: Un trabajo separado y programado (p. ej., un DAG semanal de Airflow) recupera los datos de entrenamiento del almacén de características, realiza la selección del modelo y entrena al candidato de modelo final.
4. Registro de modelos: El modelo entrenado, sus métricas de rendimiento y sus artefactos serializados se versionan y registran en un registro de modelos(p. ej., MLflow, Vertex AI Registry, SageMaker Model Registry). Esto es crucial para la gobernanza y el reenvío.
5. Servicio de inferencia:
   • Lote (Offline): Un trabajo diario carga el modelo "de producción" del registro, evalúa a todos los clientes activos y escribe las probabilidades de abandono en una base de datos para el equipo de Customer Success.
   • En tiempo real (Online): Una API contenedorizada (p. ej., FastAPI, Flask) carga el modelo y expone un /predictendpoint. Esto permite que otros servicios obtengan predicciones instantáneas (p. ej., "¿Debemos mostrar esta oferta de descuento a este usuario ahora mismo?").
6. Monitoreo: Los paneles y las alertas rastrean el desplazamiento de datos(¿están cambiando las entradas?) y el desplazamiento del modelo(¿se está degradando el rendimiento?).

Fase 1: Ingeniería de características y la variable objetivo

Esta es la fase más crítica. La calidad de los datos determina el resultado. Un modelo sencillo con excelentes características siempre superará a un modelo complejo con características deficientes.

Definir la variable objetivo ("El evento de abandono")

Primero, obtenga una definición precisa y sin ambigüedades de "abandono" del negocio.

• ¿Es una cancelación activa?
• ¿Es una incumplimiento de renovación de un contrato?
• ¿Es una inactividad prolongada?

A continuación, defina su ventana de predicción. Un objetivo común y muy práctico es:

target = 1 si el cliente abandonó el servicio dentro de los 30 días posteriores a la fecha en que se calcularon las características.

target = 0 de lo contrario.

Evitar fugas de datos

El error más grave en el modelado de series temporales es fugas de datos—utilizar información de tus datos de entrenamiento que no habría estado disponible en el momento de la predicción.

Ejemplo: Calculando longitud_promedio_de_sesión durante un período de 90 días.

• Incorrect (Fuga de datos): Para un usuario el 1 de junio, se utilizan los datos desde el 1 de junio hasta el 30 de agosto. Se está analizando el futuro.
• Correcto (Momento específico): Para un usuario el 1 de junio, solo se utilizan los datos desde el 1 de marzo hasta el 31 de mayo.

Su flujo de trabajo de ingeniería de características debe garantizar estrictamente la corrección en el momento.

Ejemplos de características concretas

Aquí se presentan las categorías de características comunes. Su consulta para crear esta "tabla de entrada de modelo" será el código SQL o Spark más complejo del proyecto.

Fase 2: Canal de entrenamiento del modelo

Para datos estructurados y tabulares, los modelos complejos de aprendizaje profundo rara vez son la mejor opción. Árboles de Decisión con Refuerzo de Gradiente (GBDT) son el algoritmo dominante para esta tarea, con XGBoost y LightGBM siendo de última generación. Una solución más simple como la Regresión Logística es una excelente opción, especialmente si el negocio requiere alta interpretabilidad.

El pipeline de Scikit-learn

No se debe escribir código de preprocesamiento aislado. Aplique toda su lógica de preprocesamiento y modelado en un objeto "Pipeline". Esto evita la desalineación de los datos entre el entrenamiento e la inferencia, ya que las mismas transformaciones (imputación, escalado) se guardan junto con el modelo.Pipeline object.with the model.

Aquí hay un ejemplo listo para producción utilizando scikit-learnPipelineColumnTransformer,..

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from xgboost import XGBClassifier

# --- 1. Define feature types ---
# Assume X_train is a pandas DataFrame loaded from your feature store
numeric_features = ['days_since_last_login', 'session_length_avg_90d', 'support_tickets_opened_30d']
categorical_features = ['plan_type', 'user_region']

# --- 2. Create preprocessing transformers ---
# Pipeline for numeric features: Impute missing values with the median, then scale
numeric_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='median')),
    ('scaler', StandardScaler())
])

# Pipeline for categorical features: Impute missing with a constant, then one-hot encode
categorical_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='constant', fill_value='missing')),
    ('onehot', OneHotEncoder(handle_unknown='ignore'))
])

# --- 3. Combine transformers with ColumnTransformer ---
# This applies the correct transformer to the correct column
preprocessor = ColumnTransformer(
    transformers=[
        ('num', numeric_transformer, numeric_features),
        ('cat', categorical_transformer, categorical_features)
    ],
    remainder='passthrough' # Pass through any columns not specified
)

# --- 4. Handle Imbalanced Data ---
# Churn rates are low (e.g., 2%). Simply using accuracy is useless.
# We must either oversample (e.g., SMOTE) or use class weighting.
# XGBoost's `scale_pos_weight` is highly effective and computationally cheaper than SMOTE.
# scale_pos_weight = count(negative_class) / count(positive_class)
y_train = # ... your target variable (0s and 1s)
scale_pos_weight = (y_train == 0).sum() / (y_train == 1).sum()

# --- 5. Create the full model pipeline ---
model = XGBClassifier(
    objective='binary:logistic',
    eval_metric='aucpr',  # Area Under Precision-Recall Curve: The best metric for imbalanced data
    scale_pos_weight=scale_pos_weight,
    n_estimators=200,
    learning_rate=0.05,
    use_label_encoder=False
)

# This final 'churn_pipeline' object is what you will save
churn_pipeline = Pipeline(steps=[
    ('preprocessor', preprocessor),
    ('model', model)
])

# --- 6. Train ---
# X_train, y_train are your features and target from a point-in-time snapshot
churn_pipeline.fit(X_train, y_train)

# --- 7. Evaluate ---
# X_test, y_test must be from a *later* time period than training data
from sklearn.metrics import classification_report, precision_recall_curve, auc

y_probs = churn_pipeline.predict_proba(X_test)[:, 1]
precision, recall, _ = precision_recall_curve(y_test, y_probs)
print(f"Model AUPRC: {auc(recall, precision)}")
print(classification_report(y_test, churn_pipeline.predict(X_test)))

Validación del modelo: División basada en el tiempo

No puedes utilizar una función de división estándar train_test_splitmezcla aleatoria. Esto mezcla datos de todos los períodos de forma aleatoria, lo que provoca una fuga de información y crea una puntuación de rendimiento artificialmente optimista.

Debe validar en un conjunto de datos de prueba independiente del futuro..

• Estrategia: Capacitar entre enero y marzo, validar en abril. O capacitar en 2023, validar en el Q1 de 2024.
• Métricas: Centrarse en Precisión, Recall (Sensibilidad), Puntuación F1, y AUPRC (Área bajo la curva de precisión-recall). La exactitud es irrelevante. El negocio quiere saber: "De los 100 usuarios que predijeron que abandonarían (Precisión), ¿cuántos realmente lo hicieron? (Recall)".

Fase 3: Implementación y MLOps

Un artefacto de modelo entrenado (..pklo.joblib) por sí solo es inútil. Debe integrarse en un sistema para la formación, el control de versiones y la prestación de servicios.

Servicios de Ingeniería de Productos

Trabaje con nuestros Gerentes de Proyecto, Ingenieros de Software y Pruebadores de Calidad para desarrollar su nuevo producto de software personalizado o para apoyar su flujo de trabajo actual, siguiendo metodologías Agile, DevOps y Lean.

Build with 4Geeks

Registro de modelos con MLflow

Un repositorio de modelos es tu "Git para modelos". MLflow es el estándar de código abierto.

En su script de entrenamiento, debe registrar el modelo y sus metadatos:

import mlflow
import mlflow.sklearn
from sklearn.metrics import f1_score, precision_score, recall_score, roc_auc_score

# Set up MLflow tracking (can be a local folder or a remote server)
mlflow.set_tracking_uri("http://your-mlflow-server:5000")
mlflow.set_experiment("churn_prediction_v2")

# ... (your training code from above) ...

with mlflow.start_run() as run:
    # --- 1. Log parameters ---
    params = {
        "model_type": "XGBClassifier",
        "n_estimators": 200,
        "learning_rate": 0.05,
        "scale_pos_weight": scale_pos_weight
    }
    mlflow.log_params(params)

    # --- 2. Train the pipeline ---
    churn_pipeline.fit(X_train, y_train)

    # --- 3. Log metrics ---
    y_pred = churn_pipeline.predict(X_test)
    y_probs = churn_pipeline.predict_proba(X_test)[:, 1]
    
    metrics = {
        "f1_score": f1_score(y_test, y_pred),
        "precision": precision_score(y_test, y_pred),
        "recall": recall_score(y_test, y_pred),
        "auprc": auc(recall, precision) # From precision_recall_curve
    }
    mlflow.log_metrics(metrics)

    # --- 4. Log the model artifact ---
    # This logs the *entire* Scikit-learn pipeline
    mlflow.sklearn.log_model(
        sk_model=churn_pipeline,
        artifact_path="model",
        registered_model_name="production_churn_model" # Registers the model
    )
    
    print(f"Run ID: {run.info.run_id} logged to MLflow.")

Desde la interfaz de usuario de MLflow, ahora puede ver todas las ejecuciones de experimentos y, crucialmente, promover una versión del modelo desde "Staging" a "Producción."

Inferencia: API en tiempo real con FastAPI

Para la inferencia en tiempo real, un ligero marco de trabajo web en Python es ideal. FastAPI es el estándar moderno debido a su velocidad y validación automática de datos con Pydantic.

1. Crea tu archivo de API (main.py):

from fastapi import FastAPI
from pydantic import BaseModel
import pandas as pd
import mlflow
import os

# --- 1. Define the input data schema using Pydantic ---
# This provides automatic data validation
class UserFeatures(BaseModel):
    days_since_last_login: int
    session_length_avg_90d: float
    support_tickets_opened_30d: int
    plan_type: str
    user_region: str
    
    # Example for Pydantic v2
    class Config:
        extra = 'allow' # Allows other features not explicitly defined

# --- 2. Load the production model from MLflow ---
# Set the tracking URI
os.environ["MLFLOW_TRACKING_URI"] = "http://your-mlflow-server:5000"

# Load the model version currently tagged as "Production"
model_uri = "models:/production_churn_model/Production"
model = mlflow.pyfunc.load_model(model_uri)

app = FastAPI(title="Churn Prediction API")

@app.post("/predict")
def predict_churn(features: UserFeatures):
    """
    Takes user features as JSON and returns a churn probability.
    """
    # 1. Convert Pydantic model to pandas DataFrame
    # The Scikit-learn pipeline expects a DataFrame
    input_df = pd.DataFrame([features.model_dump()])
    
    # 2. Make prediction
    # The loaded model is a pyfunc wrapper, which matches mlflow.pyfunc.predict signature
    # This automatically handles preprocessing and prediction
    try:
        probability = model.predict(input_df)
        
        # The raw output from an XGB pipeline might be an array
        churn_probability = float(probability[0])
        
        return {
            "user_id": features.user_id, # Assuming user_id is passed in
            "churn_probability": churn_probability,
            "model_version": model.metadata.run_id # For traceability
        }
    except Exception as e:
        return {"error": str(e)}, 500

@app.get("/health")
def health_check():
    return {"status": "ok"}

2. Contenerizar y desplegar:

Esta aplicación de FastAPI puede ser contenedorizada con un simple Dockerfile y desplegada en cualquier plataforma moderna (Kubernetes, AWS ECS, Google Cloud Run) para una API escalable y de baja latencia.

Fase 4: Monitoreo y Explicabilidad

Tu trabajo no ha terminado con la implementación. Los modelos se degradan.

Deriva de los datos frente a deriva del modelo

Debe supervisar dos tipos de deriva:

1. Desplazamiento de Datos (Desplazamiento de Características): Las propiedades estadísticas de tus características de entrada cambian.
   • Ejemplo: Una nueva campaña de marketing atrae a usuarios de un nuevo país. La user_region ahora tiene una distribución que tu modelo nunca ha visto.
   • Cómo Monitorizar: Utiliza una biblioteca como evidently.ai o whylogs. Compara la distribución estadística (por ejemplo, media, mediana, cardinalidad) de los datos de inferencia entrantes con el conjunto de entrenamiento base. Activa una alerta si el Índice de Estabilidad Poblacional (PSI) o la prueba Kolmogorov-Smirnov (KS) tiene un valor p que supera un umbral.
2. Desplazamiento del Modelo (Desplazamiento de Concepto): La relación entre las características y la variable objetivo cambia.
   • Ejemplo: Un competidor lanza una nueva característica. Ahora, los usuarios de alta actividad (que anteriormente eran "seguros") comienzan a abandonar el servicio para usar esa característica. Tu modelo, entrenado con datos antiguos, ya no es preciso.
   • Cómo Monitorizar: Debes tener un bucle de retroalimentación. Registra todas las predicciones. Cuando se conozca el evento real (o la falta de) 30 días después, compáralo con tu predicción. Realiza un seguimiento de tu métrica AUPRC a lo largo del tiempo. Si disminuye en más del 10%, activa una alerta para volver a entrenar.

Explicabilidad (XAI)

La empresa no confiará en una "caja negra". Debes poder responder a la siguiente pregunta:¿Por qué se predice que este usuario abandonará el servicio?

Utilice SHAP (SHapley Additive exPlanations). Es una biblioteca sin restricciones que asigna un valor de "impacto" a cada característica para una predicción específica.

import shap

# ... (Load your 'churn_pipeline' and 'X_test') ...

# 1. Get the 'model' part of your pipeline
model = churn_pipeline.named_steps['model']

# 2. Get the *processed* data from the 'preprocessor' part
processed_X_test = pd.DataFrame(
    churn_pipeline.named_steps['preprocessor'].transform(X_test),
    columns=churn_pipeline.named_steps['preprocessor'].get_feature_names_out()
)

# 3. Create a SHAP explainer
explainer = shap.TreeExplainer(model)
shap_values = explainer.shap_values(processed_X_test)

# 4. Explain a single prediction (e.g., for the first user in the test set)
# This force plot shows which features pushed the prediction
# from the base value (average) to the final output
shap.initjs()
shap.force_plot(explainer.expected_value, shap_values[0,:], processed_X_test.iloc[0,:])

Este análisis se puede integrar en sus paneles internos, proporcionando al equipo de "Customer Success" puntos de conversación concretos y prácticos (por ejemplo: "La puntuación de este usuario es alta porque el número de Días desde la última conexióndays_since_last_login

Servicios de Ingeniería de Productos

Trabaje con nuestros gestores de proyectos, ingenieros de software y probadores de calidad internos para crear su nuevo producto de software personalizado o para apoyar su flujo de trabajo actual, siguiendo metodologías Agile, DevOps y Lean.

Build with 4Geeks

Conclusión

Construir un modelo de abandono de clientes es una tarea fundamental de ingeniería con IA. El valor no reside en un único .pkl archivo, sino en la creación de un sistema duradero y automatizado. El éxito se mide no por el AUPRC del modelo, sino por la capacidad de la empresa para utilizar sus resultados para tomar medidas.

Al enfocarse en una arquitectura sólida, un ingenieramiento de características riguroso y operaciones continuas de MLOps, se pasa de un proyecto reactivo de "ciencia de datos" a un sistema de ingeniería proactivo y orientado al valor que impacta directamente en los resultados financieros de la empresa.

Preguntas frecuentes