detector.py

import numpy as np
import pandas as pd

from sklearn import metrics
from sklearn import cross_validation
from sklearn.grid_search import GridSearchCV
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.neighbors import KNeighborsClassifier
from sklearn.naive_bayes import MultinomialNB
from sklearn.svm import SVC
from sklearn.ensemble import RandomForestClassifier

from sklearn.externals import joblib

"""
Author: bienkma
Date: 11/07/2016
About: This dataset is composed of a selection of Windows API/System-Call trace files, intended for testing on classifiers
treating with sequences. You can download data set http://csmining.org/index.php/malicious-software-datasets-.html link
"""


class detector_malware:

    def __init__(self):
        '''__init__ method for all value define'''

        # Training data
        train_file = "./CSDMC_API_Train.csv"

        # Test data. Note test data not contain malware. So I will mush-up to training data. And Split 1/4 for test
        test_file = "./CSDMC_API_TestData.csv"

        data_train = pd.read_csv(train_file, header=0, names=['0', '1'])

        # Train data set
        X_train = data_train['1'].tolist()
        Y_train = data_train['0'].tolist()

        data_test = pd.read_csv(test_file, header=0, delimiter=',', names=['0', '1'])

        # Data test
        X_test = data_test['1'].tolist()
        Y_test = data_test['0'].tolist()

        # Mush-Up data
        corpus = X_train + X_test
        labels = Y_train + Y_test

        # Split data for testing
        doc_train, doc_test, self.y_train, self.y_test = cross_validation.train_test_split(corpus, labels,
                                                                                           test_size=0.25,
                                                                                           stratify=labels,
                                                                                           random_state=7)

        # TTfidfVectorizer vector
        vectorizer = TfidfVectorizer(ngram_range=(1, 1), min_df=1, lowercase=True)
        # X_train_ = vectorizer.fit_transform(X_train)
        vectorizer.fit(X_train)
        self.X_train_vec = vectorizer.transform(doc_train)
        self.X_test_vec = vectorizer.transform(doc_test)
        joblib.dump(vectorizer, './model/vt.pkl')


    def knn(self):
        """__K-Neighbors Algorithm__"""
        # Create array 30 item for calculate the best K
        knn = KNeighborsClassifier()
        k = np.arange(30) + 1
        parameters = {'n_neighbors': k}

        clf = GridSearchCV(knn, parameters, cv=10)
        clf.fit(self.X_train_vec, self.y_train)

        # Prediction
        y_pred = clf.predict(self.X_test_vec)
        print metrics.classification_report(self.y_test, y_pred)
        print "The best choice k for KNN algorithm is: {0}".format(clf.best_estimator_.n_neighbors)
        print "------------"


    def navie_bayes(self):
        """__Navie Bayes Algorithm__"""
        nb = MultinomialNB()
        k = np.arange(100) * 0.1
        parameters = {'alpha': k}

        clf = GridSearchCV(nb, parameters, cv=10)
        clf.fit(self.X_train_vec, self.y_train)

        # Prediction
        y_pred = clf.predict(self.X_test_vec)
        print metrics.classification_report(self.y_test, y_pred)
        print "The best choice Alpha for NB algorithm is: {0}".format((clf.best_params_)['alpha'])
        print "------------"


    def svm(self):
        """__Support Vector Machine for regression implemented using libsvm__"""
        svc = SVC()
        parameters = [
            {'C': [1, 10, 100, 1000, 10000], 'kernel': ['linear']},
            {'C':[1, 10, 100, 1000], 'gamma':[0.01, 0.001, 0.0001], 'kernel':['rbf']},
        ]
        clf = GridSearchCV(svc, parameters, cv=10)
        clf.fit(self.X_train_vec, self.y_train)

        # Prediction
        y_pred = clf.predict(self.X_test_vec)
        print metrics.classification_report(self.y_test, y_pred)
        print "The best choice parameters for SVM algorithm is: ", clf.best_estimator_
        print "------------"

    def rf(self):
        """__Random forest for classifier__"""
        rf = RandomForestClassifier()
        parameters = {"max_depth":[3, None],
                      "max_features": [1, 3, 10],
                      "min_samples_split":[1, 3, 10],
                      "min_samples_leaf": [1, 3, 10],
                      "bootstrap": [True, False],
                      "criterion": ["gini", "entropy"]}

        clf = GridSearchCV(rf, parameters)
        clf.fit(self.X_train_vec, self.y_train)

        # Prediction
        y_pred = clf.predict(self.X_test_vec)
        print metrics.classification_report(self.y_test, y_pred)
        print "The best choice parameters for RF algorithm is: ", clf.best_estimator_
        print "------------"

        # clf = RandomForestClassifier()
        # clf.fit(self.X_train_vec, self.y_train)
        # y_pred = clf.predict(self.X_test_vec)
        # print metrics.classification_report(self.y_test, y_pred)

    def export_rf(self):
        rf = RandomForestClassifier(max_depth=None, max_features=3, max_leaf_nodes=None,
                                    min_samples_split=10, min_weight_fraction_leaf=0.0,
                                    n_estimators=10, n_jobs=1, oob_score=False, random_state=None,
                                    verbose=0, warm_start=False)
        rf.fit(self.X_train_vec, self.y_train)
        joblib.dump(rf, './model/rf.pkl')


if __name__ == '__main__':
    # Implement programe
    alg = detector_malware()

    # Navie Bayes
    alg.navie_bayes()


    # K-Neighbors
    alg.knn()

    #SVM
    alg.svm()

    # Random Forest
    alg.rf()


    # Export Model for API web services call predict.py
    alg.export_rf()