Detección de Anomalías con Autoencoders Variacionales (VAE) en TensorFlow y PyTorch

💡 Introducción a la Detección de Anomalías con VAEs

La detección de anomalías es una tarea crucial en muchos campos, desde la ciberseguridad y el monitoreo de equipos hasta la detección de fraudes y el diagnóstico médico. Se trata de identificar puntos de datos que no se ajustan a un patrón esperado. Tradicionalmente, esto se ha logrado con métodos estadísticos o de machine learning basados en reglas. Sin embargo, cuando los datos son complejos, de alta dimensionalidad o no se ajustan a distribuciones conocidas, los Autoencoders Variacionales (VAE) emergen como una solución poderosa y flexible.

Los VAEs son un tipo de red neuronal generativa que aprende una representación latente comprimida de los datos de entrada. A diferencia de los autoencoders tradicionales, los VAEs introducen una restricción probabilística en el espacio latente, lo que les permite generar nuevas muestras similares a los datos de entrenamiento y, lo que es más importante para nuestro propósito, cuantificar la "normalidad" de una muestra basándose en su capacidad para ser reconstruida o su probabilidad dentro del espacio latente.

En este tutorial, exploraremos cómo construir y entrenar VAEs utilizando dos de los frameworks de aprendizaje profundo más populares: TensorFlow y PyTorch. Cubriremos los fundamentos teóricos, la arquitectura, el proceso de entrenamiento y, lo más importante, cómo aplicar estos modelos para la detección efectiva de anomalías.

¿Por qué VAEs para la Detección de Anomalías? 🤔

Los VAEs son particularmente adecuados para la detección de anomalías por varias razones:

• Modelado Generativo: Aprenden la distribución de los datos "normales", lo que permite identificar lo que se desvía de esa distribución.
• Espacio Latente Continuo: La naturaleza probabilística del espacio latente asegura una interpolación suave y una mejor generalización.
• Cuantificación de la Anormalidad: Podemos utilizar métricas como el error de reconstrucción o la probabilidad en el espacio latente para asignar una "puntuación de anomalía" a cada punto de datos.
• No Supervisado: Generalmente, la detección de anomalías se realiza en escenarios donde los datos anómalos son raros o no están etiquetados, lo que hace que los métodos no supervisados como los VAEs sean ideales.

🛠️ Fundamentos Teóricos del Autoencoder Variacional (VAE)

Antes de sumergirnos en el código, es crucial comprender la teoría detrás de los VAEs. Un VAE se compone principalmente de dos partes: un encoder (codificador) y un decoder (decodificador).

El Encoder (Codificador) 🔍

El encoder toma un dato de entrada x y lo mapea a los parámetros de una distribución de probabilidad en el espacio latente, típicamente una distribución Gaussiana. Esto significa que, en lugar de mapear x a un único vector latente z, el encoder produce un vector de medias μ (mu) y un vector de desviaciones estándar σ (sigma) para cada dimensión latente. Generalmente, trabajamos con log-varianzas log(σ^2) para garantizar que las varianzas sean positivas y por estabilidad numérica.

q(z|x): Esta es la distribución posterior aproximada de la variable latente z dado el dato de entrada x. Nuestro encoder modela esta distribución.

El Decoder (Decodificador) 🖌️

El decoder toma una muestra del espacio latente z (que obtenemos muestreando la distribución parametrizada por μ y σ) y trata de reconstruir el dato de entrada original x. En esencia, el decoder aprende a generar datos a partir de la representación latente.

p(x|z): Esta es la distribución de probabilidad de los datos x dados un vector latente z. Nuestro decoder modela esta distribución.

La Función de Pérdida (Loss Function) 📉

La función de pérdida de un VAE es una combinación de dos términos importantes:

1. Pérdida de Reconstrucción (Reconstruction Loss): Mide qué tan bien el decoder reconstruye la entrada original. Para datos continuos, a menudo se usa el error cuadrático medio (MSE) o la entropía cruzada binaria (BCE) para datos binarios (como imágenes en blanco y negro). L_reconstruccion = -E_q(z|x) [log p(x|z)]
2. Divergencia KL (KL Divergence Loss): Este término mide la diferencia entre la distribución posterior aproximada q(z|x) (generada por el encoder) y una distribución prior (a priori) sobre el espacio latente p(z). Comúnmente, la prior p(z) se asume como una distribución Gaussiana estándar N(0, I). Este término actúa como un regularizador, forzando al espacio latente a ser bien comportado y continuo. L_KL = D_KL (q(z|x) || p(z))

