# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3
#     language: python
#     name: python3
# ---

# +
import shap
import joblib
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import seaborn as sns; sns.set()

from sklearn.preprocessing import normalize
from sklearn.model_selection import train_test_split

from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import GridSearchCV

from sklearn.metrics import plot_roc_curve, classification_report
# -

import os
import logging as lg

a = 'toto'



# - prévoir un argument pour définir si l'on veut entrainer les modèles ou récupérer les .pkl

IMAGES_EDA_OUTPUT_DIR = 'images/eda'
IMAGES_EDA_SUB_DIR = [
    'univariate_num_analysis',
    'univarite_cat_analysis',
    'bivariate_cat_analysis',
    'bivariate_num_analysis',
    'target'
]

for sub_dir in IMAGES_EDA_SUB_DIR:
    path = os.path.join(IMAGES_EDA_OUTPUT_DIR, sub_dir)
    safe_creation_directory(path)


def safe_creation_directory(path):
    """
    Check if directory exists, if not, create it
    :param path: path to store eda results
    """
    if not os.path.isdir(path):
        os.makedirs(path)
        lg.info(f'folder has been created at {path}')
    else:
        lg.info(f'EDA images are stored in {path}')


path_test = 'images/eda/univarite_cat_analysis/_histplot_distribution.png', bbox_inches='tight'

# +
"""
File to store all functions for customer churn predictions
"""
import logging as lg
import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns; sns.set()

from sklearn.model_selection import train_test_split
from sklearn.model_selection import GridSearchCV

from sklearn.metrics import plot_roc_curve, classification_report


def import_data(input_path: str, num_columns: list, cat_columns: list, target_column: list) -> pd.DataFrame:
    """
    Read csv file
    :param input_path: path to csv file
    :param num_columns: columns names for numeric variables
    :param cat_columns: columns names for categorical variables
    :param target_column: target column
    :return: a pandas dataframe without unnamed columns
    """
    try:
        dataframe = pd.read_csv(input_path)
        return dataframe[num_columns + cat_columns + target_column]
    except FileNotFoundError as err:
        lg.info('File not found')
        raise err

'''
def perform_eda(df, col):
   plt.figure(figsize=(20, 10))
   if isinstance(col, object):
       perform_eda_on_cat_columns()
   else:
       perform_eda_on_num_columns()
'''


#def perform_eda_on_cat_columns(df, cat_col):
# def performn_eda_on_num_columns(df, num_col)


#def encode_target(df: pd.DataFrame, target_name: str, target_encoding: dict) -> dict:
 #   """
  #  Encode target variable
   # :param df: a pandas dataframe
    #:param target_name: the target column to encode
    #:param target_encoding: how to encode the target value
    #:return: a dictionary with <new_encode : value to be encoded>
    #"""
    #for key, value in target_encoding.items():
    #    return {key: df[target_name] == value}

    


#def encoder_helper(df: pd.DataFrame, feature: str) -> pd.DataFrame:
#    """
#    Feature engineering - compute a target encoding
#    :param df: the pandas dataframe
#    :param feature: a column from the dataframe
#    :return: a pandas dataframe with the target encoding to the feature
#    """
#    try:
#        feature_groups = df.groupby(feature).mean()['Churn']
#        df['{0}_Churn'.format(feature)] = df[feature].map(feature_groups)
#        return df
#    except:
#        lg.info("Something went wrong")


def perform_feature_engineering(df, target, test_size, random_state):
    """
    Split data into train and test labels
    :param df: a pandas dataframe
    :param target: target column
    :param test_size: size of test sample
    :param random_state: controls the randomness of the sampled classes
    :return: data and labels for train and test samples - return pandas dataframes & series
    """
    df = df.select_dtypes(exclude=['object'])
    return train_test_split(df.drop(target, axis=1), df[target], test_size=test_size, random_state=random_state), df.drop(target, axis=1).columns


