# -*- coding: utf-8 -*-
"""Using Auto Encoder with Outlier Detection
"""
# Author: Yue Zhao <zhaoy@cmu.edu>
# License: BSD 2 clause
from __future__ import division
from __future__ import print_function
import numpy as np
from sklearn.preprocessing import StandardScaler
from sklearn.utils import check_array
from sklearn.utils.validation import check_is_fitted
from .base import BaseDetector
from .base_dl import _get_tensorflow_version
from ..utils.stat_models import pairwise_distances_no_broadcast
from ..utils.utility import check_parameter
# if tensorflow 2, import from tf directly
if _get_tensorflow_version() < 200:
from keras.models import Sequential
from keras.layers import Dense, Dropout
from keras.regularizers import l2
from keras.losses import mean_squared_error
else:
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Dropout
from tensorflow.keras.regularizers import l2
from tensorflow.keras.losses import mean_squared_error
# noinspection PyUnresolvedReferences,PyPep8Naming,PyTypeChecker
[docs]
class AutoEncoder(BaseDetector):
"""Auto Encoder (AE) is a type of neural networks for learning useful data
representations unsupervisedly. Similar to PCA, AE could be used to
detect outlying objects in the data by calculating the reconstruction
errors. See :cite:`aggarwal2015outlier` Chapter 3 for details.
Parameters
----------
hidden_neurons : list, optional (default=[64, 32, 32, 64])
The number of neurons per hidden layers.
hidden_activation : str, optional (default='relu')
Activation function to use for hidden layers.
All hidden layers are forced to use the same type of activation.
See https://keras.io/activations/
output_activation : str, optional (default='sigmoid')
Activation function to use for output layer.
See https://keras.io/activations/
loss : str or obj, optional (default=keras.losses.mean_squared_error)
String (name of objective function) or objective function.
See https://keras.io/losses/
optimizer : str, optional (default='adam')
String (name of optimizer) or optimizer instance.
See https://keras.io/optimizers/
epochs : int, optional (default=100)
Number of epochs to train the model.
batch_size : int, optional (default=32)
Number of samples per gradient update.
dropout_rate : float in (0., 1), optional (default=0.2)
The dropout to be used across all layers.
l2_regularizer : float in (0., 1), optional (default=0.1)
The regularization strength of activity_regularizer
applied on each layer. By default, l2 regularizer is used. See
https://keras.io/regularizers/
validation_size : float in (0., 1), optional (default=0.1)
The percentage of data to be used for validation.
preprocessing : bool, optional (default=True)
If True, apply standardization on the data.
verbose : int, optional (default=1)
Verbosity mode.
- 0 = silent
- 1 = progress bar
- 2 = one line per epoch.
For verbose >= 1, model summary may be printed.
random_state : random_state: int, RandomState instance or None, optional
(default=None)
If int, random_state is the seed used by the random
number generator; If RandomState instance, random_state is the random
number generator; If None, the random number generator is the
RandomState instance used by `np.random`.
contamination : float in (0., 0.5), optional (default=0.1)
The amount of contamination of the data set, i.e.
the proportion of outliers in the data set. When fitting this is used
to define the threshold on the decision function.
Attributes
----------
encoding_dim_ : int
The number of neurons in the encoding layer.
compression_rate_ : float
The ratio between the original feature and
the number of neurons in the encoding layer.
model_ : Keras Object
The underlying AutoEncoder in Keras.
history_: Keras Object
The AutoEncoder training history.
decision_scores_ : numpy array of shape (n_samples,)
The outlier scores of the training data.
The higher, the more abnormal. Outliers tend to have higher
scores. This value is available once the detector is
fitted.
threshold_ : float
The threshold is based on ``contamination``. It is the
``n_samples * contamination`` most abnormal samples in
``decision_scores_``. The threshold is calculated for generating
binary outlier labels.
labels_ : int, either 0 or 1
The binary labels of the training data. 0 stands for inliers
and 1 for outliers/anomalies. It is generated by applying
``threshold_`` on ``decision_scores_``.
"""
def __init__(self, hidden_neurons=None,
hidden_activation='relu', output_activation='sigmoid',
loss=mean_squared_error, optimizer='adam',
epochs=100, batch_size=32, dropout_rate=0.2,
l2_regularizer=0.1, validation_size=0.1, preprocessing=True,
verbose=1, random_state=None, contamination=0.1):
super(AutoEncoder, self).__init__(contamination=contamination)
self.hidden_neurons = hidden_neurons
self.hidden_activation = hidden_activation
self.output_activation = output_activation
self.loss = loss
self.optimizer = optimizer
self.epochs = epochs
self.batch_size = batch_size
self.dropout_rate = dropout_rate
self.l2_regularizer = l2_regularizer
self.validation_size = validation_size
self.preprocessing = preprocessing
self.verbose = verbose
self.random_state = random_state
# default values
if self.hidden_neurons is None:
self.hidden_neurons = [64, 32, 32, 64]
# Verify the network design is valid
if not self.hidden_neurons == self.hidden_neurons[::-1]:
print(self.hidden_neurons)
raise ValueError("Hidden units should be symmetric")
self.hidden_neurons_ = self.hidden_neurons
check_parameter(dropout_rate, 0, 1, param_name='dropout_rate',
include_left=True)
def _build_model(self):
model = Sequential()
# Input layer
model.add(Dense(
self.hidden_neurons_[0], activation=self.hidden_activation,
input_shape=(self.n_features_,),
activity_regularizer=l2(self.l2_regularizer)))
model.add(Dropout(self.dropout_rate))
# Additional layers
for i, hidden_neurons in enumerate(self.hidden_neurons_, 1):
model.add(Dense(
hidden_neurons,
activation=self.hidden_activation,
activity_regularizer=l2(self.l2_regularizer)))
model.add(Dropout(self.dropout_rate))
# Output layers
model.add(Dense(self.n_features_, activation=self.output_activation,
activity_regularizer=l2(self.l2_regularizer)))
# Compile model
model.compile(loss=self.loss, optimizer=self.optimizer)
if self.verbose >= 1:
print(model.summary())
return model
# noinspection PyUnresolvedReferences
[docs]
def fit(self, X, y=None):
"""Fit detector. y is ignored in unsupervised methods.
Parameters
----------
X : numpy array of shape (n_samples, n_features)
The input samples.
y : Ignored
Not used, present for API consistency by convention.
Returns
-------
self : object
Fitted estimator.
"""
# validate inputs X and y (optional)
X = check_array(X)
self._set_n_classes(y)
# Verify and construct the hidden units
self.n_samples_, self.n_features_ = X.shape[0], X.shape[1]
# Standardize data for better performance
if self.preprocessing:
self.scaler_ = StandardScaler()
X_norm = self.scaler_.fit_transform(X)
else:
X_norm = np.copy(X)
# Shuffle the data for validation as Keras do not shuffling for
# Validation Split
np.random.shuffle(X_norm)
# Validate and complete the number of hidden neurons
if np.min(self.hidden_neurons) > self.n_features_:
raise ValueError("The number of neurons should not exceed "
"the number of features")
self.hidden_neurons_.insert(0, self.n_features_)
# Calculate the dimension of the encoding layer & compression rate
self.encoding_dim_ = np.median(self.hidden_neurons)
self.compression_rate_ = self.n_features_ // self.encoding_dim_
# Build AE model & fit with X
self.model_ = self._build_model()
self.history_ = self.model_.fit(X_norm, X_norm,
epochs=self.epochs,
batch_size=self.batch_size,
shuffle=True,
validation_split=self.validation_size,
verbose=self.verbose).history
# Reverse the operation for consistency
self.hidden_neurons_.pop(0)
# Predict on X itself and calculate the reconstruction error as
# the outlier scores. Noted X_norm was shuffled has to recreate
if self.preprocessing:
X_norm = self.scaler_.transform(X)
else:
X_norm = np.copy(X)
pred_scores = self.model_.predict(X_norm)
self.decision_scores_ = pairwise_distances_no_broadcast(X_norm,
pred_scores)
self._process_decision_scores()
return self
[docs]
def decision_function(self, X):
"""Predict raw anomaly score of X using the fitted detector.
The anomaly score of an input sample is computed based on different
detector algorithms. For consistency, outliers are assigned with
larger anomaly scores.
Parameters
----------
X : numpy array of shape (n_samples, n_features)
The training input samples. Sparse matrices are accepted only
if they are supported by the base estimator.
Returns
-------
anomaly_scores : numpy array of shape (n_samples,)
The anomaly score of the input samples.
"""
check_is_fitted(self, ['model_', 'history_'])
X = check_array(X)
if self.preprocessing:
X_norm = self.scaler_.transform(X)
else:
X_norm = np.copy(X)
# Predict on X and return the reconstruction errors
pred_scores = self.model_.predict(X_norm)
return pairwise_distances_no_broadcast(X_norm, pred_scores)