La función de pérdida total es la suma de estos dos términos: L_VAE = L_reconstruccion + L_KL

El Truco de la Reparametrización (Reparameterization Trick) ✨

Para poder usar descenso de gradiente, necesitamos que el proceso de muestreo sea diferenciable. Sin embargo, muestrear directamente de q(z|x) no es diferenciable. El truco de la reparametrización resuelve esto: en lugar de muestrear z directamente, muestreamos una variable aleatoria ε (épsilon) de una distribución simple (como N(0, 1)) y luego calculamos z como:

z = μ + σ * ε

Donde μ y σ son las salidas del encoder. Esto mueve la aleatoriedad fuera de la parte diferenciable de la red, permitiendo el cálculo de gradientes.

🧑‍💻 Implementación en TensorFlow

Comenzaremos implementando el VAE en TensorFlow 2.x, utilizando la API funcional de Keras para mayor flexibilidad.

📦 Preparación del Entorno y Datos

Primero, asegurémonos de tener TensorFlow instalado. Utilizaremos el conjunto de datos MNIST para este ejemplo, ya que es simple y bueno para ilustrar el concepto. Sin embargo, los VAEs son aplicables a cualquier tipo de datos.

import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import numpy as np
import matplotlib.pyplot as plt

# Cargar y preprocesar datos MNIST
(x_train, _), (x_test, _) = keras.datasets.mnist.load_data()
x_train = x_train.astype("float32") / 255.
x_test = x_test.astype("float32") / 255.

# Aplanar imágenes para un VAE completamente conectado (FC)
x_train = x_train.reshape((len(x_train), np.prod(x_train.shape[1:])))
x_test = x_test.reshape((len(x_test), np.prod(x_test.shape[1:])))

print(f"Shape de los datos de entrenamiento: {x_train.shape}")
print(f"Shape de los datos de prueba: {x_test.shape}")

🏗️ Construcción del Modelo VAE en Keras

Definiremos una clase VAE que hereda de keras.Model. Esto nos permitirá personalizar el train_step y encapsular la lógica del VAE.

class Sampling(layers.Layer):
    """Utiliza (z_mean, z_log_var) para muestrear z, el vector latente."""
    def call(self, inputs):
        z_mean, z_log_var = inputs
        batch = tf.shape(z_mean)[0]
        dim = tf.shape(z_mean)[1]
        epsilon = tf.keras.backend.random_normal(shape=(batch, dim))
        return z_mean + tf.exp(0.5 * z_log_var) * epsilon


def build_vae_tensorflow(original_dim, latent_dim, intermediate_dim=256):
    # Encoder
    encoder_inputs = keras.Input(shape=(original_dim,))
    x = layers.Dense(intermediate_dim, activation="relu")(encoder_inputs)
    z_mean = layers.Dense(latent_dim, name="z_mean")(x)
    z_log_var = layers.Dense(latent_dim, name="z_log_var")(x)
    z = Sampling()([z_mean, z_log_var])
    encoder = keras.Model(encoder_inputs, [z_mean, z_log_var, z], name="encoder")

    # Decoder
    latent_inputs = keras.Input(shape=(latent_dim,))
    x = layers.Dense(intermediate_dim, activation="relu")(latent_inputs)
    decoder_outputs = layers.Dense(original_dim, activation="sigmoid")(x)
    decoder = keras.Model(latent_inputs, decoder_outputs, name="decoder")
    
    return encoder, decoder