def classification_report_image(true_labels, predictions, name_model, mode, output_path):
    """
    produces classification report for training and testing results and stores report as image
    in images folder
    :param true_labels:
    :param predictions:
    :param name_model:
    :output_path:
    :return:
    """
    try:
        lg.info(f'{name_model} results')
        if mode == 'train':
            lg.info(f'{mode} results')
            return classification_report(true_labels, predictions)
    except:
        lg.info('Something went from')


def train_models(x_train_data, x_test_data, y_train_data, model, param_grid=None, cv=None,
                 grid_search=False, do_probabilities=False):
    try:
        if grid_search:
            gs = GridSearchCV(
                estimator=model,
                param_grid=param_grid,
                cv=cv,
                n_jobs=-1,
                verbose=2
            )
            fitted_model = gs.fit(x_train_data, y_train_data)
        else:
            fitted_model = model.fit(x_train_data, y_train_data)

        predictions = fitted_model.predict_proba(x_test_data) if do_probabilities else fitted_model.predict(x_test_data)
        return fitted_model, predictions
    except:
        lg.info('complicated life')


def plot_roc_curves(model, true_labels, predictions):
    """
    Plot roc curve associated to the model
    :param model:
    :param true_labels:
    :param predictions:
    :return:
    """
    try:
        return plot_roc_curve(model, true_labels, predictions)
    except:
        lg.info('try something')


def encoder_helper(df: pd.DataFrame, target_name: str, new_target_name: str, target_encoding_dict: dict, cat_feature: list) -> pd.DataFrame:
    """
    Encode target and categorical features
    :param df: a pandas dataframe
    :param target_name: the target column to encode
    :param new_target_name: name of the target col encoded
    :param target_encoding: how to encode the target value
    :param cat_feature: list of categorical columns
    :return: a pandas dataframe with the target and categorical column encoded
    """
    try:
        dataframe = encode_target(df, target_name, new_target_name, target_encoding_dict)
        for col in cat_feature:
            encoder_cat_features(dataframe, col)  
        return dataframe
    except:
        lg.info("Something went wrong")

        
        
def encoder_cat_features(df: pd.DataFrame, feature: str) -> pd.DataFrame:
    """
    Feature engineering - compute a target encoding
    :param df: the pandas dataframe
    :param feature: a column from the dataframe
    :return: a pandas dataframe with the target encoding to the feature
    """
    try:
        feature_groups = df.groupby(feature).mean()['Churn']
        df['{0}_Churn'.format(feature)] = df[feature].map(feature_groups)
        return df
    except:
        lg.info("Something went wrong")
        
def compute_encoding_target_dict(df: pd.DataFrame, target_name: str, target_encoding_dict: dict) -> dict:
    """
    Compute dictionary for encoding target
    :param df: a pandas dataframe
    :param target_name: the target column to encode
    :param target_encoding_dict: how to encode the target value
    :return: a dictionary with <new_encode : value to be encoded>
    """
    for key, value in target_encoding_dict.items():
        return {key: df[target_name] == value}
        

def encode_target(df: pd.DataFrame, target_name: str, new_target_name: str, target_encoding_dict: dict) -> pd.Series:
    """
    Encode target variable
    :param df: a pandas dataframe
    :param target_name: the target column to encode
    :param new_target_name: name of the target col encoded
    :param target_encoding: how to encode the target value
    :return: a pandas series that corresponds to the new target column encoded
    """
    try:
        encoding_dict = compute_encoding_target_dict(df, target_name, target_encoding_dict)
        df[new_target_name] = np.select(encoding_dict.values(), encoding_dict.keys(), default=1)
        return df.drop(target_name, axis=1)
    except KeyError as error:
        lg.info('something went wrong')
        raise err


# -

def perform_eda(df, NUM_COLS, CAT_COLS, TARGET):
    #plt.figure(figsize=(20,10)) 
    for col in NUM_COLS:
        univariate_num_analysis(df, col)
        bivariate_num_analysis(df, col)
    for col in CAT_COLS:
        univariate_cat_analysis(df, col)
        bivariate_cat_analysis(df, col)
    correlation_matrix(df)
    plot_target_distribution(df, TARGET)


