Cómo hacer mantenimiento predictivo con Python y Scikit-learn

En las operaciones industriales y de software modernas, el cambio del mantenimiento reactivo ("arreglar cuando falla") y preventivo ("arreglar cada N horas") al mantenimiento predictivo (PdM) representa una ventaja competitiva significativa. El PdM aprovecha el análisis de datos y el aprendizaje automático para detectar anomalías en el funcionamiento y predecir defectos o fallos antes de que ocurran.representa una ventaja competitiva significativa. PdM utiliza el análisis de datos y el aprendizaje automático para detectar anomalías en el funcionamiento y predecir fallos o averías.antesocurren.

Para un Director de Tecnología (CTO) o ingeniero de software, implementar una estrategia de Mantenimiento Predictivo (PdM) no es solo un ejercicio de ciencia de datos; es un desafío complejo de ingeniería de sistemas. Implica construir una robusta y escalable tubería de datos, seleccionar y validar modelos apropiados, e integrar la inferencia de modelos en los flujos de trabajo operativos.

Este artículo ofrece una guía práctica y basada en código para construir un modelo de clasificación PdM utilizando Python, Pandas y Scikit-learn. Nos centraremos en las decisiones de ingeniería, transformaciones de datos y patrones arquitectónicos necesarios para pasar de los datos brutos de los sensores a un modelo desplegable que pueda predecir un evento de fallo.

Equipo de Ingeniería de Software Compartido Bajo Demanda, Por Suscripción.

Acceda a un equipo flexible y compartido de ingeniería de software bajo demanda a través de una suscripción mensual predecible. Desarrolladores, diseñadores, ingenieros de control de calidad y un gerente de proyecto gratuito le ayudan a crear MVPs, escalar productos e innovar con tecnologías modernas como React, Node.js y más.

Try 4Geeks Teams

La arquitectura del sistema PdM

Antes de escribir cualquier código de aprendizaje automático, debemos diseñar el flujo de datos. Un sistema de PdM (Mantenimiento Predictivo) de nivel de producción es un desafío de MLOps que normalmente implica varios componentes:

1. Ingesta de datos: Los datos brutos (p. ej., vibración, temperatura, presión, registros de errores) se recopilan de los activos (p. ej., sensores IoT, registros de aplicaciones). Estos datos se transmiten a menudo a través de protocolos como MQTT o Kafka.
2. Almacenamiento y ETL de datos: Los datos se almacenan en un lago de datos (p. ej., S3, ADLS) o en una base de datos de series temporales (p. ej., InfluxDB). Una capa de procesamiento (p. ej., Spark, Dask o Pandas para conjuntos de datos más pequeños) limpia, agrega y fusiona los datos de los sensores con los registros de mantenimiento y los registros de fallas.
3. Ingeniería de características: Este es el paso más crítico. Los datos de series temporales brutos se transforman en características significativas (p. ej., promedios móviles, desviaciones estándar, características de retardo) que capturan el "estado" de un activo.
4. Entrenamiento y registro del modelo: Una tubería de entrenamiento ingiere las características, entrena uno o más modelos candidatos y los evalúa. El modelo final (p. ej., un archivo .pkl) y sus metadatos se almacenan en un registro de modelos (p. ej., MLflow, Vertex AI Registry).
5. Inferencia y despliegue:
   • Predicción por lotes: Un trabajo programado (p. ej., Airflow, Kubernetes CronJob) se ejecuta diariamente, evaluando todos los activos y generando una "lista de riesgos" para los equipos de mantenimiento.
   • Inferencia en tiempo real: Un punto final de API (p. ej., FastAPI, Flask) sirve el modelo, permitiendo que los sistemas en tiempo real verifiquen la salud de un activo bajo demanda.

Para esta guía, nos centraremos en los componentes principales: Ingeniería de Características, Entrenamiento del Modelo, e Inferencia, utilizando un conjunto de datos simulado.

