def psi(data_base, data_new, num_bins = 10): # Sort the data data_base = sorted(data_base) data_new = sorted(data_new) # Prepare the bins min_val = min(data_base[0], data_new[0]) max_val = max(data_base[-1], data_new[-1]) bins = [min_val + (max_val - min_val)*(i)/num_bins for i in range(num_bins+1)] bins[0] = min_val - 0.0001 bins[-1] = max_val + 0.0001 # Bucketize the baseline data and count the samples bins_base = pd.cut(data_base, bins = bins, labels = range(1,num_bins+1)) df_base = pd.DataFrame({'base': data_base, 'bin': bins_base}) grp_base = df_base.groupby('bin').count() grp_base['percent_base'] = grp_base['base'] / grp_base['base'].sum() # Bucketize the new data and count the samples bins_new = pd.cut(data_new, bins = bins, labels = range(1,num_bins+1)) df_new = pd.DataFrame({'new': data_new, 'bin': bins_new}) grp_new = df_new.groupby('bin').count() grp_new['percent_new'] = grp_new['new'] / grp_new['new'].sum() # Compare the bins psi_df = grp_base.join(grp_new, on = "bin", how = "inner") # Calculate the PSI psi_df['percent_base'] = psi_df['percent_base'].replace(0, 0.0001) psi_df['percent_new'] = psi_df['percent_new'].replace(0, 0.0001) psi_df['psi'] = (psi_df['percent_base'] - psi_df['percent_new']) * np.log(psi_df['percent_base'] / psi_df['percent_new']) # Return the total PSI value return np.sum(psi_df['psi'].values) # Conduct K-S Test for numeric feature psi(original_df['Credit_History_Age'], new_df1['Credit_History_Age']) psi(original_df['Credit_History_Age'], new_df2['Credit_History_Age']) psi(original_df['Credit_History_Age'], new_df3['Credit_History_Age']) # Conduct K-S Test for categorical feature psi(original_df['Payment_Behaviour'], new_df1['Payment_Behaviour']) psi(original_df['Payment_Behaviour'], new_df2['Payment_Behaviour']) psi(original_df['Payment_Behaviour'], new_df3['Payment_Behaviour'])