How to use of anogan ??

[edwardcho]

Hello Sir,

I have interesting on Anomaly Detection Task. My task is to detect anomaly on Image.

So I tried to find anogan model and I got some source from another github-site. Surprisingly, PyOD already has anogan. (But I think that PyOD's input type is feature vector-based, but anogan's input type is image-based)

If you don't mind, please tell me sample code about anogan on PyOD. The input type of pyOD is image-based, Right??

Thanks, Edward Cho.

[Shan2L]

I have the same problem.

[yzhao062]

maybe @mbongaerts can shed some light

[mbongaerts]

Dear @edwardcho and @GosTraight2020 ,

AnoGAN as implemented in PyOD takes only vector-based input since as far as I am aware this is the default of PyOD (@yzhao062 ?). It is true that according to the original paper AnoGAN takes 2D based (i.e. image) input. However, the AnoGAN framework can equally be implemented with vector-based input (as done in PyOD).

One way to use 'image-based input' and the pyOD AnoGAN implementation would be to just flatten your image into a 1D array and feed that into the AnoGAN method (see link ). My expectation is that this should work as well; Just make sure you play around with G_layers and D_layers. Also you might want to define another activation function for your generator (see link) because in images often pixels take values between 0 and 255. For example use:

def custom_activation(x):
    return tf.sigmoid(x) * 255

then use that function as

clf = AnoGAN(output_activation=custom_activation)

Of course, when working with images you might want to use convolutional neural network layers to extract features. In that case I believe you have two options: 1) either you find another implementation of AnoGAN online that takes images as input (and uses CNNs) or 2) you write your own 'wrapper' around the AnoGAN implementation of pyOD that uses also CNN layers before the AnoGAN is applied. The latter takes maybe a bit more time.

I hope this answers your question.

[Shan2L]

Thanks for your answer, I think maybe I should seek for another implementation of AnoGAN.

[mbongaerts]

Dear @GosTraight2020 and @edwardcho,

Not sure if you are still interested but I made an example code for the digit dataset using AnoGAN. Overall I see that the pyOD AnoGAN implementation works quite well for this dataset.

import seaborn as sns
import matplotlib.pyplot as plt
import random
import pandas as pd
import numpy as np

import tensorflow as tf
from tensorflow.keras.models import Model
from tensorflow.keras.layers import (Input, Dense, Dropout)
from tensorflow.keras.optimizers import Adam

from sklearn.datasets import fetch_openml
from sklearn.utils import check_random_state
from sklearn.model_selection import train_test_split

from pyod.models.anogan import AnoGAN

# Load data from https://www.openml.org/d/554
X, y = fetch_openml("mnist_784", version=1, return_X_y=True, as_frame=False)

# Scale all numbers between 0 and 1
X = X/X.max()

random_state = check_random_state(0)
permutation = random_state.permutation(X.shape[0])
X = X[permutation]
y = y[permutation]
X = X.reshape((X.shape[0], -1))

X_train, X_test, y_train, y_test = train_test_split(X, y, train_size=20000, test_size=10000,random_state=1)

# We take the numbers 5 to be our one-class
TF = y_train =='5'
X_train_learned = X_train[TF]
y_train_learned = y_train[TF]

# Test set for inliers (i.e. the number 5)
TF = y_test =='5'
X_test_1 = X_test[TF]
y_test_1 = y_test[TF]

# Test set for outliers
TF = y_test !='5'
X_test_2 = X_test[TF]
y_test_2 = y_test[TF]

def custom_activation(x):
    return tf.sigmoid(x) 

# train AnoGAN detector
clf_name = 'AnoGAN'
clf = AnoGAN(activation_hidden='tanh',
             dropout_rate=0.1,
             latent_dim_G=2,
             G_layers=[50,100,500],
             verbose=1,
             D_layers=[250,75,25],
             index_D_layer_for_recon_error=1,
             epochs=2, # We make an custom train loop later
             preprocessing=False,
             learning_rate= 0.000075,
             learning_rate_query=0.05,
             epochs_query=200,
             batch_size=100,
             output_activation=custom_activation,
             contamination=0.1)

# Just initialize model
clf.fit(X_train_learned[0:1])

# Custom train loop
for n in range(45000):
    if ((n % 100 == 0) and (n != 0) and (clf.verbose == 1)):
        print('Train iter:{}'.format(n))

    # Shuffle train 
    ind = random.sample( list(range(len(X_train_learned))), len(X_train_learned))[0:clf.batch_size]

    X_train_sel = X_train_learned[ind,:]
    latent_noise = np.random.normal(0, 1, (X_train_sel.shape[0], clf.latent_dim_G))

    clf.train_step((np.float32(X_train_sel), np.float32(latent_noise)))

clf.plot_learning_curves( start_ind=0,  window_smoothening=200)

Now we will generate some 'new numbers 5'

z_gamma = np.random.normal(0,1, size = (16,2))
generated_numbers = clf.generator({'I1': z_gamma}, training=False).numpy()

fig = plt.figure(figsize= (10,10))
for i in range(16):
    ax = fig.add_subplot(4,4,1+i)
    sns.heatmap( generated_numbers[i,:].reshape((28,28)) )
    ax.xaxis.set_visible(False)
    ax.yaxis.set_visible(False)

We tweak the fit_query() function a little bit for displaying purposes

