Mantenimiento predictivo en fresadoras

01Problema

Una parada no planificada de una máquina herramienta cuesta mucho más que una intervención programada. El objetivo es predecir si una operación de fresado terminará en fallo a partir de variables de proceso, para intervenir antes.

Tipo de problema: clasificación binaria muy desbalanceada. Métrica principal: recall de la clase «fallo» con una precisión mínima aceptable, y área bajo la curva precisión-recall (PR-AUC), más informativa que el acierto cuando los positivos son pocos.

02Datos

Dataset AI4I 2020 Predictive Maintenance del UCI Machine Learning Repository: 10.000 registros sintéticos que reproducen datos industriales reales. Variables: tipo de producto (L, M, H según calidad), temperatura del aire y del proceso (K), velocidad de giro (rpm), par (Nm) y desgaste de la herramienta (min). La etiqueta Machine failure vale 1 cuando se produce alguno de los cinco modos de fallo (desgaste de herramienta, disipación de calor, potencia, sobreesfuerzo y aleatorio).

03Exploración

Qué mirar primero: la proporción de fallos (en torno al 3 %, así que un modelo que diga siempre «no falla» acertaría el 97 % y no serviría de nada), las distribuciones de cada variable por clase y las correlaciones. Temperatura del aire y del proceso están muy correlacionadas, y velocidad y par tienen una relación inversa.

import pandas as pd

df = pd.read_csv("data/ai4i2020.csv")
print(df["Machine failure"].mean())            # proporción de fallos
print(df.groupby("Machine failure").mean(numeric_only=True))
df.hist(figsize=(12, 8))

04Preprocesado

Evitar la fuga de información: las columnas de los cinco modos de fallo (TWF, HDF, PWF, OSF, RNF) describen el propio fallo, así que se eliminan de las entradas, igual que los identificadores. El tipo de producto se codifica con one-hot. Se añaden variables con significado físico: potencia mecánica, diferencia de temperaturas y desgaste por par.

import numpy as np

X = df.drop(columns=["UDI", "Product ID", "Machine failure", "TWF", "HDF", "PWF", "OSF", "RNF"])
X = pd.get_dummies(X, columns=["Type"])
X["potencia_W"] = X["Torque [Nm]"] * X["Rotational speed [rpm]"] * 2 * np.pi / 60
X["delta_T"] = X["Process temperature [K]"] - X["Air temperature [K]"]
X["desgaste_par"] = X["Tool wear [min]"] * X["Torque [Nm]"]
y = df["Machine failure"]

05Modelo

Línea base con regresión logística y, como modelo principal, un Random Forest con class_weight="balanced" para compensar el desbalance. Como alternativa, gradient boosting (XGBoost o LightGBM), que suele rendir mejor en datos tabulares.

06Entrenamiento

from sklearn.model_selection import train_test_split, StratifiedKFold, cross_val_score
from sklearn.ensemble import RandomForestClassifier

X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.2, stratify=y, random_state=42)
modelo = RandomForestClassifier(n_estimators=400, class_weight="balanced", random_state=42, n_jobs=-1)
cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
print(cross_val_score(modelo, X_tr, y_tr, cv=cv, scoring="average_precision"))
modelo.fit(X_tr, y_tr)

07Evaluación

Se evalúa en el conjunto de test reservado con PR-AUC, recall y precisión de la clase «fallo» y la matriz de confusión. Además se compara con la línea base para comprobar que el modelo aporta algo.

from sklearn.metrics import average_precision_score, classification_report

prob = modelo.predict_proba(X_te)[:, 1]
print(average_precision_score(y_te, prob))
print(classification_report(y_te, prob > 0.5))

08Optimización

Dos palancas. La primera, los hiperparámetros, con búsqueda aleatoria y validación cruzada estratificada. La segunda, y la que más importa en negocio, el umbral de decisión: se elige el que minimiza el coste esperado, dado lo que cuesta una parada no detectada frente a una revisión innecesaria.

coste_fn, coste_fp = 5000, 300          # € por fallo no detectado / por revisión innecesaria
umbrales = np.linspace(0.05, 0.95, 91)
coste = [coste_fn * ((prob < u) & (y_te == 1)).sum() + coste_fp * ((prob >= u) & (y_te == 0)).sum() for u in umbrales]
umbral = umbrales[int(np.argmin(coste))]

09Despliegue

El modelo se sirve con una API FastAPI dentro de Docker. Pydantic valida cada entrada con rangos físicos plausibles, de modo que una lectura imposible devuelve un error 422 en lugar de una predicción sin sentido.

from fastapi import FastAPI
from pydantic import BaseModel, Field
import joblib

app = FastAPI(title="Mantenimiento predictivo")
modelo = joblib.load("models/rf_ai4i.joblib")

class Lectura(BaseModel):
    tipo: str = Field(pattern="^[LMH]{{MAIN}}quot;)
    temp_aire_K: float = Field(ge=250, le=350)
    temp_proceso_K: float = Field(ge=250, le=380)
    rpm: float = Field(gt=0, le=5000)
    par_Nm: float = Field(ge=0, le=150)
    desgaste_min: float = Field(ge=0, le=400)

@app.post("/predict")
def predecir(l: Lectura):
    x = construir_features(l)          # mismas transformaciones que en entrenamiento
    p = float(modelo.predict_proba(x)[0, 1])
    return {"prob_fallo": p, "alerta": p >= UMBRAL}

10Resultados