def safe_creation_directory(path):
    '''
    Check if directory exists, if not, create it
    '''
    if not os.path.isdir(path):
        os.makedirs(path)
        lg.info(f'folder has been created at {path}')
    else:
        lg.info(f'EDA images are stored in {path}')


IMAGES_OUTPUT_DIR = 'images/eda'
if not os.path.isdir(IMAGES_OUTPUT_DIR):
    os.makedirs(IMAGES_OUTPUT_DIR)
    logging.info(f'folder has been created at {IMAGES_OUTPUT_DIR}')
else:
    logging.info(f'EDA images are stored in {IMAGES_OUTPUT_DIR}')

os.path.isdir('images')

a = ['AA']


def bivariate_cat_analysis(df, feature, target):
    return sns.catplot(x=feature, hue=target, data=df, kind="count")   


def bivariate_cat_analysis(df, feature, target, output_dir):
    """
    Bivariate analysis of numerical columns with target col
    :param df: a pandas dataframe
    :param feature: numerical column to cross with target col
    :param target: target column
    :param element: figure style with seaborn
    :param output_dir: path to store result
    """
    cat_bivariate_plot = sns.catplot(x=feature, hue=target, data=df)
    cat_bivariate_plot.figure.savefig(f'{output_dir}/{feature}_bivariate_cat_analysis.png')


for col in CAT_COLUMNS:
    sns.catplot(x=col, hue='Attrition_Flag', data=df, kind='count')


def univariate_num_analysis(df: pd.DataFrame, feature: str, output_dir: str):
    """
    Plot distribution of continuous columns and save result
    :param df: A pandas dataframe
    :param feature: numerical column to analyse
    :param output_dir: path to store result
    """
    plt.figure(figsize=(20,10)) 
    num_plot = sns.displot(df[feature])
    num_plot.savefig(f'{output_dir}/{feature}_distplot_distribution.png')


for col in NUM_COLUMNS:
    univariate_num_analysis(df, col, IMAGES_OUTPUT_DIR)


def univariate_cat_analysis(df, feature, output_dir):
    plt.figure(figsize=(20,10)) 
    plot = sns.histplot(df[feature])
    plot.figure.savefig(f'{output_dir}/{col}_histplot_distribution.png')


for col in CAT_COLUMNS:
    univariate_cat_analysis(df, col, IMAGES_OUTPUT_DIR)


def univariate_num_analysis(df, feature):
    return sns.distplot(df[feature])


def bivariate_num_analysis(df, feature, target, element):
    #return sns.histplot(df, x=feature, hue=target, element='poly')
    return sns.histplot(df, x=feature, hue=target, element=element)


def bivariate_cat_analysis(df, feature, target):
    return sns.catplot(x=feature, hue=target, data=df, kind="count")    


for col in CAT_COLUMNS:
    bivariate_cat_analysis(df, col, 'Attrition_Flag')

sns.catplot("Pclass", hue="Survived", data=df, kind="count", 
            palette={1:"blue", 0:"green"}, ax=ax)


def correlation_matrix(df):
    return sns.heatmap(df.corr(), annot=False, cmap='Dark2_r', linewidths = 2)


def plot_target_distribution(df, target):
    plt.title('target distribution')
    df[target].value_counts('normalize').plot(kind='bar');


plot_target_distribution(df, 'Attrition_Flag')

correlation_matrix(df)



plt.figure(figsize=(20,10))
plt.title('Gender')
df['Gender'].hist()

for col in NUM_COLUMNS:
    univariate_num_analysis(df, col)

for col in CAT_COLUMNS:
    univariate_cat_analysis(df, col)

# roc curve


# +
def save_model(model, output_path):
    safe_creation_directory(output_path)
    return joblib.dump(model, output_path)

def load_model(model, input_path):
    return joblib.load(input_path)


# +
CAT_COLUMNS = [
    'Gender',
    'Education_Level',
    'Marital_Status',
    'Income_Category',
    'Card_Category'
]

