Spaces:
Sleeping
Sleeping
File size: 2,804 Bytes
f3a6f24 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | import streamlit as st
import numpy as np
from sklearn.linear_model import LogisticRegression, SGDClassifier
from sklearn.ensemble import RandomForestClassifier
from sklearn.naive_bayes import GaussianNB
from xgboost import XGBClassifier
from sklearn.metrics import (
accuracy_score, precision_score, recall_score, f1_score, roc_auc_score, log_loss,
)
import shap
@st.cache_resource
def train_tree_models(_X_train, _y_train):
models = {
"Random Forest": RandomForestClassifier(n_estimators=200, random_state=42, n_jobs=-1),
"XGBoost": XGBClassifier(
n_estimators=200, max_depth=5, learning_rate=0.1,
random_state=42, eval_metric="logloss",
),
}
for model in models.values():
model.fit(_X_train, _y_train)
return models
@st.cache_resource
def train_lr_model(_X_train_oh, _y_train):
lr = LogisticRegression(max_iter=2000, random_state=42)
lr.fit(_X_train_oh, _y_train)
return lr
@st.cache_resource
def train_nb_model(_X_train, _y_train):
"""Train Gaussian Naive Bayes - assumes features are independent."""
nb = GaussianNB()
nb.fit(_X_train, _y_train)
return nb
def evaluate_model(model, X_test, y_test) -> dict:
y_pred = model.predict(X_test)
if hasattr(model, "predict_proba"):
y_proba = model.predict_proba(X_test)[:, 1]
else:
y_proba = model.decision_function(X_test)
return {
"Accuracy": float(accuracy_score(y_test, y_pred)),
"Precision": float(precision_score(y_test, y_pred, zero_division=0)),
"Recall": float(recall_score(y_test, y_pred, zero_division=0)),
"F1 Score": float(f1_score(y_test, y_pred, zero_division=0)),
"AUC": float(roc_auc_score(y_test, y_proba)),
}
def evaluate_all_models(models: dict, X_test, y_test) -> dict:
return {name: evaluate_model(m, X_test, y_test) for name, m in models.items()}
@st.cache_resource
def get_shap_explainer(_model, _X_train):
explainer = shap.TreeExplainer(_model)
shap_values = explainer(_X_train)
return explainer, shap_values
def get_shap_single(explainer, X_single):
return explainer(X_single)
def create_sgd_model(classes=None):
model = SGDClassifier(
loss="log_loss", penalty="l2", alpha=0.0001,
random_state=42, warm_start=False,
)
if classes is not None:
model.partial_fit(
np.zeros((1, 1)), np.array([classes[0]]), classes=classes
)
return model
def evaluate_streaming(model, X_test, y_test) -> dict:
y_pred = model.predict(X_test)
y_proba = model.predict_proba(X_test)[:, 1]
return {
"Accuracy": accuracy_score(y_test, y_pred),
"F1 Score": f1_score(y_test, y_pred, zero_division=0),
"Log Loss": log_loss(y_test, y_proba),
}
|