Source code for alibi_detect.od.aegmm

import logging
import numpy as np
import tensorflow as tf
from typing import Callable, Dict, Tuple
from alibi_detect.models.tensorflow.autoencoder import AEGMM, eucl_cosim_features
from alibi_detect.models.tensorflow.gmm import gmm_energy, gmm_params
from alibi_detect.models.tensorflow.losses import loss_aegmm
from alibi_detect.models.tensorflow.trainer import trainer
from alibi_detect.base import BaseDetector, FitMixin, ThresholdMixin, outlier_prediction_dict
from alibi_detect.utils.tensorflow.prediction import predict_batch
from alibi_detect.utils._types import OptimizerTF

logger = logging.getLogger(__name__)

[docs] class OutlierAEGMM(BaseDetector, FitMixin, ThresholdMixin):
[docs] def __init__(self, threshold: float = None, aegmm: tf.keras.Model = None, encoder_net: tf.keras.Model = None, decoder_net: tf.keras.Model = None, gmm_density_net: tf.keras.Model = None, n_gmm: int = None, recon_features: Callable = eucl_cosim_features, data_type: str = None ) -> None: """ AEGMM-based outlier detector. Parameters ---------- threshold Threshold used for outlier score to determine outliers. aegmm A trained tf.keras model if available. encoder_net Layers for the encoder wrapped in a tf.keras.Sequential class if no 'aegmm' is specified. decoder_net Layers for the decoder wrapped in a tf.keras.Sequential class if no 'aegmm' is specified. gmm_density_net Layers for the GMM network wrapped in a tf.keras.Sequential class. n_gmm Number of components in GMM. recon_features Function to extract features from the reconstructed instance by the decoder. data_type Optionally specifiy the data type (tabular, image or time-series). Added to metadata. """ super().__init__() if threshold is None: logger.warning('No threshold level set. Need to infer threshold using `infer_threshold`.') self.threshold = threshold # check if model can be loaded, otherwise initialize AEGMM model if isinstance(aegmm, tf.keras.Model): self.aegmm = aegmm elif (isinstance(encoder_net, tf.keras.Sequential) and isinstance(decoder_net, tf.keras.Sequential) and isinstance(gmm_density_net, tf.keras.Sequential)): self.aegmm = AEGMM(encoder_net, decoder_net, gmm_density_net, n_gmm, recon_features) else: raise TypeError('No valid format detected for `aegmm` (tf.keras.Model) ' 'or `encoder_net`, `decoder_net` and `gmm_density_net` (tf.keras.Sequential).') # set metadata self.meta['detector_type'] = 'outlier' self.meta['data_type'] = data_type self.meta['online'] = False self.phi,, self.cov, self.L, self.log_det_cov = None, None, None, None, None
[docs] def fit(self, X: np.ndarray, loss_fn: tf.keras.losses = loss_aegmm, w_energy: float = .1, w_cov_diag: float = .005, optimizer: OptimizerTF = tf.keras.optimizers.Adam, epochs: int = 20, batch_size: int = 64, verbose: bool = True, log_metric: Tuple[str, "tf.keras.metrics"] = None, callbacks: tf.keras.callbacks = None, ) -> None: """ Train AEGMM model. Parameters ---------- X Training batch. loss_fn Loss function used for training. w_energy Weight on sample energy loss term if default `loss_aegmm` loss fn is used. w_cov_diag Weight on covariance regularizing loss term if default `loss_aegmm` loss fn is used. optimizer Optimizer used for training. epochs Number of training epochs. batch_size Batch size used for training. verbose Whether to print training progress. log_metric Additional metrics whose progress will be displayed if verbose equals True. callbacks Callbacks used during training. """ # train arguments args = [self.aegmm, loss_fn, X] optimizer = optimizer() if isinstance(optimizer, type) else optimizer kwargs = {'optimizer': optimizer, 'epochs': epochs, 'batch_size': batch_size, 'verbose': verbose, 'log_metric': log_metric, 'callbacks': callbacks, 'loss_fn_kwargs': {'w_energy': w_energy, 'w_cov_diag': w_cov_diag} } # train trainer(*args, **kwargs) # set GMM parameters x_recon, z, gamma = self.aegmm(X) self.phi,, self.cov, self.L, self.log_det_cov = gmm_params(z, gamma)
[docs] def infer_threshold(self, X: np.ndarray, threshold_perc: float = 95., batch_size: int = int(1e10) ) -> None: """ Update threshold by a value inferred from the percentage of instances considered to be outliers in a sample of the dataset. Parameters ---------- X Batch of instances. threshold_perc Percentage of X considered to be normal based on the outlier score. batch_size Batch size used when making predictions with the AEGMM. """ # compute outlier scores iscore = self.score(X, batch_size=batch_size) # update threshold self.threshold = np.percentile(iscore, threshold_perc)
[docs] def score(self, X: np.ndarray, batch_size: int = int(1e10)) -> np.ndarray: """ Compute outlier scores. Parameters ---------- X Batch of instances to analyze. batch_size Batch size used when making predictions with the AEGMM. Returns ------- Array with outlier scores for each instance in the batch. """ _, z, _ = predict_batch(X, self.aegmm, batch_size=batch_size) energy, _ = gmm_energy(z, self.phi,, self.cov, self.L, self.log_det_cov, return_mean=False) return energy.numpy()
[docs] def predict(self, X: np.ndarray, batch_size: int = int(1e10), return_instance_score: bool = True) \ -> Dict[Dict[str, str], Dict[np.ndarray, np.ndarray]]: """ Compute outlier scores and transform into outlier predictions. Parameters ---------- X Batch of instances. batch_size Batch size used when making predictions with the AEGMM. return_instance_score Whether to return instance level outlier scores. Returns ------- Dictionary containing ``'meta'`` and ``'data'`` dictionaries. - ``'meta'`` has the model's metadata. - ``'data'`` contains the outlier predictions and instance level outlier scores. """ # compute outlier scores iscore = self.score(X, batch_size=batch_size) # values above threshold are outliers outlier_pred = (iscore > self.threshold).astype(int) # populate output dict od = outlier_prediction_dict() od['meta'] = self.meta od['data']['is_outlier'] = outlier_pred if return_instance_score: od['data']['instance_score'] = iscore return od