CyberFraud1/cross_valid.py at master · simionAndrei/CyberFraud1 · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
from sklearn.metrics import precision_score, recall_score, f1_score, accuracy_score
from sklearn.metrics import confusion_matrix
from sklearn.model_selection import StratifiedKFold

from imblearn.over_sampling import SMOTE

import numpy as np
import json
import time

from hyperparam_tunning import HyperparamGridSearcher
from plot_utils import create_roc_plot

class CrossValidation():

  def __init__(self, data, k_folds, random_seed, logger):
    self.logger = logger
    self.data = data
    self.X, self.y = self.data['X'], self.data['y']
    self.k_folds = k_folds
    self.random_seed = random_seed

  # Evaluates the model using the specified parrameters
  def evaluate_model(self, model, model_hyperparams_grid = None, model_hyperparams_file = None,
    use_smote = True):

    self.logger.log("Start {}-fold cross validation on {} entries".format(
      self.k_folds, self.data['y'].shape[0]))
    start = time.time()

    strat_kfold = StratifiedKFold(n_splits= self.k_folds, shuffle=False,
      random_state = self.random_seed)

    acc_list, prec_list, rec_list, confmat_list = [[] for _ in range(4)]
    for fold_idx, (train_idx, test_idx) in enumerate(strat_kfold.split(self.X, self.y)):
      crt_X_train, crt_X_test = self.X[train_idx], self.X[test_idx]
      crt_y_train, crt_y_test = self.y[train_idx], self.y[test_idx]

      if use_smote:
        self.logger.log("Using SMOTE")
        sm = SMOTE(random_state = 13)
        crt_X_train, crt_y_train = sm.fit_sample(crt_X_train, crt_y_train.ravel())
        sample_weight = [1 if int(sample) == 1 else 1 for sample in crt_y_train]
      else:
        sample_weight = [110 if int(sample) == 1 else 1 for sample in crt_y_train]


      if model_hyperparams_grid:
        valid_data = {'X': crt_X_train, 'y': crt_y_train}
        grid_searcher = HyperparamGridSearcher(valid_data, self.logger)
        hyperparams = grid_searcher.rand_grid_search(model, model_hyperparams_grid, 50)
      else:
        with open(self.logger.get_model_file(model_hyperparams_file)) as fp:
          hyperparams = json.load(fp)

      model.set_params(**hyperparams)

      model.fit(crt_X_train, crt_y_train, sample_weight = sample_weight)

      plt_title = type(model).__name__ + " " + ("with " if use_smote else "without ") + "SMOTE"
      plt_title += " fold {}".format(fold_idx)
      filename = "roc_" + ("withS_" if use_smote else "withoutS_") + str(fold_idx)
      filename += "_" + type(model).__name__ + ".png"
      create_roc_plot(plt_title, model.predict_proba(crt_X_test), crt_y_test, filename,
        self.logger)

      y_pred = model.predict_proba(crt_X_test)
      y_pred = y_pred[:, 1]
      y_pred = y_pred > 0.5

      tn, fp, fn, tp = confusion_matrix(crt_y_test, y_pred).ravel()
      acc_list.append(accuracy_score(crt_y_test, y_pred))
      prec_list.append(precision_score(crt_y_test, y_pred))
      rec_list.append(recall_score(crt_y_test, y_pred))
      confmat_list.append([tn, fp, fn, tp])

      self.logger.log("TN: {}, FP: {}, FN: {}, TP: {} at fold#{}".format(tn, fp, fn, tp, fold_idx))
      self.logger.log("Accuracy at fold#{}: {:.2f}".format(fold_idx, acc_list[-1]))
      self.logger.log("Precision at fold#{}: {:.2f}".format(fold_idx, prec_list[-1]))
      self.logger.log("Recall at fold#{}: {:.2f}".format(fold_idx, rec_list[-1]))

    self.logger.log("Finished cross validation in {:.2f}s".format(time.time() - start))
    self.logger.log("Mean accuracy: {}".format(np.average(acc_list)))
    self.logger.log("Mean precision: {}".format(np.average(prec_list)))
    self.logger.log("Mean recall: {}".format(np.average(rec_list)))
    self.logger.log("Mean FP: {}".format(np.average([e[1] for e in confmat_list])))
    self.logger.log("Mean TP: {}".format(np.average([e[3] for e in confmat_list])))
    self.logger.log("Total FP: {}".format(sum([e[1] for e in confmat_list])))
    self.logger.log("Total TP: {}".format(sum([e[3] for e in confmat_list])))