Update app.py

app.py

@@ -9,15 +9,17 @@ 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 tensorflow as tf
+from tensorflow.keras.models import Sequential
+from tensorflow.keras.layers import Dense, Input
+from tensorflow.keras.optimizers import Adam
 import time
 
-# Helper function to prepare data
+# 1. Helper function: Prepare data
 def prepare_data(input_data, n_samples, outliers_fraction=0.01):
-    n_outliers = max(int(outliers_fraction * n_samples), 1)  # At least 1 outlier
+    n_outliers = max(int(outliers_fraction * n_samples), 1)
     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],
@@ -29,205 +31,70 @@ def prepare_data(input_data, n_samples, outliers_fraction=0.01):
     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
+    return X
+
+# 2. Autoencoder Anomaly Detection
+def build_autoencoder(input_dim):
+    model = Sequential([
+        Dense(64, activation='relu', input_dim=input_dim),
+        Dense(32, activation='relu'),
+        Dense(64, activation='relu'),
+        Dense(input_dim, activation='sigmoid'),
+    ])
+    model.compile(optimizer=Adam(learning_rate=0.001), loss='mse')
+    return model
+
+def autoencoder_anomaly_detection(X, outliers_fraction=0.01, epochs=50):
+    model = build_autoencoder(X.shape[1])
+    model.fit(X, X, epochs=epochs, batch_size=32, verbose=0)
+    reconstruction = model.predict(X)
+    reconstruction_error = np.mean((X - reconstruction) ** 2, axis=1)
+    threshold = np.percentile(reconstruction_error, 100 * (1 - outliers_fraction))
+    y_pred = (reconstruction_error > threshold).astype(int)
+    return y_pred
+
+# 3. Function to train models and generate plots
 def train_models(input_data, outliers_fraction, n_samples, clf_name):
-    # Ensure contamination is valid
-    outliers_fraction = max(outliers_fraction, 0.01)  # At least 0.01
-
-    X, _ = prepare_data(input_data, n_samples, outliers_fraction)
-
-    # Define classifiers
+    X = prepare_data(input_data, n_samples, outliers_fraction)
     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,
-            ),
+            SGDOneClassSVM(nu=outliers_fraction, random_state=42)
         ),
         "Isolation Forest": IsolationForest(contamination=outliers_fraction, random_state=42),
         "Local Outlier Factor": LocalOutlierFactor(n_neighbors=35, contamination=outliers_fraction),
+        "Autoencoders": autoencoder_anomaly_detection(X, 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)
+    if clf_name == "Autoencoders":
+        y_pred = NAME_CLF_MAPPING[clf_name]
     else:
-        clf.fit(X)
-        y_pred = clf.predict(X)
-    t1 = time.time()
-
-    # Plotting
+        clf = NAME_CLF_MAPPING[clf_name]
+        if clf_name == "Local Outlier Factor":
+            y_pred = clf.fit_predict(X)
+        else:
+            clf.fit(X)
+            y_pred = clf.predict(X)
     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(())
+    plt.scatter(X[:, 0], X[:, 1], c=colors[(y_pred + 1) // 2], s=20)
+    plt.title(clf_name)
     return plt.gcf()
 
-# Function to detect anomalies and generate anomaly records
-def detect_anomalies(input_data, n_samples, outliers_fraction, model_name):
-    X, labels = 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[model_name]
-    if model_name == "Local Outlier Factor":
-        clf.fit(X)
-        scores = -clf.negative_outlier_factor_
-    else:
-        clf.fit(X)
-        scores = -clf.decision_function(X)
-
-    # Normalize scores to a consistent range
-    scores = (scores - scores.min()) / (scores.max() - scores.min())
-
-    # Create 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).reset_index(drop=True)
-
-    return df
-
-# Function to get anomaly samples
-def get_anomaly_samples(input_data, n_samples, outliers_fraction, model_name):
-    outliers_fraction = max(outliers_fraction, 0.01)  # Ensure fraction is valid
-    df = detect_anomalies(input_data, n_samples, outliers_fraction, model_name)
-
-    # Debugging: Check anomaly label counts
-    print("Anomaly Label Counts:", df["Anomaly_Label"].value_counts())
-
-    # Top 10 anomalies
-    top_10 = df[df["Anomaly_Label"] == "Anomaly"].head(10)
-
-    if top_10.empty:
-        print("No anomalies found in Top 10 Anomalies.")
-        top_10 = pd.DataFrame({"Message": ["No anomalies found"]})
-
-    # Middle 10 (mixed records)
-    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
-
-# Function to plot 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]
-
-    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()
-
-# Gradio Interface
+# 4. Gradio Interface
 with gr.Blocks() as demo:
-    gr.Markdown("## 🕵️‍♀️ Anomaly Detection App 🕵️‍♂️")
-
-    # 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,
+    gr.Markdown("## Anomaly Detection Comparison App")
+    input_data = gr.Radio(choices=["Central Blob", "Two Blobs", "Blob with Noise", "Moons", "Noise"], value="Moons", label="Dataset")
+    n_samples = gr.Slider(minimum=50, maximum=2000, step=50, value=500, label="Number of Samples")
+    outliers_fraction = gr.Slider(minimum=0.01, maximum=0.5, step=0.01, value=0.05, label="Fraction of Outliers")
+    input_models = gr.Radio(
+        choices=["Robust covariance", "One-Class SVM", "One-Class SVM (SGD)", "Isolation Forest", "Local Outlier Factor", "Autoencoders"],
+        value="Isolation Forest",
+        label="Select Model"
     )
+    plot = gr.Plot(label="Model Results")
+    generate_plot = gr.Button("Generate Plot")
+    generate_plot.click(fn=train_models, inputs=[input_data, outliers_fraction, n_samples, input_models], outputs=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)
-
-    '''
-    # Example Anomaly Records
-    gr.Markdown("### 3. Example Anomaly Records")
-    model_dropdown = gr.Dropdown(choices=input_models, value="Isolation Forest", label="Select Model")
-    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, model_dropdown],
-        outputs=[top_table, middle_table, bottom_table],
-    )
-    '''
-demo.launch(debug=True)
+demo.launch()