class VAE(keras.Model):
    def __init__(self, encoder, decoder, original_dim, **kwargs):
        super(VAE, self).__init__(**kwargs)
        self.encoder = encoder
        self.decoder = decoder
        self.original_dim = original_dim
        self.total_loss_tracker = keras.metrics.Mean(name="total_loss")
        self.reconstruction_loss_tracker = keras.metrics.Mean(name="reconstruction_loss")
        self.kl_loss_tracker = keras.metrics.Mean(name="kl_loss")

    @property
    def metrics(self):
        return [
            self.total_loss_tracker,
            self.reconstruction_loss_tracker,
            self.kl_loss_tracker,
        ]

    def train_step(self, data):
        with tf.GradientTape() as tape:
            z_mean, z_log_var, z = self.encoder(data)
            reconstruction = self.decoder(z)
            
            reconstruction_loss = tf.reduce_mean(
                tf.reduce_sum(
                    keras.losses.binary_crossentropy(data, reconstruction),
                    axis=1
                )
            )
            kl_loss = -0.5 * (1 + z_log_var - tf.square(z_mean) - tf.exp(z_log_var))
            kl_loss = tf.reduce_mean(tf.reduce_sum(kl_loss, axis=1))
            total_loss = reconstruction_loss + kl_loss

        grads = tape.gradient(total_loss, self.trainable_weights)
        self.optimizer.apply_gradients(zip(grads, self.trainable_weights))
        self.total_loss_tracker.update_state(total_loss)
        self.reconstruction_loss_tracker.update_state(reconstruction_loss)
        self.kl_loss_tracker.update_state(kl_loss)
        return {
            "loss": self.total_loss_tracker.result(),
            "reconstruction_loss": self.reconstruction_loss_tracker.result(),
            "kl_loss": self.kl_loss_tracker.result(),
        }

# Parámetros del VAE
ORIGINAL_DIM = 784  # 28*28 para MNIST
LATENT_DIM = 2
INTERMEDIATE_DIM = 256

# Construir y compilar el VAE
encoder_tf, decoder_tf = build_vae_tensorflow(ORIGINAL_DIM, LATENT_DIM, INTERMEDIATE_DIM)
vae_tf = VAE(encoder_tf, decoder_tf, ORIGINAL_DIM)
vae_tf.compile(optimizer=keras.optimizers.Adam())

# Entrenar el VAE
print("\n--- Entrenamiento del VAE en TensorFlow ---")
vae_tf.fit(x_train, epochs=20, batch_size=64)

📊 Detección de Anomalías con VAE de TensorFlow

Una vez entrenado el VAE, podemos usar el error de reconstrucción para detectar anomalías. Los datos "normales" deberían ser reconstruidos con un error bajo, mientras que los datos anómalos, al no ajustarse a la distribución aprendida, tendrán un error de reconstrucción alto.

# Calcular errores de reconstrucción en el conjunto de prueba
x_test_encoded_mean, _, _ = vae_tf.encoder.predict(x_test)
x_test_reconstructed = vae_tf.decoder.predict(x_test_encoded_mean)

reconstruction_errors_tf = np.mean(np.square(x_test - x_test_reconstructed), axis=1)

# Visualizar la distribución de errores de reconstrucción
plt.figure(figsize=(10, 6))
plt.hist(reconstruction_errors_tf, bins=50, density=True, alpha=0.7)
plt.title('Distribución de Errores de Reconstrucción (TensorFlow)')
plt.xlabel('Error de Reconstrucción')
plt.ylabel('Densidad')
plt.grid(True)
plt.show()

# Establecer un umbral para la detección de anomalías
# Una estrategia común es usar la media + k * desviaciones estándar
threshold_tf = np.mean(reconstruction_errors_tf) + 2 * np.std(reconstruction_errors_tf)
print(f"Umbral de anomalía (TensorFlow): {threshold_tf:.4f}")

anomalies_tf = x_test[reconstruction_errors_tf > threshold_tf]
normal_data_tf = x_test[reconstruction_errors_tf <= threshold_tf]

print(f"Número de anomalías detectadas (TensorFlow): {len(anomalies_tf)}")
print(f"Número de datos normales (TensorFlow): {len(normal_data_tf)}")

# Visualizar algunas anomalías y datos normales
def plot_digits(data, title, num_digits=10):
    plt.figure(figsize=(15, 3))
    plt.suptitle(title, fontsize=16)
    for i in range(num_digits):
        ax = plt.subplot(1, num_digits, i + 1)
        plt.imshow(data[i].reshape(28, 28), cmap="gray")
        ax.get_xaxis().set_visible(False)
        ax.get_yaxis().set_visible(False)
    plt.show()

