doc2vec_training_plus_binary_classification.py

# -*- coding: utf-8 -*-
"""
Created on Fri Jun 18 14:37:31 2021
@author: s441121
"""
import math

# !/usr/bin/env python
# coding: utf-8

# In[1]:
from androidMalwareDetectionWithNLP.utilities import constants
import subprocess
import logging
import os
import collections
import gensim
import multiprocessing

from collections import OrderedDict
from collections import defaultdict

import gensim.models.doc2vec

from gensim.models.doc2vec import Doc2Vec
from gensim.test.test_doc2vec import ConcatenatedDoc2Vec

from sklearn.linear_model import LogisticRegression
from sklearn.naive_bayes import GaussianNB
from sklearn.metrics import accuracy_score, confusion_matrix
from sklearn import tree
from sklearn.svm import SVC
from sklearn.neighbors import KNeighborsClassifier
import numpy as np
import statsmodels.api as sm
from random import shuffle

assert gensim.models.doc2vec.FAST_VERSION > -1, "This will be painfully slow otherwise"

# In[2]:


logging.basicConfig(format='%(asctime)s : %(levelname)s : %(message)s', level=logging.INFO)
DexDocument = collections.namedtuple('DexDocument', 'words tags split label')


# In[3]:


def list_dir(directory):
    cmd_out = []
    for root, dirs, files in os.walk(directory):
        for file in files:
            if file.endswith(".txt"):
                cmd_out.append(os.path.join(root, file))
    return cmd_out


# In[4]:


def dex_doc(label, text, index, split):
    # tokens = gensim.utils.to_unicode(text).split()
    tokens = gensim.parsing.preprocess_string(text)
    return DexDocument(tokens, [index], split, label)


# IN[5a]

ls_mal =[file for file in os.listdir(constants.DEXDUMP_PATH_PLAIN) if "malicious" in file]
len_ls_mal = len(ls_mal)
len_ls_mal_test=math.ceil(len_ls_mal*0.2)

ls_benign =[file for file in os.listdir(constants.DEXDUMP_PATH_PLAIN) if "benign" in file]
len_ls_benign = len(ls_benign)
len_ls_benign_test=math.ceil(len_ls_benign*0.2)


for filename in os.listdir(constants.DEXDUMP_PATH_PLAIN):
    if "malicious" in filename:
        if len_ls_mal_test>0:
            subprocess.run('mv {} {}'.format(constants.DEXDUMP_PATH_PLAIN + filename,
                                             constants.DEXDUMP_PATH_PLAIN + "test_" + filename),
                           shell=True,
                           capture_output=True,
                           text=True,
                           check=True).stdout.split('\n')
            len_ls_mal_test = len_ls_mal_test - 1
        else:
            subprocess.run('mv {} {}'.format(constants.DEXDUMP_PATH_PLAIN + filename,
                                             constants.DEXDUMP_PATH_PLAIN + "train_" + filename),
                           shell=True,
                           capture_output=True,
                           text=True,
                           check=True).stdout.split('\n')

    else:
        if len_ls_benign_test>0:
            subprocess.run('mv {} {}'.format(constants.DEXDUMP_PATH_PLAIN + filename,
                                             constants.DEXDUMP_PATH_PLAIN + "test_" + filename),
                           shell=True,
                           capture_output=True,
                           text=True,
                           check=True).stdout.split('\n')
            len_ls_benign_test = len_ls_benign_test - 1
        else:
            subprocess.run('mv {} {}'.format(constants.DEXDUMP_PATH_PLAIN + filename,
                                             constants.DEXDUMP_PATH_PLAIN + "train_" + filename),
                           shell=True,
                           capture_output=True,
                           text=True,
                           check=True).stdout.split('\n')











# In[5]:
def extract_dex():

    # file_names = list_dir("appData/dexDumps/hexdump")
    file_names = list_dir("dataset/dexDumps/plain/")
    index = 0
    for dex_file in file_names:
        if len(dex_file) > 0:
            with open(dex_file, encoding="utf8") as f:

                content = f.read(120000)
                # print(content)
                if "malicious" in dex_file:
                    label = 1
                else:
                    label = 0

                if "train" in dex_file:
                    split = "train"
                elif "test" in dex_file:
                    split = "test"
                yield dex_doc(label, content, index, split)
                index += 1
        else:
            print("Probably the end of list!")


# In[6]:

all_dex_files = list(extract_dex())

print(all_dex_files[1])

# In[7]:


train_docs = [doc for doc in all_dex_files if doc.split == 'train']
test_docs = [doc for doc in all_dex_files if doc.split == 'test']
print('%d docs: %d train-document, %d test-document' % (len(all_dex_files), len(train_docs), len(test_docs)))

# In[8]:


common_kwargs = dict(
    vector_size=150, epochs=20, min_count=2,
    sample=0, workers=multiprocessing.cpu_count(), negative=5, hs=0,
)

# In[9]:


simple_models = [
    # PV-DBOW plain
    Doc2Vec(dm=0, **common_kwargs),
    # PV-DM w/ default averaging; a higher starting alpha may improve CBOW/PV-DM modes
    Doc2Vec(dm=1, window=10, alpha=0.05, comment='alpha=0.05', **common_kwargs),
    # PV-DM w/ concatenation - big, slow, experimental mode
    # window=5 (both sides) approximates paper's apparent 10-word total window size
    Doc2Vec(dm=1, dm_concat=1, window=5, **common_kwargs),
]

