lemna.py

# -*- coding: utf-8 -*-
"""lemna.ipynb

Automatically generated by Colaboratory.

Original file is located at
    https://colab.research.google.com/drive/1v98GmeeGQyKPjCgmAlToyy-NBZSL3EmP
"""

# !pip install lime
from sklearn.metrics import accuracy_score, precision_score, confusion_matrix, mean_squared_error, recall_score, roc_curve
from sklearn.utils import shuffle, check_random_state
from sklearn.model_selection import train_test_split
from sklearn.mixture import GaussianMixture
from google.colab import drive, files
from keras.models import Sequential
from keras.layers import Dense
import pandas as pd 
import numpy as np 
import pickle
import lime
import lime.lime_tabular
import matplotlib.pyplot as plt

drive.mount('/content/drive')

ben  = pd.read_csv('/content/drive/My Drive/benign.csv')
mal  = pd.read_csv('/content/drive/My Drive/malicious.csv')
data = ben.append(mal, ignore_index=True)
data = shuffle(data)

y = data['class']
x = data.drop(['class'], axis=1)
x = x.drop(['filename'],axis=1)

x_train, x_test, y_train, y_test = train_test_split(x, y, test_size= 0.3, random_state=27)

feature_means = []
feature_maxs = []

def categorical_helper_fun(x):
    if isinstance(x, int) or isinstance(x,float):
        return int(x)
    else:
        return 0

print (feature_maxs)

for i in data:
    if True in data[i]:
        data[i] = list(map(lambda x: categorical_helper_fun(x), data[i]))
    feature_means.append(np.mean(data[i]))
    feature_maxs.append(max(data[i]))

feature_names = []
for i in data:
    feature_names.append(i)
print (feature_names)

l = lime.lime_tabular.LimeTabularExplainer(np.array(x_train), feature_names = feature_names, class_names=['True', 'False'], discretize_continuous=True)

def data_inverse(l, data_row, num_samples):
    """Generates a neighborhood around a prediction.

    For numerical features, perturb them by sampling from a Normal(0,1) and
    doing the inverse operation of mean-centering and scaling, according to
    the means and stds in the training data. For categorical features,
    perturb by sampling according to the training distribution, and making
    a binary feature that is 1 when the value is the same as the instance
    being explained.

    Args:
        data_row: 1d numpy array, corresponding to a row
        num_samples: size of the neighborhood to learn the linear model

    Returns:
        A tuple (data, inverse), where:
            data: dense num_samples * K matrix, where categorical features
            are encoded with either 0 (not equal to the corresponding value
            in data_row) or 1. The first row is the original instance.
            inverse: same as data, except the categorical features are not
            binary, but categorical (as the original data)
    """
    data = np.zeros((num_samples, data_row.shape[0]))
    categorical_features = range(data_row.shape[0])
    if l.discretizer is None:
        data = l.random_state.normal(
            0, 1, num_samples * data_row.shape[0]).reshape(
            num_samples, data_row.shape[0])
        if l.sample_around_instance:
            data = data * l.scaler.scale_ + data_row
        else:
            data = data * l.scaler.scale_ + l.scaler.mean_
        categorical_features = l.categorical_features
        first_row = data_row
    else:
        first_row = l.discretizer.discretize(data_row)
    data[0] = data_row.copy()
    inverse = data.copy()
    for column in categorical_features:
        values = l.feature_values[column]
        freqs = l.feature_frequencies[column]
        inverse_column = l.random_state.choice(values, size=num_samples,
                                                  replace=True, p=freqs)
        binary_column = np.array([1 if x == first_row[column]
                                  else 0 for x in inverse_column])
        binary_column[0] = 1
        inverse_column[0] = data[0, column]
        data[:, column] = binary_column
        inverse[:, column] = inverse_column
    if l.discretizer is not None:
        inverse[1:] = l.discretizer.undiscretize(inverse[1:])
    inverse[0] = data_row
    return data, inverse

model = pickle.load(open('/content/drive/My Drive/classifier.sav', 'rb'))

a = data_inverse(l, x_train.iloc[0], 1000)

gm = GaussianMixture(n_components=2, covariance_type = 'diag')

p_data_train = a[1]

p_data_test = list(map(lambda x: x>=0.5, model.predict(p_data_train)))

gm.fit(p_data_train, p_data_test)

components = gm.precisions_

ind = []
target = gm.predict(x_train.iloc[0].values.reshape(1,-1))[0]
for i in range(len(p_data_train)):
    if gm.predict(p_data_train[i].reshape(1,-1))[0]==target:
        ind.append(i)

feature_count = np.zeros(135)

for i in ind:
    feature_count = list(map(int, p_data_train[i]!=0))+feature_count

def fidelity_test1(model, x_train, feature_count):
    '''
    If features Fx are accurately selected, then removing Fx from
    the input x will lead to classifying this image to a different
    label, i.e., “shoe” 
        '''
    o = []
    n = []
    new_prob = []
    for x_train_point in x_train.values:
        original = model.predict(x_train_point.reshape(1,-1))>=0.5
        len_ind = 0
        ma = max(feature_count)
        for i in range(len(feature_count)):
            if feature_count[i] == ma and x_train_point[i]!=0:
                x_train_point[i] = 0
                len_ind += 1
            if len_ind == 30:
                break
        np = model.predict(x_train_point.reshape(1,-1))
        new_prob.append(np)
        o.append(original)
        n.append(np>=0.5)
    return o, n, new_prob

def fidelity_test2(model, x_train, feature_count):
    '''
    If features Fx are accurately selected, then adding the feature values of Fx
    to an image of “shoe” is likely to lead to a
    misclassification, i.e., classifying it as a “sweater”
    '''
    new_prob = []
    orig_prob = []
    len_ind = 0
    o = []
    n = []
    ma = max(feature_count)
    ind = []
    for i in range(len(feature_count)):
        #feature_count[i] == ma and 
        if x_train.values[0][i]==0 and feature_count[i] == ma-200:
            ind.append(i)
            len_ind += 1
        if len_ind == 30:
            break
    for x_train_point in x_train.values:
        original = model.predict(x_train_point.reshape(1,-1))
        op = original
        original = original<=0.5
        if original==True:
            original = 1
        else:
            original = 0
        if sum(x_train_point[ind]==0)==len_ind and original!=target:
            for i in range(len_ind):
                x_train_point[ind[i]] = feature_maxs[ind[i]]
            new_predict = model.predict(x_train_point.reshape(1,-1))
            new_prob.append(new_predict)
            new_predict = new_predict>=0.5
            orig_prob.append(op)
            o.append(original)
            n.append(new_predict)

    return new_prob, new_prob, o, n

def fidelity_test3(model, feature_count):
    '''
    If features Fx are accurately selected, we can craft a synthetic
    images that only contains the features in Fx, and this synthetic 
    image is likely to be classified as “sweater”
    '''
    len_ind = 0
    o = []
    n = []
    ma = max(feature_count)
    ind = []
    for i in range(len(feature_count)):
        if feature_count[i] == ma:
            ind.append(i)
            len_ind += 1
        if len_ind == 30:
            break
    x_train_points = [0]*len(feature_count)
    for i in ind:
        x_train_points[i] = feature_means[i]
    return model.predict(np.array(x_train_points).reshape(1,-1))

o1, n1, n_prob = fidelity_test1(model, x_train, feature_count)
orig_prob, ft2_prob, o2, n2 = fidelity_test2(model, x_train, feature_count)
# model.predict(x_train.iloc[0].reshape(1,-1))

# print (ft2_prob, o2, n2)

print (model.predict(x_train.iloc[0].reshape(1,-1)))

# o, n = fidelity_test2(model, x_train, feature_count)
# print (sum(np.array(o)^np.array(n)))
print (fidelity_test3(model, feature_count))
# model.predict(x_train.iloc[0].reshape(1,-1))

# model = Sequential()
# model.add(Dense(1024, input_dim=len(x.columns), activation='relu'))
# model.add(Dense(1024, activation='relu'))
# model.add(Dense(1024, activation='sigmoid'))
# model.add(Dense(1, activation='sigmoid'))
# model.compile(loss='binary_crossentropy', optimizer='adam', metrics=['accuracy'])
# model.fit(x_train,y_train, epochs=20, batch_size=10)
# y_pred = model.predict(x_test)
# y_pred = list(map(lambda x: x>=0.5, y_pred))
# accuracy_score(y_test, y_pred)
# pickle.dump(model, open('classifier.sav', 'wb'))
# from google.colab import files
# files.download("classifier.sav")
# a = [1,1,1,1,1,1,1,1,1,1]
# b = [1,2,3,4]
# # a = np.array(a)
# a = x_train.values[0]
# b = np.array(b)
# sum(a[b]!=0)

y_pred=model.predict(p_data_train)

gm_pred1=[]
gm_pred2=[]
def LocalApproxAccuracy():      
    for i in range(len(y_pred)):
        gm_pred1.append(gm.predict_proba(p_data_train)[i][1])
        gm_pred2.append(gm.predict_proba(p_data_train)[i][0])
    rmse1=mean_squared_error(y_pred,gm_pred1)
    rmse2=mean_squared_error(y_pred,gm_pred2)
    return(min(rmse1,rmse2))

#LIME Prediction 
pred1=[]
for i in range(len(y_pred)):
    pred1.append(l.explain_instance(p_data_train[i],gm.predict_proba, num_features=5).predict_proba[0])
    print (i)
rmse=mean_squared_error(y_pred,pred1)

LocalApproxAccuracy()

def precision(ytrue,ypred):
    return precision_score(ytrue,ypred)
def recall(ytrue,ypred):
    return recall_score(ytrue,ypred)
def roc(ytrue,yscore):         
    fpr, tpr, thresholds = roc_curve(ytrue,yscore)
    plt.figure()
#     print (fpr[2], tpr[2], thresholds, sep='\n')
    plt.plot(fpr, tpr, color='darkorange')
    plt.plot([0, 1], [0, 1], color='navy', lw=2, linestyle='--')
    plt.xlim([0.0, 1.0])
    plt.ylim([0.0, 1.05])
    plt.xlabel('False Positive Rate')
    plt.ylabel('True Positive Rate')
    plt.title('Receiver operating characteristic ')
    plt.show()
def confusionMatrix(ytrue,ypred):
    return confusion_matrix(ytrue,ypred)

for i in range(len(n2)):
    n2[i] = n2[i][0][0]

`range(len(o1)):
    o1[i]=o1[i][0][0]
    n1[i]=n1[i][0][0]

print(precision(o1,n1))
print(recall(o1,n1))
print (confusionMatrix(o1, n1))

for i in range(len(o1)):
    if o1[i]==True:
        o1[i]=1
    else:
        o1[i]=0
for i in range(len(n_prob)):
    n_prob[i] = n_prob[i][0][0]

roc(o1, n_prob)

for i in range(len(o2)):
    n2[i] = n2[i][0][0]

print(precision(o2,n2))
print(recall(o2,n2))
print (confusionMatrix(o2, n2))

for i in range(len(ft2_prob)):
    ft2_prob[i] = ft2_prob[i][0][0]

print(roc(o2,ft2_prob))