if len(anomalies_tf) > 0:
    plot_digits(anomalies_tf[:10], 'Ejemplos de Anomalías Detectadas (TensorFlow)')
else:
    print("No se detectaron anomalías con el umbral actual.")

plot_digits(normal_data_tf[:10], 'Ejemplos de Datos Normales Reconstruidos (TensorFlow)')

🧑‍💻 Implementación en PyTorch

Ahora, implementemos el mismo VAE utilizando PyTorch. Aunque la lógica es similar, la estructura del código y el manejo de los tensores varían.

📦 Preparación del Entorno y Datos

import torch
import torch.nn as nn
import torch.optim as optim
from torchvision import datasets, transforms
from torch.utils.data import DataLoader
import matplotlib.pyplot as plt
import numpy as np

# Configuración del dispositivo (GPU si está disponible, si no CPU)
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Usando dispositivo: {device}")

# Cargar y preprocesar datos MNIST
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Lambda(lambda x: x.view(-1)) # Aplanar imágenes
])

train_dataset = datasets.MNIST(root='./data', train=True, download=True, transform=transform)
test_dataset = datasets.MNIST(root='./data', train=False, download=True, transform=transform)

batch_size = 64
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)
test_loader = DataLoader(test_dataset, batch_size=batch_size, shuffle=False)

# Dimensiones
ORIGINAL_DIM_PT = 784
LATENT_DIM_PT = 2
INTERMEDIATE_DIM_PT = 256

🏗️ Construcción del Modelo VAE en PyTorch

En PyTorch, definimos el encoder y el decoder como módulos separados, y luego los combinamos en una clase VAE que hereda de nn.Module.

class VAE_PyTorch(nn.Module):
    def __init__(self, original_dim, latent_dim, intermediate_dim=256):
        super(VAE_PyTorch, self).__init__()

        # Encoder
        self.fc1 = nn.Linear(original_dim, intermediate_dim)
        self.fc2_mean = nn.Linear(intermediate_dim, latent_dim)
        self.fc2_log_var = nn.Linear(intermediate_dim, latent_dim)

        # Decoder
        self.fc3 = nn.Linear(latent_dim, intermediate_dim)
        self.fc4 = nn.Linear(intermediate_dim, original_dim)

        self.relu = nn.ReLU()
        self.sigmoid = nn.Sigmoid()

    def encode(self, x):
        h1 = self.relu(self.fc1(x))
        return self.fc2_mean(h1), self.fc2_log_var(h1)

    def reparameterize(self, mu, log_var):
        std = torch.exp(0.5 * log_var)
        eps = torch.randn_like(std)
        return mu + eps * std

    def decode(self, z):
        h3 = self.relu(self.fc3(z))
        return self.sigmoid(self.fc4(h3))

    def forward(self, x):
        mu, log_var = self.encode(x.view(-1, ORIGINAL_DIM_PT))
        z = self.reparameterize(mu, log_var)
        return self.decode(z), mu, log_var


def loss_function(recon_x, x, mu, log_var):
    # Pérdida de reconstrucción (Binary Cross-Entropy para MNIST)
    BCE = nn.functional.binary_cross_entropy(recon_x, x.view(-1, ORIGINAL_DIM_PT), reduction='sum')

    # Pérdida KL Divergence
    # 0.5 * sum(1 + log(sigma^2) - mu^2 - sigma^2)
    KLD = -0.5 * torch.sum(1 + log_var - mu.pow(2) - log_var.exp())

    return BCE + KLD

# Inicializar modelo y optimizador
vae_pt = VAE_PyTorch(ORIGINAL_DIM_PT, LATENT_DIM_PT, INTERMEDIATE_DIM_PT).to(device)
optimizer_pt = optim.Adam(vae_pt.parameters(), lr=1e-3)