# In[10]:


print("Building the Vocab..")

for model in simple_models:
    model.build_vocab(all_dex_files)
    print("%s vocabulary scanned & state initialized" % model)

# In[11]:


models_by_name = OrderedDict((str(model), model) for model in simple_models)

models_by_name['dbow+dmm'] = ConcatenatedDoc2Vec([simple_models[0], simple_models[1]])
models_by_name['dbow+dmc'] = ConcatenatedDoc2Vec([simple_models[0], simple_models[2]])


# In[12]:


def logistic_predictor_from_sklearn(train_targets, train_regressors):
    """Fit a sklearn logistic predictor on supplied data"""
    logit = LogisticRegression(max_iter=500)
    predictor = logit.fit(train_regressors, train_targets)
    return predictor


def naive_bayes_from_sklearn(train_targets, train_regressors):
    gnb = GaussianNB()
    predictor = gnb.fit(train_regressors, train_targets)
    return predictor


def decision_tree_from_sklearn(train_targets, train_regressors):
    clf = tree.DecisionTreeClassifier()
    predictor = clf.fit(train_regressors, train_targets)
    return predictor


def support_vector_machine_from_sklearn(train_targets, train_regressors):
    clf = SVC(kernel='rbf')
    predictor = clf.fit(train_regressors, train_targets)
    return predictor


def kNN_from_sklearn(train_targets, train_regressors):
    knn = KNeighborsClassifier(n_neighbors=3)
    predictor = knn.fit(train_regressors, train_targets)
    return predictor


# In[13]:


def error_rate_for_model(test_model, train_set, test_set):
    train_labels = [doc.label for doc in train_set]
    #print("train_labels", train_labels)
    train_features = [test_model.dv[doc.tags[0]] for doc in train_set]
    #print("train_features", train_features)
    # train_features = sm.add_constant(train_features)
    predictor1 = logistic_predictor_from_sklearn(train_labels, train_features)
    predictor2 = naive_bayes_from_sklearn(train_labels, train_features)
    predictor3 = kNN_from_sklearn(train_labels, train_features)
    predictor4 = support_vector_machine_from_sklearn(train_labels, train_features)
    predictor5 = decision_tree_from_sklearn(train_labels, train_features)

    test_labels = [doc.label for doc in test_set]
    test_features = [test_model.dv[doc.tags[0]] for doc in test_set]
    #print("test_labels", test_labels)
    #print("test_features", test_features)
    # test_features = sm.add_constant(test_features)

    # Predict & evaluate
    test_predictions = predictor1.predict(test_features)
    train_accuracy = accuracy_score(test_labels, test_predictions)
    print("----------- predictor 1 -------------------")
    print("Training Accuracy: ", train_accuracy)
    print("Confusion Matrix: ", confusion_matrix(test_labels, test_predictions))

    test_predictions = predictor2.predict(test_features)
    train_accuracy = accuracy_score(test_labels, test_predictions)
    print("----------- predictor 2 -------------------")
    print("Training Accuracy: ", train_accuracy)
    print("Confusion Matrix: ", confusion_matrix(test_labels, test_predictions))

    test_predictions = predictor3.predict(test_features)
    print("test_predictions", test_predictions)
    train_accuracy = accuracy_score(test_labels, test_predictions)
    print("----------- predictor 3 -------------------")
    print("Training Accuracy: ", train_accuracy)
    print("Confusion Matrix: ", confusion_matrix(test_labels, test_predictions))

    test_predictions = predictor4.predict(test_features)
    train_accuracy = accuracy_score(test_labels, test_predictions)
    print("----------- predictor 4 -------------------")
    print("Training Accuracy: ", train_accuracy)
    print("Confusion Matrix: ", confusion_matrix(test_labels, test_predictions))

    test_predictions = predictor5.predict(test_features)
    train_accuracy = accuracy_score(test_labels, test_predictions)
    print("----------- predictor 5 -------------------")
    print("Training Accuracy: ", train_accuracy)
    print("Confusion Matrix: ", confusion_matrix(test_labels, test_predictions))


# In[14]:


error_rates = defaultdict(lambda: 1.0)

shuffled_alldocs = all_dex_files[:]
shuffle(shuffled_alldocs)

# print(train_docs)

print("dece", error_rates)


# In[ ]:


for model in simple_models:
    print("Training %s" % model)
    model.train(shuffled_alldocs, total_examples=len(shuffled_alldocs), epochs=model.epochs)

    print("\nEvaluating %s" % model)
    error_rate_for_model(model, train_docs, test_docs)
    # error_rates[str(model)] = err_rate
    # print("\n%f %s\n" % (err_rate, model))

# In[ ]:


for model in [models_by_name['dbow+dmm'], models_by_name['dbow+dmc']]:
    print("\nEvaluating %s" % model)
    error_rate_for_model(model, train_docs, test_docs)
    # error_rates[str(model)] = err_rate
    # print("\n%f %s\n" % (err_rate, model))

# In[ ]:


   # print("Err_rate Model")
   # for rate, name in sorted((rate, name) for name, rate in error_rates.items()):
   #  print("%f %s" % (rate, name))


# In[ ]: