Conv_AE.py

from tensorflow.keras.callbacks import EarlyStopping
from tensorflow.keras.layers import Conv1D, Conv1DTranspose, Dropout, Input
from tensorflow.keras.models import Sequential
from tensorflow.keras.optimizers import Adam


class Conv_AE:
    """
    A reconstruction convolutional autoencoder model to detect anomalies in timeseries data using reconstruction error as an anomaly score.

    Parameters
    ----------
    No parameters are required for initializing the class.

    Attributes
    ----------
    model : Sequential
        The trained convolutional autoencoder model.

    Examples
    --------
    >>> from Conv_AE import Conv_AE
    >>> CAutoencoder = Conv_AE()
    >>> CAutoencoder.fit(train_data)
    >>> prediction = CAutoencoder.predict(test_data)
    """

    def __init__(self):
        self._Random(0)

    def _Random(self, seed_value):
        import os

        os.environ["PYTHONHASHSEED"] = str(seed_value)

        import random

        random.seed(seed_value)

        import numpy as np

        np.random.seed(seed_value)

        import tensorflow as tf

        tf.random.set_seed(seed_value)

    def _build_model(self):
        model = Sequential(
            [
                Input(shape=(self.shape[1], self.shape[2])),
                Conv1D(
                    filters=32,
                    kernel_size=7,
                    padding="same",
                    strides=2,
                    activation="relu",
                ),
                Dropout(rate=0.2),
                Conv1D(
                    filters=16,
                    kernel_size=7,
                    padding="same",
                    strides=2,
                    activation="relu",
                ),
                Conv1DTranspose(
                    filters=16,
                    kernel_size=7,
                    padding="same",
                    strides=2,
                    activation="relu",
                ),
                Dropout(rate=0.2),
                Conv1DTranspose(
                    filters=32,
                    kernel_size=7,
                    padding="same",
                    strides=2,
                    activation="relu",
                ),
                Conv1DTranspose(filters=1, kernel_size=7, padding="same"),
            ]
        )
        model.compile(optimizer=Adam(learning_rate=0.001), loss="mse")

        return model

    def fit(self, data):
        """
        Train the convolutional autoencoder model on the provided data.

        Parameters
        ----------
        data : numpy.ndarray
            Input data for training the autoencoder model.
        """

        self.shape = data.shape
        self.model = self._build_model()

        self.model.fit(
            data,
            data,
            epochs=100,
            batch_size=32,
            validation_split=0.1,
            verbose=0,
            callbacks=[
                EarlyStopping(
                    monitor="val_loss", patience=5, mode="min", verbose=0
                )
            ],
        )

    def predict(self, data):
        """
        Generate predictions using the trained convolutional autoencoder model.

        Parameters
        ----------
        data : numpy.ndarray
            Input data for generating predictions.

        Returns
        -------
        numpy.ndarray
            Predicted output data.
        """

        return self.model.predict(data)