Create app.py

app.py

@@ -0,0 +1,322 @@
+import gradio as gr
+import pandas as pd
+import numpy as np
+import joblib
+import tempfile
+from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, roc_auc_score, classification_report
+from sklearn.base import BaseEstimator, TransformerMixin
+from sklearn.preprocessing import OneHotEncoder, OrdinalEncoder, StandardScaler
+from sklearn.compose import ColumnTransformer
+
+
+# FeatureEngineer Class
+class FeatureEngineer(BaseEstimator, TransformerMixin):
+    def __init__(self): # Save the learned values during training to be used to populate the missing data in any test set
+
+        # Numeric group means (LTV excluded as it is a logical computation and will be performed on the actual test set)
+        self.rate_of_interest_means = None
+        self.interest_rate_spread_means = None
+        self.upfront_charges_means = None
+        self.overall_rate_of_interest_mean = None
+        self.overall_interest_rate_spread_mean = None
+        self.overall_upfront_charges_mean = None
+        self.income_means_by_age = None
+        self.overall_income_mean = None
+        self.term_mean = None
+        self.property_value_mean = None
+        self.dtir1_mean = None
+        self.loan_amount_mean = None
+        self.credit_score_mean = None
+
+
+        # Most frequent categorical values
+        self.categorical_features = [
+            'loan_limit', 'approv_in_adv', 'loan_type', 'loan_purpose', 'Credit_Worthiness',
+            'open_credit', 'business_or_commercial', 'Neg_ammortization', 'interest_only',
+            'lump_sum_payment', 'construction_type', 'occupancy_type', 'Secured_by', 'total_units',
+            'credit_type', 'co-applicant_credit_type', 'age', 'submission_of_application', 'Security_Type'
+        ]
+
+        self.most_frequent_cats = {}
+
+    def fit(self, X, y=None): # Learn parameters from training data only. Called only during training
+        X = X.copy()
+
+        # Calculate the numeric means for imputation
+        self.rate_of_interest_means = X.groupby(['loan_type', 'term'])['rate_of_interest'].mean() # pandas series indexed by (loan_type and term) tuples
+        self.interest_rate_spread_means = X.groupby(['loan_type', 'term'])['Interest_rate_spread'].mean()
+        self.upfront_charges_means = X.groupby(['loan_type', 'term'])['Upfront_charges'].mean()
+
+        self.overall_rate_of_interest_mean = X['rate_of_interest'].mean() # calculate the over global mean if combination not found
+        self.overall_interest_rate_spread_mean = X['Interest_rate_spread'].mean()
+        self.overall_upfront_charges_mean = X['Upfront_charges'].mean()
+
+        self.income_means_by_age = X.groupby('age')['income'].mean()
+        self.overall_income_mean = X['income'].mean()
+
+        self.term_mean = X['term'].mean().round(0)
+        self.property_value_mean = round(X['property_value'].mean(), -3)
+        self.dtir1_mean = X['dtir1'].mean().round(0)
+
+        self.loan_amount_mean = X['loan_amount'].mean() # Remaining numerical features with global mean
+        self.credit_score_mean = X['Credit_Score'].mean()
+
+        # Impute the categorical with the most frequent
+        for col in self.categorical_features:
+            if col in X.columns:
+                self.most_frequent_cats[col] = X[col].mode(dropna=True)[0]
+
+        return self
+
+    def transform(self, X): # Use during test set using self value
+        X = X.copy()
+
+        # Search for an available combination group for numeric imputations
+        def impute_feature(row, feature_name, group_means, overall_mean, group_keys):
+            if pd.isna(row[feature_name]):
+                key = tuple(row[k] for k in group_keys) # look up the group_keys such as ('Type1', 360) for ['loan_type', 'term']
+                if key in group_means:
+                    return group_means[key]
+                else:
+                    return overall_mean
+            else:
+                return row[feature_name]
+
+        # Impute rate_of_interest
+        X['rate_of_interest'] = X.apply(
+            lambda row: impute_feature(row, 'rate_of_interest',
+                                       self.rate_of_interest_means,
+                                       self.overall_rate_of_interest_mean,
+                                       ['loan_type', 'term']),
+            axis=1
+        ).round(3)
+
+        # Impute Interest_rate_spread
+        X['Interest_rate_spread'] = X.apply(
+            lambda row: impute_feature(row, 'Interest_rate_spread',
+                                       self.interest_rate_spread_means,
+                                       self.overall_interest_rate_spread_mean,
+                                       ['loan_type', 'term']),
+            axis=1
+        ).round(4)
+
+        # Impute Upfront_charges
+        X['Upfront_charges'] = X.apply(
+            lambda row: impute_feature(row, 'Upfront_charges',
+                                       self.upfront_charges_means,
+                                       self.overall_upfront_charges_mean,
+                                       ['loan_type', 'term']),
+            axis=1
+        ).round(2)
+
+        # Impute income by age
+        def impute_income(row):
+            if pd.isna(row['income']):
+                age = row['age']
+                if age in self.income_means_by_age:
+                    return self.income_means_by_age[age]
+                else:
+                    return self.overall_income_mean
+            else:
+                return row['income']
+
+        X['income'] = X.apply(impute_income, axis=1)
+        X['income'] = X['income'].fillna(self.overall_income_mean)
+        X['income'] = X['income'].round(-2)
+
+        # Impute term, property_value, dtir1, loan_amount, Credit_Score
+        X['term'] = X['term'].fillna(self.term_mean).round(0)
+        X['property_value'] = X['property_value'].fillna(self.property_value_mean).round(-3)
+        X['dtir1'] = X['dtir1'].fillna(self.dtir1_mean).round(0)
+        X['loan_amount'] = X['loan_amount'].fillna(self.loan_amount_mean)
+        X['Credit_Score'] = X['Credit_Score'].fillna(self.credit_score_mean)
+
+
+        # LTV calculation: LTV = (loan_amount / property_value) * 100
+        missing_ltv_mask = X['LTV'].isna()
+        X.loc[missing_ltv_mask, 'LTV'] = (
+            (X.loc[missing_ltv_mask, 'loan_amount'] /
+             X.loc[missing_ltv_mask, 'property_value']) * 100
+        ).round(8)
+
+        # Impute categorical with the most frequent
+        for col, most_freq in self.most_frequent_cats.items():
+            if col in X.columns:
+                X[col] = X[col].fillna(most_freq)
+
+        numeric_cols = X.select_dtypes(include=[np.number]).columns
+
+        return X
+
+
+
+# Custom Ordinal Mapper
+class OrdinalMapper(BaseEstimator, TransformerMixin):
+    def __init__(self, columns=None, mapping=None):
+        self.columns = columns
+        self.mapping = mapping
+
+    def fit(self, X, y=None):
+        return self
+
+    def transform(self, X):
+        X_ = X.copy()
+        for col in self.columns:
+            X_[col] = X_[col].map(self.mapping).fillna(-1)  # Handle unexpected or missing values
+        return X_
+
+# Define the feature lists
+ordinal_cols = ['age']
+
+binary_nominal_cols = [
+    'loan_limit', 'approv_in_adv', 'Credit_Worthiness', 'open_credit',
+    'business_or_commercial', 'Neg_ammortization', 'interest_only',
+    'lump_sum_payment', 'construction_type', 'Secured_by',
+    'co-applicant_credit_type', 'Security_Type'
+]
+
+multi_nominal_cols = [
+    'loan_type', 'loan_purpose', 'occupancy_type', 'total_units',
+    'credit_type', 'submission_of_application'
+]
+
+numeric_cols = [
+    'loan_amount', 'rate_of_interest', 'Interest_rate_spread',
+    'Upfront_charges', 'term', 'property_value', 'income',
+    'Credit_Score', 'LTV', 'dtir1'
+]
+
+# Ordinal mapping for 'age'
+condition_order = ['<25', '25-34', '35-44', '45-54', '55-64', '65-74', '>74']
+ordinal_map = {code: idx for idx, code in enumerate(condition_order)}
+
+# Define the transformers
+ordinal_transformer = OrdinalMapper(columns=ordinal_cols, mapping=ordinal_map)
+binary_transformer = OrdinalEncoder(dtype=int)  # maps binary categories to 0/1
+onehot_transformer = OneHotEncoder(drop='first', handle_unknown='ignore')
+# numeric_transformer = StandardScaler()
+
+# Building the column transformer, similar to how a pipeline works
+preprocessor = ColumnTransformer(transformers=[
+    ('ord', ordinal_transformer, ordinal_cols),
+    ('bin', binary_transformer, binary_nominal_cols),
+    ('ohe', onehot_transformer, multi_nominal_cols),
+    ('num', 'passthrough', numeric_cols)  # leave numeric untouched before passing to SMOTE
+
+])
+
+
+
+# Transformer to scale the last 10 columns after SMOTE.
+# Last 10 columns are numerical based on the number of numerical features of this dataset and the order in preprocessing
+# Transformer to scale the last `n_numeric` columns
+class ScaleLastColumns(BaseEstimator, TransformerMixin):
+    def __init__(self, n_numeric):
+        self.n_numeric = n_numeric
+        self.scaler = StandardScaler()  # Save 10 sets of mean/std for each numerical feature and apply on the test set during scaling
+
+    def fit(self, X, y=None):
+        # Assume X is NumPy array after SMOTE
+        self.scaler.fit(X[:, -self.n_numeric:])
+        return self
+
+    def transform(self, X):
+        X_ = X.copy()
+        X_[:, -self.n_numeric:] = self.scaler.transform(X_[:, -self.n_numeric:])
+        return X_
+
+
+
+# Load trained pipeline
+log_best_pipeline = joblib.load("best_logreg_pipeline.pkl")
+xgb_best_pipeline = joblib.load("best_xgb_pipeline.pkl")
+rf_best_pipeline  = joblib.load("best_rf_pipeline.pkl")
+nb_best_pipeline  = joblib.load("best_nb_pipeline.pkl")
+
+# Custom threshold
+thresholds = {
+    "Logistic Regression": 0.2680,
+    "Random Forest": 0.4850,
+    "XGBoost": None,
+    "Naive Bayes": None
+}
+
+# Map model name to pipeline
+pipelines = {
+    "Logistic Regression": log_best_pipeline,
+    "XGBoost": xgb_best_pipeline,
+    "Random Forest": rf_best_pipeline,
+    "Naive Bayes": nb_best_pipeline
+}
+
+
+def predict_from_excel(file, model_name):
+    # Load Excel file
+    test_df = pd.read_excel(file.name)
+
+    # Split into features and target
+    X_test = test_df.drop(columns=['ID', 'year', 'Gender', 'Region', 'Status'])
+    y_test = test_df['Status']
+
+    # Get pipeline
+    pipeline = pipelines[model_name]
+    
+    # Predict probabilities
+    y_proba = pipeline.predict_proba(X_test)[:, 1]
+
+    # Apply custom threshold if defined
+    thresh = thresholds.get(model_name)
+    if thresh is not None:
+        y_pred = (y_proba >= thresh).astype(int)
+    else:
+        y_pred = (y_proba >= 0.5).astype(int)
+
+    # Compute metrics
+    acc = accuracy_score(y_test, y_pred)
+    prec = precision_score(y_test, y_pred)
+    rec = recall_score(y_test, y_pred)
+    f1 = f1_score(y_test, y_pred)
+    auc = roc_auc_score(y_test, y_proba)
+    report = classification_report(y_test, y_pred, output_dict=True)
+
+    # Return metrics + results table
+    metrics = {
+        "Accuracy": round(acc, 4),
+        "Precision": round(prec, 4),
+        "Recall": round(rec, 4),
+        "F1 Score": round(f1, 4),
+        "ROC AUC": round(auc, 4),
+    }
+
+    # Add predictions to dataframe for inspection
+    results_df = test_df.copy()
+    results_df["Predicted"] = y_pred
+    results_df["Probability"] = y_proba
+
+    # Save temporary Excel file
+    temp_file = tempfile.NamedTemporaryFile(delete=False, suffix=".xlsx")
+    results_df.to_excel(temp_file.name, index=False)
+ 
+    return metrics, results_df, temp_file.name
+
+# Gradio UI
+demo = gr.Interface(
+    fn=predict_from_excel,
+    inputs=[
+        gr.File(label="Upload Excel"),
+        gr.Dropdown(
+            ["Logistic Regression", "XGBoost", "Random Forest", "Naive Bayes"],
+            label="Select Model"
+        )
+    ],
+    outputs=[
+        gr.JSON(label="Evaluation Metrics"),
+        gr.Dataframe(label="Predictions with Probabilities"),
+        gr.File(label="Download Predictions")
+    ],
+    title="Loan Default Prediction",
+    description="Upload an Excel file with loan applications to predict loan default risk."
+)
+
+if __name__ == "__main__":
+    demo.launch(share=False)