# Función de entrenamiento
def train_pytorch(epoch):
    vae_pt.train()
    train_loss = 0
    for batch_idx, (data, _) in enumerate(train_loader):
        data = data.to(device)
        optimizer_pt.zero_grad()
        recon_batch, mu, log_var = vae_pt(data)
        loss = loss_function(recon_batch, data, mu, log_var)
        loss.backward()
        train_loss += loss.item()
        optimizer_pt.step()
    print(f'Epoch: {epoch} \tLoss: {train_loss / len(train_loader.dataset):.4f}')

# Entrenar el VAE
print("\n--- Entrenamiento del VAE en PyTorch ---")
epochs_pt = 20
for epoch in range(1, epochs_pt + 1):
    train_pytorch(epoch)

📊 Detección de Anomalías con VAE de PyTorch

De manera similar a TensorFlow, utilizaremos el error de reconstrucción para identificar anomalías.

# Calcular errores de reconstrucción en el conjunto de prueba
vae_pt.eval() # Poner el modelo en modo evaluación
reconstruction_errors_pt = []

with torch.no_grad():
    for data, _ in test_loader:
        data = data.to(device)
        recon_batch, mu, log_var = vae_pt(data)
        # Calcular MSE como error de reconstrucción
        recon_error = torch.mean(torch.square(data.view(-1, ORIGINAL_DIM_PT) - recon_batch), dim=1)
        reconstruction_errors_pt.extend(recon_error.cpu().numpy())

reconstruction_errors_pt = np.array(reconstruction_errors_pt)

# Visualizar la distribución de errores de reconstrucción
plt.figure(figsize=(10, 6))
plt.hist(reconstruction_errors_pt, bins=50, density=True, alpha=0.7)
plt.title('Distribución de Errores de Reconstrucción (PyTorch)')
plt.xlabel('Error de Reconstrucción')
plt.ylabel('Densidad')
plt.grid(True)
plt.show()

# Establecer un umbral para la detección de anomalías
threshold_pt = np.mean(reconstruction_errors_pt) + 2 * np.std(reconstruction_errors_pt)
print(f"Umbral de anomalía (PyTorch): {threshold_pt:.4f}")

anomalies_pt_indices = np.where(reconstruction_errors_pt > threshold_pt)[0]
normal_data_pt_indices = np.where(reconstruction_errors_pt <= threshold_pt)[0]

# Recopilar los datos originales para visualización
all_test_data = torch.cat([data for data, _ in test_loader], dim=0).view(-1, ORIGINAL_DIM_PT).cpu().numpy()

anomalies_pt = all_test_data[anomalies_pt_indices]
normal_data_pt = all_test_data[normal_data_pt_indices]

print(f"Número de anomalías detectadas (PyTorch): {len(anomalies_pt)}")
print(f"Número de datos normales (PyTorch): {len(normal_data_pt)}")

if len(anomalies_pt) > 0:
    plot_digits(anomalies_pt[:10], 'Ejemplos de Anomalías Detectadas (PyTorch)')
else:
    print("No se detectaron anomalías con el umbral actual.")

plot_digits(normal_data_pt[:10], 'Ejemplos de Datos Normales Reconstruidos (PyTorch)')

📈 Evaluación y Consideraciones Avanzadas

La detección de anomalías con VAEs no termina con la implementación básica. Hay varias consideraciones y técnicas avanzadas que pueden mejorar el rendimiento y la robustez de tu sistema.

Métricas de Evaluación para Detección de Anomalías 📊

A diferencia de las tareas de clasificación estándar, la detección de anomalías a menudo se enfrenta a conjuntos de datos altamente desequilibrados (las anomalías son raras). Por lo tanto, métricas como la precisión y el recall (o F1-score) son más informativas que la exactitud.

• Precisión (Precision): De todas las instancias predichas como anómalas, ¿cuántas son realmente anómalas? (TP / (TP + FP))
• Recall (Sensibilidad): De todas las instancias anómalas reales, ¿cuántas fueron detectadas? (TP / (TP + FN))
• F1-Score: Media armónica de precisión y recall.
• Curva ROC/AUC: Mide el rendimiento del clasificador en varios umbrales. Un AUC alto indica un buen rendimiento.
• Curva de Precisión-Recall (PR-Curve): Es a menudo más informativa que la curva ROC en conjuntos de datos desequilibrados.