NUM_COLUMNS = [
    'Customer_Age',
    'Dependent_count',
    'Months_on_book',
    'Total_Relationship_Count',
    'Months_Inactive_12_mon',
    'Contacts_Count_12_mon',
    'Credit_Limit',
    'Total_Revolving_Bal',
    'Avg_Open_To_Buy',
    'Total_Amt_Chng_Q4_Q1',
    'Total_Trans_Amt',
    'Total_Trans_Ct',
    'Total_Ct_Chng_Q4_Q1',
    'Avg_Utilization_Ratio'
]

TARGET_COL = [
    'Attrition_Flag'
]

ENCODING_TARGET = {
    0: 'Existing Customer',
    1: 'Attrited Customer'
}

PARAM_GRID = {
    'n_estimators': [200, 500],
    'max_features': ['auto', 'sqrt'],
    'max_depth': [4, 5, 100],
    'criterion': ['gini', 'entropy']
}


# +
# type(fitted_model).__name__  -> get model name
# -

df = import_data("../churn/data/bank_data.csv", NUM_COLUMNS, CAT_COLUMNS, TARGET_COL)

df.head()

df[CAT_COLUMNS+TARGET_COL].head(10)

df.sample(100).to_csv('..')





# ### ENCODE TARGET COLUMN

# + active=""
# -> AttributeError

# +
def encoder_helper(df: pd.DataFrame, target_name: str, new_target_name: str, target_encoding_dict: dict, cat_feature: list) -> pd.DataFrame:
    """
    Encode target and categorical features
    :param df: a pandas dataframe
    :param target_name: the target column to encode
    :param new_target_name: name of the target col encoded
    :param target_encoding: how to encode the target value
    :param cat_feature: list of categorical columns
    :return: a pandas dataframe with the target and categorical column encoded
    """
    try:
        dataframe = encode_target(df, target_name, new_target_name, target_encoding_dict)
        for col in cat_feature:
            encoder_cat_features(dataframe, col)  
        return dataframe
    except:
        lg.info("Something went wrong")

        
        
def encoder_cat_features(df: pd.DataFrame, feature: str) -> pd.DataFrame:
    """
    Feature engineering - compute a target encoding
    :param df: the pandas dataframe
    :param feature: a column from the dataframe
    :return: a pandas dataframe with the target encoding to the feature
    """
    try:
        feature_groups = df.groupby(feature).mean()['Churn']
        df['{0}_Churn'.format(feature)] = df[feature].map(feature_groups)
        return df
    except:
        lg.info("Something went wrong")
        
def compute_encoding_target_dict(df: pd.DataFrame, target_name: str, target_encoding_dict: dict) -> dict:
    """
    Compute dictionary for encoding target
    :param df: a pandas dataframe
    :param target_name: the target column to encode
    :param target_encoding_dict: how to encode the target value
    :return: a dictionary with <new_encode : value to be encoded>
    """
    for key, value in target_encoding_dict.items():
        return {key: df[target_name] == value}
        

def encode_target(df: pd.DataFrame, target_name: str, new_target_name: str, target_encoding_dict: dict) -> pd.Series:
    """
    Encode target variable
    :param df: a pandas dataframe
    :param target_name: the target column to encode
    :param new_target_name: name of the target col encoded
    :param target_encoding: how to encode the target value
    :return: a pandas series that corresponds to the new target column encoded
    """
    try:
        encoding_dict = compute_encoding_target_dict(df, target_name, target_encoding_dict)
        df[new_target_name] = np.select(encoding_dict.values(), encoding_dict.keys(), default=1)
        return df.drop(target_name, axis=1)
    except KeyError as error:
        lg.info('something went wrong')
        raise err


# -

dataframe = encode_target(df, 'Attrition_Flag', 'Churn', ENCODING_TARGET)

df = encoder_helper(df, 'Attrition_Flag', 'Churn', ENCODING_TARGET, CAT_COLUMNS)

df.shape

df.columns

# ### PERFORM FEATURE ENGINEERING

(X_train, X_test, y_train, y_test), feature_names = perform_feature_engineering(df, 'Churn', 0.3, 42)

