Spaces:
Sleeping
Sleeping
| import io | |
| import numpy as np | |
| import streamlit as st | |
| from collections import Counter | |
| from sklearn import metrics | |
| from sklearn.preprocessing import StandardScaler | |
| from sklearn.cluster import KMeans | |
| from imblearn.over_sampling import RandomOverSampler, SMOTE, ADASYN | |
| from joblib import dump | |
| from sklearn.metrics import roc_curve, silhouette_score, calinski_harabasz_score, davies_bouldin_score, f1_score, r2_score, mean_squared_error, mean_absolute_error | |
| from sklearn.model_selection import train_test_split | |
| def split_data(X, Y, test_size = 0.2, random_state = 42, perform_pca = False): | |
| """ | |
| Splits the dataset into training and testing sets, optionally standardizing the data if PCA is not performed. | |
| :param X: Feature matrix. | |
| :param Y: Target vector. | |
| :param test_size: Proportion of the dataset to include in the test split. | |
| :param random_state: Controls the shuffling applied to the data before applying the split. | |
| :param perform_pca: Has PCA been performed or not. If not, standardizes the data. | |
| :return: A tuple containing split and optionally transformed datasets: X_train, X_test, Y_train, Y_test. | |
| """ | |
| X_train, X_test, Y_train, Y_test = train_test_split(X, Y, test_size=test_size, random_state=random_state) | |
| if not perform_pca: | |
| scaler = StandardScaler() | |
| X_train = scaler.fit_transform(X_train) | |
| X_test = scaler.transform(X_test) | |
| return X_train, X_test, Y_train, Y_test | |
| def check_and_balance(X, Y, balance_threshold=0.5, method=1): | |
| """ | |
| Check if the dataset is imbalanced and perform oversampling if necessary using RandomOverSampler, SMOTE, or ADASYN. | |
| Args: | |
| X (DataFrame): Feature set. | |
| Y (Series): Target variable. | |
| balance_threshold (float): Threshold for class balance. | |
| method (int): Method for oversampling. Options are 'random', 'smote', or 'adasyn'. | |
| Returns: | |
| X_resampled, Y_resampled (DataFrame/Series): Resampled data if imbalance is detected, else original data. | |
| """ | |
| try: | |
| # Check the distribution of the target variable | |
| class_distribution = Counter(Y) | |
| # Determine if the dataset is imbalanced | |
| min_class_samples = min(class_distribution.values()) | |
| max_class_samples = max(class_distribution.values()) | |
| is_imbalanced = min_class_samples / max_class_samples < balance_threshold | |
| if is_imbalanced and method != 4: | |
| if method == 1: | |
| oversampler = RandomOverSampler(random_state=0) | |
| elif method == 2: | |
| oversampler = SMOTE(random_state=0) | |
| elif method == 3: | |
| oversampler = ADASYN(random_state=0) | |
| X_resampled, Y_resampled = oversampler.fit_resample(X, Y) | |
| return X_resampled, Y_resampled | |
| else: | |
| return X, Y | |
| except Exception as e: | |
| st.error("The target attribute may be continuous. Please check the data type.") | |
| st.stop() | |
| def estimate_optimal_clusters(df): | |
| """ | |
| Estimates the optimal number of clusters for KMeans clustering using the elbow method and silhouette scores. | |
| :param df: DataFrame containing the dataset to cluster. | |
| :return: The estimated optimal number of clusters. | |
| """ | |
| sse = {} | |
| for k in range(2, 11): | |
| kmeans = KMeans(n_clusters=k, random_state=42).fit(df) | |
| sse[k] = kmeans.inertia_ | |
| # Find the elbow point: compute the first and second differences of the SSE | |
| sse_values = list(sse.values()) | |
| first_diff = np.diff(sse_values) # first difference | |
| second_diff = np.diff(first_diff) # second difference | |
| knee_point = np.argmax(second_diff) + 2 | |
| # find the optimal number of clusters around the knee point | |
| silhouette_avg_scores = {} | |
| for k in range(knee_point - 1, knee_point + 2): | |
| if k >= 2: # make sure k is at least 2 | |
| kmeans = KMeans(n_clusters=k, random_state=42).fit(df) | |
| silhouette_avg_scores[k] = silhouette_score(df, kmeans.labels_) | |
| # Find the optimal number of clusters based on the highest average silhouette score | |
| optimal_clusters = max(silhouette_avg_scores, key=silhouette_avg_scores.get) | |
| return optimal_clusters | |
| def calculate_f1_score(model, X_test, Y_test, binary_classification=True): | |
| """ | |
| Calculates the F1 score for the predictions made by a model on a test set. | |
| The function supports both binary and multi-class settings by adjusting the 'average' parameter in the f1_score calculation. | |
| :param model: The trained machine learning model used for predictions. | |
| :param X_test: The feature matrix for the test set. | |
| :param Y_test: The true labels for the test set. | |
| :param binary_classification: If True, calculates the F1 score for binary classification. Otherwise, calculates for multi-class classification using the 'macro' average. | |
| :return: The F1 score of the model predictions. | |
| """ | |
| y_pred = model.predict(X_test) | |
| if binary_classification: | |
| f1 = f1_score(Y_test, y_pred, average='binary') | |
| else: | |
| f1 = f1_score(Y_test, y_pred, average='macro') | |
| return f1 | |
| def model_score(model, X_test, Y_test): | |
| """ | |
| Calculate the model score for classification models. | |
| """ | |
| score = model.score(X_test, Y_test) | |
| return score | |
| def fpr_and_tpr(model, X_test, Y_test): | |
| """ | |
| Calculate the false positive rate and true positive rate for classification models. | |
| """ | |
| Y_pred = model.predict_proba(X_test)[:, 1] | |
| fpr, tpr, _ = roc_curve(Y_test, Y_pred) | |
| return fpr, tpr | |
| def auc(fpr, tpr): | |
| """ | |
| Calculate the area under the ROC curve for classification models. | |
| """ | |
| auc = metrics.auc(fpr, tpr) | |
| return auc | |
| def calculate_silhouette_score(X, labels): | |
| """ | |
| Calculate the silhouette score for clustering models. | |
| """ | |
| return silhouette_score(X, labels) | |
| def calculate_calinski_harabasz_score(X, labels): | |
| """ | |
| Calculate the calinski harabasz score for clustering models. | |
| """ | |
| return calinski_harabasz_score(X, labels) | |
| def calculate_davies_bouldin_score(X, labels): | |
| """ | |
| Calculate the davies bouldin score for clustering models. | |
| """ | |
| return davies_bouldin_score(X, labels) | |
| def gmm_predict(X, model): | |
| """ | |
| Get the predicted labels for a GMM model. | |
| """ | |
| labels = model.predict(X) | |
| return labels | |
| def calculate_r2_score(y_pred, Y_test): | |
| """ | |
| Calculate the r2 score for regression models. | |
| """ | |
| r2 = r2_score(Y_test, y_pred) | |
| return r2 | |
| def calculate_mse_and_rmse(y_pred, Y_test): | |
| """ | |
| Calculate the mean squared error and root mean squared error for regression models. | |
| """ | |
| mse = mean_squared_error(Y_test, y_pred) | |
| rmse = np.sqrt(mse) | |
| return mse, rmse | |
| def calculate_mae(y_pred, Y_test): | |
| """ | |
| Calculate the mean absolute error for regression models. | |
| """ | |
| mae = mean_absolute_error(Y_test, y_pred) | |
| return mae | |
| def save_model(model): | |
| """ | |
| Serializes a machine learning model into a binary format using joblib's dump function and stores it in a BytesIO buffer. | |
| """ | |
| buffer = io.BytesIO() | |
| dump(model, buffer) | |
| buffer.seek(0) | |
| return buffer.getvalue() |