# Suponiendo que tienes: 
# true_labels: array con 0 para normal, 1 para anómalo
# anomaly_scores: array con las puntuaciones de reconstrucción para cada instancia
# threshold: el umbral que elegiste

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

predicted_labels = (anomaly_scores > threshold).astype(int)

print(f"Precision: {precision_score(true_labels, predicted_labels):.4f}")
print(f"Recall: {recall_score(true_labels, predicted_labels):.4f}")
print(f"F1-Score: {f1_score(true_labels, predicted_labels):.4f}")
print(f"ROC AUC: {roc_auc_score(true_labels, anomaly_scores):.4f}")
print(f"Average Precision (PR-AUC): {average_precision_score(true_labels, anomaly_scores):.4f}")

Variaciones y Mejoras del VAE 🚀

• VAEs con Arquitecturas Convolucionales (CVAE): Para datos de imágenes, usar capas convolucionales en el encoder y decoder puede capturar mejor las características espaciales.
• VAEs Condicionales (CVAE): Permiten generar muestras condicionadas a ciertas etiquetas o atributos, lo que puede ser útil si se tiene alguna información parcial sobre el tipo de dato.
• Beta-VAE: Introduce un hiperparámetro β en la pérdida KL, permitiendo un mayor control sobre la independencia de las dimensiones latentes y la fidelidad de la reconstrucción.
• Semi-Supervised VAEs: Combinan datos etiquetados (cuando están disponibles) y no etiquetados para mejorar el rendimiento, especialmente si hay algunas anomalías conocidas.
• Normalización y Escalado de Datos: Es crucial escalar los datos de entrada adecuadamente (ej. a [0, 1] o z-score) para un entrenamiento estable del VAE.

Consideraciones Prácticas 🧐

1. Dimensión del Espacio Latente: Elegir la latent_dim adecuada es importante. Una dimensión muy pequeña puede no capturar toda la variabilidad de los datos normales, mientras que una muy grande podría permitir que el VAE reconstruya anomalías con facilidad, reduciendo su capacidad de detección.
2. Umbral de Anomalía: Como se mencionó, el umbral es clave. Podrías necesitar un conjunto de validación con algunas anomalías conocidas para optimizar este umbral. Un enfoque es usar un percentil alto (ej., el 99.5% de los errores de reconstrucción de datos normales) como umbral inicial.
3. Tamaño del Conjunto de Datos: Los VAEs, como otros modelos de aprendizaje profundo, se benefician de grandes volúmenes de datos. Asegúrate de tener suficientes datos "normales" para que el modelo aprenda una representación robusta.
4. Balance entre Reconstrucción y KL: El equilibrio entre la pérdida de reconstrucción y la pérdida KL puede ajustarse mediante un factor de ponderación. Esto es especialmente cierto en Beta-VAEs, donde se ajusta β para priorizar una representación latente más disentangled (independiente) o una mejor reconstrucción.

✅ Conclusión

Hemos cubierto los fundamentos teóricos y las implementaciones prácticas de los Autoencoders Variacionales (VAE) en TensorFlow y PyTorch para la detección de anomalías. Hemos visto cómo construir el encoder y el decoder, implementar el truco de la reparametrización y definir la función de pérdida. Además, hemos explorado cómo utilizar el error de reconstrucción para identificar puntos de datos anómalos y las consideraciones clave para evaluar y mejorar estos sistemas.

Los VAEs ofrecen una herramienta potente para abordar el desafío de la detección de anomalías en entornos complejos y no supervisados. Con este conocimiento, estás bien equipado para aplicar VAEs a tus propios conjuntos de datos y descubrir patrones ocultos de comportamiento inusual.

Próximos Pasos y Retos 🎯

• Experimenta con diferentes arquitecturas de encoder y decoder (ej. convolucionales para imágenes).
• Prueba con otros conjuntos de datos donde la detección de anomalías sea relevante (ej. series temporales, datos de sensores).
• Investiga las variaciones de VAEs como Beta-VAE o Conditional VAE para entender cómo modifican el espacio latente y la generación de muestras.
• Desarrolla una estrategia de despliegue para monitorear anomalías en tiempo real.

¡Sigue explorando el fascinante mundo del aprendizaje profundo y la detección de anomalías!