Preparación y etiquetado de datos

Nuestro objetivo es predecir el fracaso. Esto requiere dos fuentes de datos principales:

1. Datos de sensores de series temporales: Lecturas continuas del activo.
2. Registros de fallas: Una lista discreta de eventos de fallas y sus marcas de tiempo.

La decisión de ingeniería más importante es definir la etiqueta. No estamos prediciendo el fallo en el momento exacto en que ocurre; lo estamos prediciendo con antelación. Debemos definir una ventana de predicción (por ejemplo, "¿Este activo fallará en las próximas 24 horas?").

Supongamos que tenemos sensors.csv y failures.csv. Utilizaremos Pandas para combinar y etiquetar nuestros datos.

import pandas as pd

# Load sample data
# sensors.csv: [timestamp, asset_id, sensor_1, sensor_2, sensor_3]
# failures.csv: [timestamp, asset_id, failure_type]
try:
    sensors = pd.read_csv('sensors.csv', parse_dates=['timestamp'])
    failures = pd.read_csv('failures.csv', parse_dates=['timestamp'])
except FileNotFoundError:
    print("Sample data files not found. Using placeholder dataframes.")
    # Create placeholder data for demonstration
    sensors = pd.DataFrame({
        'timestamp': pd.to_datetime(pd.date_range(start='2023-01-01', periods=1000, freq='H')),
        'asset_id': 1,
        'sensor_1': [100 + i/100 + (i//800)*50 for i in range(1000)],
        'sensor_2': [50 - i/200 + (i//800)*20 for i in range(1000)]
    })
    failures = pd.DataFrame({
        'timestamp': pd.to_datetime(['2023-02-12T15:00:00']),
        'asset_id': 1,
        'failure_type': 'Component B'
    })

# --- Label Engineering ---
# Define our prediction window: e.g., predict failure 24 hours in advance
prediction_window = pd.Timedelta('24 hours')

# Initialize labels to 0 (No Failure)
sensors['failure_imminent'] = 0

# Iterate over each failure event to create labels
# This is an O(n*m) operation, can be slow. 
# For large datasets, use window functions or merge_asof.
for _, fail_row in failures.iterrows():
    asset = fail_row['asset_id']
    fail_time = fail_row['timestamp']
    
    # Define the start of the "at-risk" window
    window_start = fail_time - prediction_window
    
    # Find all sensor readings for that asset within the window
    mask = (
        (sensors['asset_id'] == asset) &
        (sensors['timestamp'] >= window_start) &
        (sensors['timestamp'] < fail_time)
    )
    sensors.loc[mask, 'failure_imminent'] = 1

print(f"Total data points: {len(sensors)}")
print(f"Imminent failure points (Label=1): {sensors['failure_imminent'].sum()}")
print(sensors.tail())

Esta estrategia de etiquetado transforma el problema en una tarea de clasificación estándar binaria. Tenga en cuenta el desequilibrio de clases; tendremos muchas más 0 (muestras de "saludables") que 1 (muestras de "fallidas").

Ingeniería de características para datos de series temporales

Un modelo rara vez aprende de los valores brutos de los sensores por sí solo. Debemos crear características que describan el comportamiento de los sensores a lo largo del tiempo. Las estadísticas de ventana móvil son las características más efectivas para el mantenimiento predictivo.

Equipo de Ingeniería de Software Compartido Bajo Demanda, Por Suscripción.

Acceda a un equipo flexible y compartido de ingeniería de software bajo demanda a través de una suscripción mensual predecible. Desarrolladores, diseñadores, ingenieros de control de calidad y un gerente de proyecto gratuito le ayudan a crear MVPs, escalar productos e innovar con tecnologías modernas como React, Node.js y más.

Try 4Geeks Teams

Calcularemos la media, la desviación estándar y el valor máximo para cada sensor en diferentes ventanas de tiempo (por ejemplo, 6 horas, 12 horas).

import pandas as pd

# Assuming 'sensors' DataFrame from the previous step
# Set timestamp as index for easier time-based operations
sensors = sensors.set_index('timestamp').sort_index()

# Define sensor columns and window sizes (in hours)
sensor_cols = ['sensor_1', 'sensor_2']
window_sizes = ['6H', '12H', '24H']

# Group by asset to prevent windows from crossing over different assets
grouped = sensors.groupby('asset_id')

feature_dfs = [sensors] # Start with the original data

for window in window_sizes:
    print(f"Calculating features for window: {window}")
    
    # Calculate rolling mean
    rolling_mean = grouped[sensor_cols].rolling(window=window).mean()
    rolling_mean = rolling_mean.rename(columns={col: f'{col}_mean_{window}' for col in sensor_cols})
    
    # Calculate rolling standard deviation
    rolling_std = grouped[sensor_cols].rolling(window=window).std()
    rolling_std = rolling_std.rename(columns={col: f'{col}_std_{window}' for col in sensor_cols})

    # Calculate rolling max
    rolling_max = grouped[sensor_cols].rolling(window=window).max()
    rolling_max = rolling_max.rename(columns={col: f'{col}_max_{window}' for col in sensor_cols})
    
    feature_dfs.extend([rolling_mean, rolling_std, rolling_max])

# Combine all features
# We must reset index to align properly before concatenation
aligned_dfs = [df.reset_index() for df in feature_dfs]
features_df = pd.concat(aligned_dfs, axis=1)

# Remove duplicate columns ('timestamp', 'asset_id')
features_df = features_df.loc[:, ~features_df.columns.duplicated()]

# Drop rows with NaN values generated by the rolling windows
features_df = features_df.dropna().reset_index(drop=True)

print("Feature-engineered DataFrame shape:", features_df.shape)
print(features_df.columns)

Nuestro DataFrame ahora contiene no solo la lectura actual del sensor, sino también su historial reciente, proporcionando un contexto rico para el modelo.

Selección de modelo, entrenamiento y evaluación

Para datos tabulares y estructurados como este, los complejos modelos de Deep Learning (por ejemplo, las LSTM) a menudo son superados por losconjuntos basados en árboles como Random Forest o Gradient Boosting (XGBoost, LightGBM). Estos modelos son:

• Excelente rendimiento en datos tabulares.
• Robusto frente a valores atípicos y características no escaladas (aunque el escalado sigue siendo una buena práctica).
• Fácil de interpretar (podemos extraer la importancia de las características).

Conciencia de Series Temporales en la División

No podemos mezclar aleatoriamente los datos de series temporales para el entrenamiento y las pruebas. Esto causaría una fuga de datos, ya que el modelo estaría entrenado con datos del futuro y probado con datos del pasado. Debemos dividir nuestros datos de forma cronológica.fuga de datos, ya que el modelo se entrenaría con datos del futuro y se probaría con datos del pasado. Debemos dividir nuestros datos de forma cronológica.

Manejo de desequilibrio de clases

Dado que los errores son poco frecuentes, nuestro conjunto de datos está muy desequilibrado. Podemos solucionar esto utilizando:

1. Re-muestreo: Sobremuestreo de la clase minoritaria (por ejemplo, SMOTE) o submuestreo de la clase mayoritaria.
2. Ponderación a nivel de modelo: Utilizar el parámetro class_weight en los modelos de Scikit-learn para penalizar con mayor fuerza las clasificaciones incorrectas de la clase minoritaria.

Utilizaremos el parámetro class_weight='balanced', que es un enfoque sencillo y eficaz.

Plataforma de formación

Utilizaremos el objeto Pipeline de Scikit-learn para encadenar el preprocesamiento (escalado) y el modelo.

from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
from sklearn.pipeline import Pipeline
from sklearn.metrics import classification_report, confusion_matrix, PrecisionRecallDisplay
import matplotlib.pyplot as plt

# Define features (X) and target (y)
target = 'failure_imminent'
# Exclude non-feature columns
features = [col for col in features_df.columns if col not in [target, 'asset_id', 'timestamp']]

X = features_df[features]
y = features_df[target]

# --- Time-Series Split ---
# We split based on time, not randomly.
# Let's use the first 80% of data for training, 20% for testing.
split_index = int(len(X) * 0.8)

X_train, X_test = X.iloc[:split_index], X.iloc[split_index:]
y_train, y_test = y.iloc[:split_index], y.iloc[split_index:]

print(f"Train shapes: X={X_train.shape}, y={y_train.shape}")
print(f"Test shapes:  X={X_test.shape}, y={y_test.shape}")
print(f"Test set failure rate: {y_test.mean():.2%}")

# --- Build the Pipeline ---
# 1. StandardScaler: Scales features
# 2. RandomForestClassifier: Our model
#    class_weight='balanced' is CRITICAL for imbalanced data.
pipeline = Pipeline([
    ('scaler', StandardScaler()),
    ('model', RandomForestClassifier(
        n_estimators=100, 
        random_state=42, 
        class_weight='balanced',
        n_jobs=-1
    ))
])

# --- Train the Model ---
print("Training the model...")
pipeline.fit(X_train, y_train)

# --- Evaluate the Model ---
print("Evaluating the model...")
y_pred = pipeline.predict(X_test)

# --- Performance Metrics ---
# For PdM, accuracy is useless. We care about Precision and Recall.
# Precision: Of all "failure" predictions, how many were correct? (Minimizes false positives / unnecessary maintenance)
# Recall: Of all actual failures, how many did we catch? (Minimizes false negatives / catastrophic failures)
# We MUST optimize for HIGH RECALL.

print("\n--- Classification Report ---")
print(classification_report(y_test, y_pred))

print("\n--- Confusion Matrix ---")
# [[True Negative,  False Positive],
#  [False Negative, True Positive]]
cm = confusion_matrix(y_test, y_pred)
print(cm)

# Plot Precision-Recall Curve
fig, ax = plt.subplots(figsize=(8, 6))
PrecisionRecallDisplay.from_estimator(pipeline, X_test, y_test, ax=ax)
ax.set_title("Precision-Recall Curve")
plt.show()
# 

# --- Feature Importance ---
# Get feature importances from the model in the pipeline
importances = pipeline.named_steps['model'].feature_importances_
feature_importance_df = pd.DataFrame({
    'feature': features,
    'importance': importances
}).sort_values(by='importance', ascending=False)

print("\n--- Top 10 Features ---")
print(feature_importance_df.head(10))

El Informe de Clasificación y la Matriz de Confusión son la base de su evaluación. Su objetivo es maximizar el Número de Verdaderos Positivos y minimizar el Número de Falsos Negativos(errores no detectados). Es probable que tenga que aceptar algunos Falsos Positivos(mantenimiento innecesario) como compensación.

Implementación y puesta en marcha

Un objeto Pipeline entrenado encapsula toda tu lógica (escalado + modelo). Puede ser serializado y desplegado fácilmente.

Serialización de modelos

Utilizamosjoblib para guardar nuestro objeto de pipeline entrenado en el disco.

import joblib

# Save the entire pipeline
model_filename = 'pdm_model_pipeline.joblib'
joblib.dump(pipeline, model_filename)

print(f"Model saved to {model_filename}")

Crear una API de predicción (FastAPI)

Este modelo serializable se puede cargar en un servidor web ligero para crear una API de predicción. FastAPI es una excelente opción debido a su alto rendimiento y validación automática de datos con Pydantic.

Archivo: main.py

import joblib
import pandas as pd
from fastapi import FastAPI
from pydantic import BaseModel
from typing import List

# Define the input data structure
# This MUST match the features the model was trained on
class SensorFeatures(BaseModel):
    sensor_1: float
    sensor_2: float
    sensor_1_mean_6H: float
    sensor_1_std_6H: float
    sensor_1_max_6H: float
    sensor_1_mean_12H: float
    # ... (add all other 18 features)
    sensor_2_max_24H: float

# Create the FastAPI app
app = FastAPI(
    title="Predictive Maintenance API",
    description="API for predicting imminent asset failure"
)

# Load the model ONCE at startup
try:
    pipeline = joblib.load('pdm_model_pipeline.joblib')
    print("Model loaded successfully.")
except FileNotFoundError:
    print("Model file not found. API will not work.")
    pipeline = None

@app.on_event("startup")
async def load_model():
    global pipeline
    if pipeline is None:
        try:
            pipeline = joblib.load('pdm_model_pipeline.joblib')
            print("Model loaded successfully.")
        except FileNotFoundError:
            print("ERROR: Model file 'pdm_model_pipeline.joblib' not found.")

@app.post("/predict")
async def predict(features: SensorFeatures):
    """
    Predicts the probability of imminent failure based on sensor features.
    
    Receives a single observation of features and returns a prediction.
    """
    if pipeline is None:
        return {"error": "Model not loaded"}, 500
        
    # Convert Pydantic model to DataFrame
    # The model expects a 2D array (or DataFrame)
    data = pd.DataFrame([features.dict()])
    
    # Ensure column order matches training
    # (A more robust API would fetch feature names from the model object)
    
    try:
        # Get class probabilities
        probabilities = pipeline.predict_proba(data)
        
        # Probability of class 1 (failure)
        failure_probability = probabilities[0][1] 
        
        # Get class prediction
        prediction = int(pipeline.predict(data)[0]) # 0 or 1
        
        return {
            "prediction": prediction,
            "failure_probability": failure_probability
        }
    except Exception as e:
        return {"error": str(e)}, 400

Para ejecutar esta API: uvicorn main:app --reload

Este punto de acceso de API proporciona un mecanismo operativo para comprobar el estado de un activo. Un sistema de monitorización ahora puede enviar sus datos, elaborados, a POST /predict y recibir una puntuación de riesgo en tiempo real, lo que desencadena alertas o órdenes de trabajo.

Equipo de Ingeniería de Software Compartido Bajo Demanda, Por Suscripción.

Acceda a un equipo flexible de ingeniería de software compartida bajo demanda a través de una suscripción mensual predecible. Desarrolladores, diseñadores, ingenieros de control de calidad y un gerente de proyectos gratuito le ayudan a crear MVPs, escalar productos e innovar con tecnologías modernas como React, Node.js, y más.

Try 4Geeks Teams

Conclusión

Hemos desarrollado con éxito un modelo completo de mantenimiento predictivo, desde los datos brutos hasta una API desplegable.

Para directores de tecnología y líderes de ingeniería, los puntos clave son:

1. PdM es un problema de ingeniería de datos: El éxito se centra más en la ingesta de datos de alta calidad y estrategias sólidas de etiquetado, que en algoritmos complejos, y menos en el modelado.
2. Comience con modelos simples e interpretables: Un RandomForestClassifier con class_weight='balanced' y características bien diseñadas, es una base potente y lista para producción que es mucho más interpretable que una red neuronal.
3. Optimice para la métrica correcta: No utilice "precisión". Concéntrese en Recall y en la Precision-Recall, equilibrando el costo de un fallo no detectado (Falso Negativo) contra el costo del mantenimiento innecesario (Falso Positivo).
4. Diseñe para la implementación: Construya su modelo dentro de un Pipeline de Scikit-learn para encapsular el preprocesamiento. Esto hace que la serialización y la implementación a través de una API sencilla (como FastAPI) sean triviales, conectando su modelo con el resto de su pila de software operativa.

Preguntas frecuentes