app.py

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 time

# Function to train models and generate plots
def train_models(input_data, outliers_fraction, n_samples, clf_name):
    # Prepare data
    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),
    }
    
    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),
    }
    
    X = DATA_MAPPING[input_data]
    rng = np.random.RandomState(42)
    X = np.concatenate([X, rng.uniform(low=-6, high=6, size=(n_outliers, 2))], axis=0)

    xx, yy = np.meshgrid(np.linspace(-7, 7, 150), np.linspace(-7, 7, 150))
    clf = NAME_CLF_MAPPING[clf_name]

    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()

    # Plot
    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()

# Gradio Interface
description = "Compare how different anomaly detection algorithms perform on various datasets."
title = "🕵️‍♀️ Compare Anomaly Detection Algorithms 🕵️‍♂️"

with gr.Blocks() as demo:
    gr.Markdown(f"## {title}")
    gr.Markdown(description)
    
    # Inputs
    input_data = gr.Radio(
        choices=["Central Blob", "Two Blobs", "Blob with Noise", "Moons", "Noise"],
        value="Moons",
        label="Dataset"
    )
    n_samples = gr.Slider(minimum=10, maximum=10000, step=25, value=500, label="Number of Samples")
    outliers_fraction = gr.Slider(minimum=0.001, maximum=0.999, step=0.1, value=0.2, label="Fraction of Outliers")
    
    # Models and their plots in a row
    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))

    # Update function
    def update(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

    # Set change triggers
    inputs = [input_data, outliers_fraction, n_samples]
    demo_outputs = [plot for _, plot in plots]
    input_data.change(fn=update, inputs=inputs, outputs=demo_outputs)
    n_samples.change(fn=update, inputs=inputs, outputs=demo_outputs)
    outliers_fraction.change(fn=update, inputs=inputs, outputs=demo_outputs)



# Function to generate interactive feature scatter plots
def plot_interactive_feature_scatter(input_data, feature_x, feature_y, n_samples):
    # Generate data based on the selected dataset
    if input_data == "Moons":
        data, _ = make_moons(n_samples=n_samples, noise=0.05)
    else:
        data, _ = make_blobs(n_samples=n_samples, random_state=0)

    # Simulate feature selection by indexing
    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 for anomaly examples (Optional feature row)
def get_anomaly_samples():
    """Returns formatted top, middle, and bottom 10 records based on anomaly score."""
    sorted_df = df.sort_values("Anomaly_Score", ascending=False)
    
    # Top 10 anomalies
    top_10 = sorted_df[sorted_df["Anomaly_Label"] == "Anomaly"].head(10)
    
    # Middle 10 (mix of anomalies and normal)
    mid_start = len(sorted_df) // 2 - 50  # Get a broader middle slice
    middle_section = sorted_df.iloc[mid_start: mid_start + 100]  # Consider a larger middle slice
    middle_anomalies = middle_section[middle_section["Anomaly_Label"] == "Anomaly"].sample(n=5, random_state=42)
    middle_normals = middle_section[middle_section["Anomaly_Label"] == "Normal"].sample(n=5, random_state=42)
    middle_10 = pd.concat([middle_anomalies, middle_normals]).sort_values("Anomaly_Score", ascending=False)
    
    # Bottom 10 normal records
    bottom_10 = sorted_df[sorted_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.")

    # Anomaly Detection Comparison
    gr.Markdown("### 1. Compare Anomaly Detection Algorithms")
    input_data = gr.Radio(
        choices=["Central Blob", "Two Blobs", "Blob with Noise", "Moons", "Noise"],
        value="Moons",
        label="Dataset"
    )
    n_samples = gr.Slider(
        minimum=10, maximum=10000, step=25, value=500, label="Number of Samples"
    )
    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)

    # Interactive Feature Scatter Plot
    gr.Markdown("### 2. Interactive Feature Scatter Plot")
    feature_x = gr.Dropdown(choices=["Feature1", "Feature2"], value="Feature1", label="Feature 1")
    feature_y = gr.Dropdown(choices=["Feature1", "Feature2"], value="Feature2", label="Feature 2")
    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,
    )

    with gr.Tab("Anomaly Samples"):
        gr.HTML("<h3 style='text-align: center; font-size: 18px; font-weight: bold;'>Top 10 Records (Anomalies)</h3>")
        top_table = gr.Dataframe(label="Top 10 Records")
        
        gr.HTML("<h3 style='text-align: center; font-size: 18px; font-weight: bold;'>Middle 10 Records (Mixed)</h3>")
        middle_table = gr.Dataframe(label="Middle 10 Records")
        
        gr.HTML("<h3 style='text-align: center; font-size: 18px; font-weight: bold;'>Bottom 10 Records (Normal)</h3>")
        bottom_table = gr.Dataframe(label="Bottom 10 Records")
        
        anomaly_samples_button = gr.Button("Show Anomaly Samples")
        anomaly_samples_button.click(
            get_anomaly_samples, 
            outputs=[top_table, middle_table, bottom_table]
        )
    )

demo.launch(debug=True)