utils.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Thu May 25 15:04:19 2023

@author: daliagala
"""

### IMPORT LIBRARIES ###
import streamlit as st
import numpy as np
import pandas as pd 
import matplotlib.pyplot as plt
import plotly.express as px
import seaborn as sns
import math
from sklearn.decomposition import PCA
from numpy import random
from sklearn.metrics import confusion_matrix
from sklearn import svm
from sklearn import metrics
from sklearn.metrics import accuracy_score
from statkit.non_parametric import bootstrap_score
import plotly.graph_objects as go

### FUNCTIONS ###

# Function: add labels by probabilities
def assign_labels_by_probabilities(df, scores_col, label_col, probs_col, quantile=0.85, num_samples=100):
    # Sort the dataframe by scores column in descending order
    annotated = df.sort_values(by=scores_col, ascending=False)
    annotated.reset_index(drop=True, inplace=True)

    # Assign probability of 0 to bottom whatever quantile of scores
    annotated.loc[annotated[scores_col] < annotated[scores_col].quantile(quantile), probs_col] = 0

    # Count the number of NaN values in the probabilities column - how many scores left
    num_nans = annotated[probs_col].isna().sum()

    # Write a linear function to assign increasing probabilities
    function = np.linspace(start=0.99, stop=0.01, num=num_nans)
    sum_func = np.sum(function)
    function = function/sum_func
    function = pd.Series(function)

    # Assign increasing probabilities to all NaNs
    annotated[probs_col].fillna(value=function, inplace=True)

    # Randomly select users based on assigned probabilities
    selected = random.choice(annotated["user_id"], size=num_samples,replace=False, p=annotated[probs_col])
    annotated[label_col] = 0
    annotated.loc[annotated['user_id'].isin(selected), label_col] = 1
    
    return annotated

# A function to remove protected characteristics and useless data
def drop_data(df):
  labels_to_drop = ["user_id", "age", "gender", "education level", "country", "test_run_id", "battery_id", "time_of_day", 
                    "model_A_scores", "model_B_scores", "Model_A_probabilities", "Model_B_probabilities"]
  clean = df.drop(labels_to_drop, axis = 1)
  return clean

# A function to train an SVM
def train_and_predict(name, X_train, X_test, y_train, y_test, kernel='poly'):
    # Define X and Y data
    name=name

    # Create a svm Classifier
    clf = svm.SVC(kernel=kernel, probability = True) # Polynomial Kernel

    # Train the model using the training sets
    model = clf.fit(X_train, y_train.values.ravel())

    # Predict the response for test dataset
    y_pred = clf.predict(X_test)
    cm = confusion_matrix(y_pred, y_test)

    # Predict the probabilities for test dataset
    y_pred_proba = clf.predict_proba(X_test)

    # Change class probabilities into 2 flat numpy arrays
    array1 = y_pred_proba[:, 0].reshape(-1, 1).flatten()
    array2 = y_pred_proba[:, 1].reshape(-1, 1).flatten()

    # Append predictions to X_test dataframe
    X_eval = X_test.copy(deep=True)
    X_eval[f"Predicted_%s" % name] = y_pred

    # Append probability predictions to X_test dataframe
    X_eval[f"Prob_0_%s" % name] = array1
    X_eval[f"Prob_1_%s" % name] = array2

    # Mark which data was used for training
    X_tr = X_train.copy(deep = True)
    X_tr[f"Predicted_%s" % name] = "train"

    # Concatenate training and test data
    X_full = pd.concat([X_eval, X_tr])

    # Reset index and retain old index to be able to get back to sensitive data
    X_full = X_full.reset_index()

    # Calculate accuracy
    accuracy = metrics.accuracy_score(y_test, y_pred)

    # Calculate precision
    precision = metrics.precision_score(y_test, y_pred)

    # Calculate recall
    recall = metrics.recall_score(y_test, y_pred)

    baseline_accuracy = bootstrap_score(y_test, y_pred, metric=accuracy_score, random_state=5)

    return accuracy, precision, recall, X_full, cm, baseline_accuracy

# A function to display proportional representation of protected characteristics

def display_proportional(data, protected_characteristic, which_model):
  if protected_characteristic == 'age':
    bins= [18,20,30,40,50,60,70,80,90]
    labels = ['18-20','21-30','31-40','41-50','51-60','61-70','71-80','81-90']
    data['age_bins'] = pd.cut(data['age'], bins=bins, labels=labels, right=False)
    data_all = data.loc[data[which_model] != "train"]
    info_all = data_all["age_bins"].value_counts()
    data_sel = data.loc[data[which_model] == 1]
    info_sel = data_sel["age_bins"].value_counts()
    dict_all = dict(info_all)
    dict_sel = dict(info_sel)
    for key in dict_all.keys():
      if key not in dict_sel.keys():
        dict_sel[key] = 0
    dict_percentage = {k: round(((dict_sel[k] / dict_all[k])*100), 2) for k in dict_all if k in dict_sel}
    values = []
    for label in labels:
      values.append(dict_percentage[label])
    fig = px.bar(x = labels, y = values, text_auto='.2s')
    fig.update_layout(yaxis_title="percentage value", xaxis_title="category")
    st.plotly_chart(fig, use_container_width=True)
  else:
    data_all = data.loc[data[which_model] != "train"]
    info_all = data_all[protected_characteristic].value_counts()
    data_sel = data.loc[data[which_model] == 1]
    info_sel = data_sel[protected_characteristic].value_counts()
    dict_all = dict(info_all)
    dict_sel = dict(info_sel)
    for key in dict_all.keys():
      if key not in dict_sel.keys():
        dict_sel[key] = 0
    dict_percentage = {k: round(((dict_sel[k] / dict_all[k])*100), 2) for k in dict_all if k in dict_sel}
    names = list(dict_percentage.keys())
    values = list(dict_percentage.values())
    fig = px.bar(x = names, y = values, text_auto='.2s')
    fig.update_layout(yaxis_title="percentage value", xaxis_title="category")
    st.plotly_chart(fig, use_container_width=True)
    
# A function to plot data depending on data type
def plot_data(data, protected_characteristic, colour_code):

  if protected_characteristic == 'age':
    mean = data.loc[:, 'age'].mean().round(2)
    st.markdown(f':green[The mean age for this group is %s years.]' % mean)
    bin_width= 1
    nbins = math.ceil((data["age"].max() - data["age"].min()) / bin_width)
    fig = px.histogram(data, x='age', nbins=nbins)
    fig.update_layout(margin=dict(l=20, r=20, t=30, b=0))
    st.plotly_chart(fig, use_container_width=True)

  elif protected_characteristic == 'education level':
    data = data[protected_characteristic].value_counts().to_frame().reset_index()
    fig = px.bar(data, x=data.iloc[:,1], y=data.iloc[:,0], orientation='h',color=data.iloc[:,1])
    fig.update_layout(margin=dict(l=20, r=20, t=30, b=0))
    fig.update_coloraxes(showscale=False)
    fig.update_layout(yaxis_title=None)
    fig.update_layout(xaxis_title=None)
    st.plotly_chart(fig, use_container_width=True)

  else:
    data = data[protected_characteristic].value_counts().to_frame().reset_index()
    fig = px.pie(data, values=data.iloc[:,1], names=data.iloc[:,0], color = data.iloc[:,0],
                  height=300, width=200, color_discrete_map=colour_code)
    fig.update_layout(margin=dict(l=20, r=20, t=30, b=0))
    st.plotly_chart(fig, use_container_width=True)

# A function to run PCA with custom no of components using sklearn
def run_PCA(df, drop_1, retain_this, n):
  df_clean = df.drop(columns = [drop_1, retain_this, "index"])
  labels = list(df_clean.columns)
  pca = PCA(n_components=n)
  principalComponents = pca.fit_transform(df_clean)
  if n == 2:
    principalDf = pd.DataFrame(data = principalComponents, columns = ['principal component 1', 'principal component 2'])
  else:
    principalDf = pd.DataFrame(data = principalComponents, columns = ['principal component 1', 'principal component 2', 'principal component 3'])
  finalDf = pd.concat([principalDf, df[[retain_this]]], axis = 1)
  finalDf2 = finalDf.rename(columns = {retain_this : 'target'})
  coeff = np.transpose(pca.components_[0:2, :])
  return pca, finalDf2, labels, coeff, principalComponents

# Plot confusion matrices as heatmaps
def create_confusion_matrix_heatmap(confusion_matrix, model):
    group_names = ['True Negative (TN)','False Positive (FP)','False Negative (FN)','True Positive (TP)']
    group_counts = ["{0:0.0f}".format(value) for value in confusion_matrix.flatten()]
    group_percentages = ["{0:.2%}".format(value) for value in confusion_matrix.flatten()/np.sum(confusion_matrix)]
    labels = [f"{v1}<br>{v2}<br>{v3}" for v1, v2, v3 in zip(group_names,group_counts,group_percentages)]
    labels = np.asarray(labels).reshape(2,2)
    
    layout = {
        "title": f"Confusion Matrix, {model}", 
        "xaxis": {
            "title": "Predicted value", 
            "tickmode" : 'array', 
            "tickvals" : [0, 1], 
            "ticktext" : ["0", "1"]}, 
        "yaxis": {
            "title": "Actual value", 
            "tickmode" : 'array', 
            "tickvals" : [0, 1], 
            "ticktext" : ["0", "1"]},
        }
    fig = go.Figure(data=go.Heatmap(
                    z=confusion_matrix,
                    text=labels,
                    texttemplate="%{text}",
                    textfont={"size":15}), layout = layout)
    st.plotly_chart(fig, use_container_width = True)

# Display model metrics as tables
def plot_conf_rates(confusion_matrix):
  TN = confusion_matrix[0,0]
  TP = confusion_matrix[1,1]
  FP = confusion_matrix[0,1]
  FN = confusion_matrix[1,0]


  # Sensitivity, hit rate, recall, or true positive rate
  TPR = TP/(TP+FN)
  # Specificity or true negative rate
  TNR = TN/(TN+FP) 
  # Precision or positive predictive value
  PPV = TP/(TP+FP)
  # Negative predictive value
  NPV = TN/(TN+FN)
  # Fall out or false positive rate
  FPR = FP/(FP+TN)
  # False negative rate
  FNR = FN/(TP+FN)
  # False discovery rate
  FDR = FP/(TP+FP)

  # Overall accuracy
  ACC = (TP+TN)/(TP+FP+FN+TN)
  d = {'Measure': ['True Positive Rate', 'True Negative Rate', 'Positive Predictive Value', 'Negative Predictive Value', 'False Positive Rate', 'False Negative Rate', 'False Discovery Rate'], 
  'Equation' : ['TPR = TP/(TP+FN)', 'TNR = TN/(TN+FP)', 'PPV = TP/(TP+FP)', 'NPV = TN/(TN+FN)', 'FPR = FP/(FP+TN)', 'FNR = FN/(TP+FN)', 'FDR = FP/(TP+FP)'], 
  'Score': [TPR, TNR, PPV, NPV, FPR, FNR, FDR]}
  return d