def fit_query(query_sample):    
    # Make pseudo input (just zeros)
    zeros = np.zeros((1, clf.latent_dim_G))

    # build model for back-propagating a approximate latent space where
    # reconstruction with query sample is optimal
    pseudo_in = Input(shape=(clf.latent_dim_G,), name='I1')
    z_gamma = Dense(clf.latent_dim_G, activation=None, use_bias=True)(pseudo_in)

    sample_gen = clf.generator({'I1': z_gamma}, training=False)
    _, sample_disc_latent = clf.discriminator({'I1': sample_gen}, training=False)

    query_model = Model(inputs=(pseudo_in), outputs=[z_gamma, sample_gen, sample_disc_latent])

    opt = Adam(learning_rate=clf.learning_rate_query)
    query_model.compile(optimizer=opt)

    ###############
    for i in range(clf.epochs_query):
        if ((i % 25 == 0) and (clf.verbose == 1)):
            print('iter:', i)

        with tf.GradientTape() as tape:

            z, sample_gen, sample_disc_latent = query_model({'I1': zeros}, training=True)
            _, sample_disc_latent_original = clf.discriminator({'I1': query_sample}, training=False)

            # Reconstruction loss generator
            abs_err = tf.keras.backend.abs(query_sample - sample_gen)
            loss_recon_gen = tf.keras.backend.mean( tf.keras.backend.mean(abs_err, axis=-1))

            # Reconstruction loss latent space of discrimator
            abs_err = tf.keras.backend.abs(sample_disc_latent_original - sample_disc_latent)
            loss_recon_disc = tf.keras.backend.mean(tf.keras.backend.mean(abs_err, axis=-1))
            total_loss = loss_recon_gen + loss_recon_disc  # equal weighting both terms

        # Compute gradients
        gradients = tape.gradient(total_loss,
                                  query_model.trainable_variables[
                                  0:2])
        # Update weights
        query_model.optimizer.apply_gradients(
            zip(gradients, query_model.trainable_variables[0:2]))

    return query_model

Find approximated samples for the test set having the number 5 (inliers)

fig = plt.figure(figsize= (10,25))

for i in range(5):
    query_sample = X_test_1[[i]]

    query_model = fit_query(query_sample)
    z_gamma = query_model( {'I1': np.zeros((1, clf.latent_dim_G))}, training=True)[0].numpy()
    print(z_gamma)
    sample_gen = clf.generator({'I1': z_gamma}, training=False)[0].numpy()

    ax = fig.add_subplot(10,3,(i*3)+1 )
    sns.heatmap( query_sample.reshape((28,28)) )
    ax.xaxis.set_visible(False)
    ax.yaxis.set_visible(False)
    ax.set_title('Query sample')

    ax = fig.add_subplot(10,3,(i*3)+2)
    sns.heatmap( sample_gen.reshape((28,28)) )
    ax.xaxis.set_visible(False)
    ax.yaxis.set_visible(False)
    ax.set_title('Generated approximation')

    ax = fig.add_subplot(10,3,(i*3)+3)
    sns.heatmap( query_sample.reshape((28,28)) - sample_gen.reshape((28,28)) )
    ax.xaxis.set_visible(False)
    ax.yaxis.set_visible(False)
    ax.set_title('Error')

Find approximated samples for the test set having another number than 5 (outliers)

fig = plt.figure(figsize= (10,25))

for i in range(5):
    query_sample = X_test_2[[i]]

    query_model = fit_query(query_sample)
    z_gamma = query_model( {'I1': np.zeros((1, clf.latent_dim_G))}, training=True)[0].numpy()
    print(z_gamma)
    sample_gen = clf.generator({'I1': z_gamma}, training=False)[0].numpy()

    ax = fig.add_subplot(10,3,(i*3)+1 )
    sns.heatmap( query_sample.reshape((28,28)) )
    ax.xaxis.set_visible(False)
    ax.yaxis.set_visible(False)
    ax.set_title('Query sample')

    ax = fig.add_subplot(10,3,(i*3)+2)
    sns.heatmap( sample_gen.reshape((28,28)) )
    ax.xaxis.set_visible(False)
    ax.yaxis.set_visible(False)
    ax.set_title('Generated approximation')

    ax = fig.add_subplot(10,3,(i*3)+3)
    sns.heatmap( query_sample.reshape((28,28)) - sample_gen.reshape((28,28)) )
    ax.xaxis.set_visible(False)
    ax.yaxis.set_visible(False)
    ax.set_title('Error')

Make ROC curve

# true inliers
N = 100
y_test_scores_1 = clf.decision_function( X_test_1[0:N]) 

# true outliers
y_test_scores_2 = clf.decision_function( X_test_2[0:N]) 

max_score = max( y_test_scores_1.max(), y_test_scores_2.max())
thresholds = np.linspace(0,max_score,1000)

TPR = [ sum(y_test_scores_2>=t)/len(y_test_scores_2) for t in thresholds]
FPR = [ sum(y_test_scores_1>=t)/len(y_test_scores_1) for t in thresholds]

fig = plt.figure(figsize = (8,8) )
ax = fig.add_subplot(1,1,1)
ax.plot(FPR,TPR)
ax.plot([0,1],[0,1])
ax.set_ylabel('TPR', fontsize=20)
ax.set_xlabel('FPR', fontsize=20)

[edwardcho]

Hello mbongaerts,

Thank you for your good help.

[Shan2L]

Hi, @mbongaerts. Thank you so much for the example. It helps me a lot!