import numpy as np import matplotlib.pyplot as plt from sklearn import svm from sklearn.covariance import EllipticEnvelope from sklearn.ensemble import IsolationForest from sklearn.neighbors import LocalOutlierFactor from sklearn.linear_model import SGDOneClassSVM from sklearn.kernel_approximation import Nystroem from sklearn.pipeline import make_pipeline from sklearn.datasets import make_blobs, make_moons import gradio as gr import pandas as pd import time # Helper function to prepare data def prepare_data(input_data, n_samples, outliers_fraction=0.0): n_outliers = int(outliers_fraction * n_samples) n_inliers = n_samples - n_outliers blobs_params = dict(random_state=0, n_samples=n_inliers, n_features=2) DATA_MAPPING = { "Central Blob": make_blobs(centers=[[0, 0], [0, 0]], cluster_std=0.5, **blobs_params)[0], "Two Blobs": make_blobs(centers=[[2, 2], [-2, -2]], cluster_std=[0.5, 0.5], **blobs_params)[0], "Blob with Noise": make_blobs(centers=[[2, 2], [-2, -2]], cluster_std=[1.5, 0.3], **blobs_params)[0], "Moons": 4.0 * (make_moons(n_samples=n_samples, noise=0.05, random_state=0)[0] - np.array([0.5, 0.25])), "Noise": 14.0 * (np.random.RandomState(42).rand(n_samples, 2) - 0.5), } X = DATA_MAPPING[input_data] rng = np.random.RandomState(42) outliers = rng.uniform(low=-6, high=6, size=(n_outliers, 2)) X = np.concatenate([X, outliers], axis=0) labels = np.array(["Normal"] * len(X)) labels[-len(outliers):] = "Anomaly" return X, labels # Function to train models and generate plots def train_models(input_data, outliers_fraction, n_samples, clf_name): X, _ = prepare_data(input_data, n_samples, outliers_fraction) # Define classifiers NAME_CLF_MAPPING = { "Robust covariance": EllipticEnvelope(contamination=outliers_fraction), "One-Class SVM": svm.OneClassSVM(nu=outliers_fraction, kernel="rbf", gamma=0.1), "One-Class SVM (SGD)": make_pipeline( Nystroem(gamma=0.1, random_state=42, n_components=150), SGDOneClassSVM( nu=outliers_fraction, shuffle=True, fit_intercept=True, random_state=42, tol=1e-6, ), ), "Isolation Forest": IsolationForest(contamination=outliers_fraction, random_state=42), "Local Outlier Factor": LocalOutlierFactor(n_neighbors=35, contamination=outliers_fraction), } clf = NAME_CLF_MAPPING[clf_name] xx, yy = np.meshgrid(np.linspace(-7, 7, 150), np.linspace(-7, 7, 150)) t0 = time.time() if clf_name == "Local Outlier Factor": y_pred = clf.fit_predict(X) else: clf.fit(X) y_pred = clf.predict(X) t1 = time.time() # Plotting plt.figure(figsize=(5, 5)) if clf_name != "Local Outlier Factor": Z = clf.predict(np.c_[xx.ravel(), yy.ravel()]) Z = Z.reshape(xx.shape) plt.contour(xx, yy, Z, levels=[0], linewidths=2, colors="black") colors = np.array(["#377eb8", "#ff7f00"]) plt.scatter(X[:, 0], X[:, 1], s=30, color=colors[(y_pred + 1) // 2]) plt.title(f"{clf_name} ({t1 - t0:.2f}s)") plt.xlim(-7, 7) plt.ylim(-7, 7) plt.xticks(()) plt.yticks(()) return plt.gcf() # Function to generate feature scatter plots def plot_interactive_feature_scatter(input_data, feature_x, feature_y, n_samples): data, _ = prepare_data(input_data, n_samples) x_data = data[:, 0] if feature_x == "Feature1" else data[:, 1] y_data = data[:, 1] if feature_y == "Feature2" else data[:, 0] # Generate scatter plot plt.figure(figsize=(6, 6)) plt.scatter(x_data, y_data, alpha=0.8, c="blue", s=20, label="Features") plt.title(f"Feature Interaction Scatter Plot - {feature_x} vs {feature_y}") plt.xlabel(feature_x) plt.ylabel(feature_y) plt.legend() return plt.gcf() # Function to simulate anomaly samples def get_anomaly_samples(input_data, n_samples, outliers_fraction): # Prepare data with labels X, labels = prepare_data(input_data, n_samples, outliers_fraction) # Assign anomaly scores with higher values for anomalies rng = np.random.default_rng(42) scores = np.where( labels == "Anomaly", rng.uniform(0.7, 1.0, len(labels)), # Higher scores for anomalies rng.uniform(0.0, 0.7, len(labels)), # Lower scores for normals ) # Create a DataFrame df = pd.DataFrame({ "Feature1": X[:, 0], "Feature2": X[:, 1], "Anomaly_Score": scores, "Anomaly_Label": labels, }) # Sort by Anomaly Score in descending order df = df.sort_values("Anomaly_Score", ascending=False) # Round values to 3 decimal places df = df.round({"Feature1": 3, "Feature2": 3, "Anomaly_Score": 3}) # Top 10 anomalies top_10 = df[df["Anomaly_Label"] == "Anomaly"].head(10) # Middle 10 (mixed) mid_start = len(df) // 2 - 5 middle_10 = df.iloc[mid_start: mid_start + 10] # Bottom 10 normals bottom_10 = df[df["Anomaly_Label"] == "Normal"].tail(10) return top_10, middle_10, bottom_10 # Gradio Interface with gr.Blocks() as demo: # App Title and Description gr.Markdown("## 🕵️‍♀️ Anomaly Detection App 🕵️‍♂️") gr.Markdown("Explore anomaly detection models, feature interactions, and anomaly examples.") # Interactive Feature Scatter Plot gr.Markdown("### 1. Interactive Feature Scatter Plot") input_data = gr.Radio( choices=["Central Blob", "Two Blobs", "Blob with Noise", "Moons", "Noise"], value="Moons", label="Dataset" ) feature_x = gr.Dropdown(choices=["Feature1", "Feature2"], value="Feature1", label="Feature 1") feature_y = gr.Dropdown(choices=["Feature1", "Feature2"], value="Feature2", label="Feature 2") n_samples = gr.Slider(minimum=10, maximum=10000, step=25, value=500, label="Number of Samples") scatter_plot_button = gr.Button("Generate Scatter Plot") scatter_plot = gr.Plot(label="Feature Scatter Plot") scatter_plot_button.click( fn=plot_interactive_feature_scatter, inputs=[input_data, feature_x, feature_y, n_samples], outputs=scatter_plot, ) # Compare Anomaly Detection Algorithms gr.Markdown("### 2. Compare Anomaly Detection Algorithms") outliers_fraction = gr.Slider(minimum=0.001, maximum=0.999, step=0.1, value=0.2, label="Fraction of Outliers") input_models = ["Robust covariance", "One-Class SVM", "One-Class SVM (SGD)", "Isolation Forest", "Local Outlier Factor"] plots = [] with gr.Row(): for model_name in input_models: plot = gr.Plot(label=model_name) plots.append((model_name, plot)) def update_anomaly_comparison(input_data, outliers_fraction, n_samples): results = [] for clf_name, plot in plots: fig = train_models(input_data, outliers_fraction, n_samples, clf_name) results.append(fig) return results anomaly_inputs = [input_data, outliers_fraction, n_samples] anomaly_outputs = [plot for _, plot in plots] input_data.change(fn=update_anomaly_comparison, inputs=anomaly_inputs, outputs=anomaly_outputs) n_samples.change(fn=update_anomaly_comparison, inputs=anomaly_inputs, outputs=anomaly_outputs) outliers_fraction.change(fn=update_anomaly_comparison, inputs=anomaly_inputs, outputs=anomaly_outputs) # Anomaly Samples Tab gr.Markdown("### 3. Example Anomaly Records") top_table = gr.Dataframe(label="Top 10 Anomalies") middle_table = gr.Dataframe(label="Middle 10 Records") bottom_table = gr.Dataframe(label="Bottom 10 Normals") anomaly_samples_button = gr.Button("Show Anomaly Samples") anomaly_samples_button.click( fn=get_anomaly_samples, inputs=[input_data, n_samples, outliers_fraction], outputs=[top_table, middle_table, bottom_table], ) demo.launch(debug=True)