print(X_train.shape, X_test.shape, len(y_train), len(y_test))

feature_names

len(feature_names)

X_test

# ### MODEL

OUTPUT_DIR_MODEL = 'models'
OUTPUT_DIR_RESULTS = 'results'

# TypeError for train_models

# + active=""
# # store errors 
# -> ValueError
#
# # for model plot roc curve
# -> TypeError
#
# # shapley values
# -> Exception
# -

lrc = LogisticRegression(max_iter=200)
rfc = RandomForestClassifier(random_state=42)

a = 'toto'

if a:
    print(f'{a}')

for model in [lrc, rfc]:
    if model == lrc:
        fitted_model, predictions = train_models(X_train, X_test, y_train, model, dump_model=True)
    elif model == rfc:
        fitted_model, predictions = train_models(X_train, X_test, y_train, model, param_grid=param_grid,
                                                 cv=cv, grid_search=True, do_probabilities=False, dump_model=True)
    evaluate_model(fitted_model, X_train, X_test, Y_test, predictions, type(fitted_model).__name__, OUTPUT_DIR_RESULTS)


def train_models(x_train_data, x_test_data, y_train_data, model, param_grid=None, cv=None,
                 grid_search=False, do_probabilities=False, output_path=None):
    try:
        if grid_search:
            gs = GridSearchCV(
                estimator=model,
                param_grid=param_grid,
                cv=cv,
                n_jobs=-1,
                verbose=2
            )
            fitted_model = gs.fit(x_train_data, y_train_data)
        else:
            fitted_model = model.fit(x_train_data, y_train_data)
        
        predictions = fitted_model.predict_proba(x_test_data) if do_probabilities else fitted_model.predict(x_test_data)
        
        if output_path:
            save_model(fitted_model, output_path)
            lg.info('Model has been dumped')
        
        return fitted_model, predictions
    except:
        lg.info('complicated life')


def evaluate_model(model_fitted, train_data, test_data, true_labels_labels, predictions, model_name, output_dir):
    safe_creation_directory(output_dir)
    store_classification_report(true_labels, predictions, model_name, output_dir)
    store_model_roc_curve(model_fitted, test_data, true_labels_labels, model_name, output_dir)
    compute_shapley_values(model_fitted, test_data, output_dir, train_data=None)
    if model_fitted.coef_.all:
        coefficients = model_fitted.coef_[0]
    coefficients = model_fitted.coef_[0] if model_fitted.coef_.all else model_fitted.best_estimator_.feature_importances_
    compute_features_importances(train_data.columns, coefficients, model_name, output_dir)


lrc = LogisticRegression(max_iter=200)

type(lrc)

logistic_args = {'x_train_data': X_train,
                'x_test_data': X_test,
                'y_train_data': y_train,
                'model': lrc,
                'output_path': OUTPUT_DIR_MODEL}

train_models(**logistic_args)

fitted_model, predictions = train_models(x_train_data=X_train, x_test_data=X_test, 
                                         y_train_data=y_train, model=lrc, output_path=OUTPUT_DIR_MODEL)


def save_model(model, output_dir: str):
    """
    Save the model fitted into a .pkl file
    :param model: model fitted - an scikit learn object in our case
    :param output_path: path where to store the model
    :return a .pkl into the output path
    """
    safe_creation_directory(output_dir)
    return joblib.dump(model, f'{output_dir}/{type(model).__name__}.pkl')


save_model(fitted_model, OUTPUT_DIR_MODEL)

OUTPUT_DIR_MODEL_TRY = 'models_bis'


def call_save_model(model, outpur_dir=None):
    if outpur_dir:
        save_model(model, outpur_dir)
        lg.info('')
    
    return type(model).__name__


call_save_model(fitted_model, OUTPUT_DIR_MODEL_TRY)

len(predictions)

if model_grid.best_estimator_:
    print('toto')
else:
    print('aaa')

# ls ../churn/models/

model_grid = joblib.load('../churn/models/GridSearchCV.pkl')

