import logging import pandas as pd import numpy as np import matplotlib.pyplot as plt from sklearn.feature_selection import RFECV from sklearn.metrics import classification_report, confusion_matrix, accuracy_score, log_loss from sklearn.model_selection import GridSearchCV from sklearn.preprocessing import MinMaxScaler, StandardScaler from sklearn.linear_model import LogisticRegression from sklearn.ensemble import RandomForestClassifier from sklearn.inspection import permutation_importance from sklearn.utils import class_weight from abc import ABC, abstractmethod class Model(ABC): def __init__(self): super().__init__() logging.info('Initialising Model') @abstractmethod def train(self): logging.info('Training Model') @abstractmethod def predict(self): logging.info('Predicting...') ############################################################################################################# ####################### Logistic Regression ################################################################ ############################################################################################################# class LogisticRegressionModel(Model): def __init__(self, prepared_data, outcome, cutoff = 0.50): super().__init__() self.X = prepared_data self.y = outcome self.model = None self.cutoff = cutoff self.class_weights = None def train(self): best_params = self.search_best_parameters(self.X, self.y) # log_model = LogisticRegression(solver = best_params['solver'], # C = best_params['C'], # max_iter = best_params['max_iter'], # penalty = best_params['penalty'], # class_weight = 'balanced') #self.model = RFECV(log_model, step=1, cv=10) #self.model.fit(self.X, self.y.values.ravel()) # <----- this does recursive feature elimination with cross-validation: in the end I decided to use Lasso. self.class_weights = class_weight.compute_class_weight('balanced', classes=np.unique(self.y), y=self.y.values.reshape(-1)) self.model = LogisticRegression(solver = best_params['solver'], C = best_params['C'], max_iter = best_params['max_iter'], penalty = best_params['penalty'], #class_weight = 'balanced').fit(self.X, self.y.values.ravel()) class_weight = dict(enumerate(self.class_weights))).fit(self.X, self.y.values.ravel()) # <--------------- tried this for cost-sensitive learning (by assigning higher misclassification cost to the minority class, model is encouraged to prioritise correctly the identification of this class. def predicted_probabilities(self, X_to_predict): return self.model.predict_proba(X_to_predict) def predict(self, X_to_predict): pred = self.predicted_probabilities(X_to_predict) prob_1 = [x[1] for x in pred] return np.where(np.array(prob_1) > self.cutoff, 1, 0) def model_intercept(self): #print(self.model.estimator_.intercept_) print(self.model.intercept_) def model_coefficients(self): for coefficients, feature in zip(self.model.coef_[0], self.X.columns): print(f"{feature}: {coefficients}") def search_best_parameters(self, X_train, y_train): """ Using grid search to identify a good set of parameters """ logging.info('Searching params for Logistic Regression ... \n') param_grid = { #'solver': ['liblinear', 'newton-cg', 'lbfgs', 'saga'], # solver and penalty params can't be on at the same time. l1 and l2 support different solvers 'solver': ['liblinear', 'saga'], 'C': [0.001, 0.01, 0.1, 1, 10, 100, 1000], 'max_iter': [1000], 'penalty' : ['l1', 'l2']} # CV_logreg = GridSearchCV(estimator=LogisticRegression(random_state=2023), param_grid=param_grid, cv=10, error_score='raise') CV_logreg.fit(X_train, y_train.values.ravel()) result = CV_logreg.best_params_ logging.info('Best parameters found: %s', result) return result # def important_features(self): # df_features = pd.DataFrame(columns= ['Feature', 'Ranking']) # for i in range(self.X.shape[1]): # row = {'Feature': self.X.columns[i], 'Ranking': self.model.ranking_[i]} # df_features = df_features.append(row, ignore_index=True) # return df_features.sort_values(by = 'Ranking') def important_features(self): model_fi = permutation_importance(self.model, self.X, self.y, scoring = 'f1') importances = model_fi.importances_mean for feature, importance in zip(self.X.columns, importances): print(f"{feature}: {importance}") def ROC_plot(self): return sklearn.metrics.plot_roc_curve(CV_logreg, X_train, y_train.values.ravel()) ############################################################################################################# ####################### Random Forest ###################################################################### ############################################################################################################# class RandomForestModel(Model): def __init__(self, prepared_data, outcome): super().__init__() self.X = prepared_data self.y = outcome self.model = None def train(self): # X_array = self.X.toarray() <----- commented params there because I tried different things # best_params = self.search_best_parameters(X_array, self.y) best_params = self.search_best_parameters(self.X, self.y) self.model = RandomForestClassifier(random_state= 2023, max_features = best_params['max_features'], n_estimators = best_params['n_estimators'], max_depth = best_params['max_depth'], criterion = best_params['criterion'], class_weight = 'balanced', oob_score=True) self.model.fit(self.X, self.y.values.ravel()) #self.model.fit(X_array, self.y.values.ravel()) def predict(self, X_to_predict): return self.model.predict(X_to_predict) def search_best_parameters(self, X_train, y_train): """ Using grid search to identofy a good set of parameters """ logging.info('Searching params for Random Forest ... \n') param_grid = { 'max_features': ['auto', 'sqrt', 'log2'], 'n_estimators': [50, 100, 200, 500, 1000], 'max_depth': [4,5,6,7,8,9,10], 'criterion': ['gini', 'entropy'] } CV_rfc = GridSearchCV(estimator=RandomForestClassifier(random_state=2023), param_grid=param_grid, cv=10) CV_rfc.fit(X_train, y_train.values.ravel()) result = CV_rfc.best_params_ logging.info('Best parameters found: %s', result) return result def important_features(self): important_features = self.model.feature_importances_ importances = pd.Series(important_features, index=self.X.columns).sort_values(ascending=False) return importances def estimate_predictor(model, X_test, y_test): """ Estimates accuracy of model """ pred_y = model.predict(X_test) observed_y = y_test return classification_report(observed_y, pred_y, output_dict=True) def confusion_matrix_calculation(model, X_test, y_test): pred_y = model.predict(X_test) observed_y = y_test return confusion_matrix(observed_y, pred_y) def accuracy_calculator(model, X_test, y_test): pred_y = model.predict(X_test) observed_y = y_test return accuracy_score(observed_y, pred_y) def loss_evaluation(model, X_test, y_test): pred_y = model.predict(X_test) observed_y = y_test return log_loss(observed_y, pred_y)