type(model_grid.best_estimator_).__name__

type(model_grid)

if isinstance(model_grid, GridSearchCV):
    print('toto')

if fitted_model.coef_.all:
    print('otot')

classification_report(y_test, predictions, output_dict=True)


def store_classification_report(true_labels, predictions, model_name, outpur_dir):
    
    report = classification_report(y_test, predictions, output_dict=True)
    return pd.DataFrame(report).transpose().to_csv(f'{output_dir}/{model_name}_classification_report.csv')


pd.DataFrame(classification_report(y_test, predictions, output_dict=True)).transpose().to_csv(f'{OUTPUT_DIR_RESULTS}/classification_report.csv')

print(classification_report(y_test, predictions))


def store_model_roc_curve(model, test_data, test_labels, model_name, output_dir):
    plt.figure(figsize=(15, 8))
    ax = plt.gca()
    plot_roc_curve(lrc, X_test, y_test, ax=ax, alpha=0.8)
    plt.savefig(f'{output_dir}/{model_name}_roc_curve.png', bbox_inches = 'tight')


plot_model_roc_curve(fitted_model, X_test, y_test)

type(fitted_model).__name__

# TO COMPLETE
TREE_EXPLAINER = ['RandomForestClassifier']
LINEAR_EXPLAINER = ['LogisticRegression', 'LinearRegression']


def compute_shapley_values(model_fitted, test_data, output_dir, model_name, train_data=None):
    plt.figure(figsize=(20, 10))
    if type(fitted_model).__name__ in LINEAR_EXPLAINER:
        masker = shap.maskers.Independent(data=train_data)
        explainer = shap.LinearExplainer(fitted_model, masker=masker)
    elif type(fitted_model).__name__ in TREE_EXPLAINER:
        explainer = shap.TreeExplainer(model_fitted)
    else:
        lg.info('Add your model type to tree, linear, gradient or deep explainer and add condition to this function')
    shap_values = explainer.shap_values(test_data)
    shap.summary_plot(shap_values, test_data, show=False)
    plt.savefig(f'{output_dir}/{model_name}_shapley_values.png', bbox_inches = 'tight')


OUTPUT_DIR_SHAPLEY = 'images/results'

compute_shapley_values(fitted_model, X_test, OUTPUT_DIR_SHAPLEY, X_train)

# compute features importances
fitted_model.get_params()

# +
len(fitted_model.coef_)

# -> could be a good test 
# assert len(fitted_model.coef_) == feature_names
# -

fitted_model.coef_[0]

fitted_model.intercept_

fitted_model.classes_

fitted_model.n_iter_

X_train.columns

feature_names

features_importances = {}
for 

print(tuple(zip(feature_names, fitted_model.coef_[0])))

print(tuple(zip(fitted_model.coef_, feature_names)))


def compute_features_importances(feature_names, coefficients, model_name, output_dir):
    plt.figure(figsize=(20, 10))
    safe_creation_directory(output_dir)
    df_features_importances = pd.DataFrame(zip(feature_names, coefficients), columns=['feature', 'coefficient']).sort_values(by=['coefficient'], ascending=False)
    #Plot Searborn bar chart
    sns.barplot(x=df_features_importances['coefficient'], y=df_features_importances['feature'])
    #Add chart labels
    plt.title(model_name + 'FEATURE IMPORTANCE')
    plt.xlabel('FEATURE IMPORTANCE')
    plt.ylabel('FEATURE NAMES')
    plt.savefig(f'{output_dir}/{model_name}_features_importances.png', bbox_inches = 'tight')


compute_features_importances(feature_names, fitted_model.coef_[0], type(fitted_model).__name__, OUTPUT_DIR_SHAPLEY)

df_feat = pd.DataFrame(zip(feature_names, fitted_model.coef_[0]), columns=['feature', 'coefficient']).sort_values(by=['coefficient'], ascending=False)

df_feat.head()

plt
sns.barplot(x=df_feat['coefficient'], y=df_feat['feature'])







StarcoderdataJupyterScriptsDedupFiltered