Deploy backend Flask app

bot_detector_api.py

@@ -1,25 +1,19 @@
 import numpy as np
-import pandas as pd
 import joblib
+import pandas as pd
 import shap
-import matplotlib.pyplot as plt
-import base64
 from flask import Flask, request, jsonify
 from urllib.parse import urlparse, parse_qs
-from io import BytesIO
-
-# Initialize SHAP
-shap.initjs()
 
-# Initialize Flask app
+# Initialize Flask
 app = Flask("Bot detector")
 
-# Health check route
+# Health check
 @app.get('/')
 def home():
     return "✅ Welcome to the Bot Prediction API!"
 
-# Load models and transformers
+# Load models and utilities
 model = joblib.load("model.joblib")
 encoders = joblib.load("encoders.joblib")
 scaler = joblib.load("scaler.joblib")
@@ -27,8 +21,8 @@ if_model = joblib.load("best_if_model.joblib")
 svm_model = joblib.load("best_svm_model.joblib")
 feature_names = joblib.load("feature_names.joblib")
 
-# Instantiate SHAP explainer once
-explainer = shap.Explainer(model, feature_names=feature_names)
+# SHAP explainer (tree-based models)
+explainer = shap.TreeExplainer(model)
 
 def parse_url_params(url):
     try:
@@ -75,57 +69,71 @@ def prepare_features(row_dict):
 
     return df[feature_names]
 
-def get_shap_explanation(X_row, encoders=None, top_n=10, class_index=1):
-    shap_vals = explainer(X_row)
-    shap_value = shap_vals[0].values[:, class_index]
-    base_val = shap_vals[0].base_values[class_index]
+def generate_shap_bot_attack_paragraph(index, shap_values, X, encoders=None, class_index=1, top_n=10):
+    shap_vals = shap_values.values[index, :, class_index]
+    base_val = shap_values.base_values[index, class_index]
+    x_vals = X.iloc[index]
+    feature_names = X.columns
 
-    decoded = {}
-    for col in X_row.columns:
-        val = X_row.iloc[0][col]
+    # Decode encoded values
+    decoded_vals = {}
+    for col in feature_names:
+        val = x_vals[col]
         try:
-            decoded[col] = encoders[col].inverse_transform([int(val)])[0] if encoders and col in encoders else val
+            if encoders and col in encoders:
+                decoded_vals[col] = encoders[col].inverse_transform([int(val)])[0]
+            else:
+                decoded_vals[col] = val
         except:
-            decoded[col] = val
+            decoded_vals[col] = val
 
-    contribs = list(zip(X_row.columns, [decoded[c] for c in X_row.columns], shap_value))
-    contribs = sorted(contribs, key=lambda x: abs(x[2]), reverse=True)[:top_n]
+    # Sort features by absolute SHAP impact
+    feature_contribs = list(zip(feature_names, decoded_vals.values(), shap_vals))
+    feature_contribs = sorted(feature_contribs, key=lambda x: abs(x[2]), reverse=True)[:top_n]
 
-    pos, neg = [], []
-    for f, v, s in contribs:
-        line = f"- {f:20} = {str(v):<15} ➤ impact: {abs(s):.4f}"
-        (pos if s > 0 else neg).append(line)
+    # Separate positive and negative impacts
+    positive_impacts = []
+    negative_impacts = []
 
-    explanation = f"Base value: {base_val:.4f} → Final: {base_val + np.sum(shap_value):.4f}\n\n"
-    if pos:
-        explanation += "🔺 Positive Influences:\n" + "\n".join(pos) + "\n\n"
-    if neg:
-        explanation += "🔻 Negative Influences:\n" + "\n".join(neg)
+    for fname, fval, sval in feature_contribs:
+        line = f"  - {fname:20} = {str(fval):<20} contributed {abs(sval):.4f}"
+        if sval > 0:
+            positive_impacts.append(line)
+        elif sval < 0:
+            negative_impacts.append(line)
 
-    # Waterfall plot
-    fig = plt.figure()
-    shap.plots.waterfall(shap_vals[0][:, class_index], max_display=top_n, show=False)
-    buf = BytesIO()
-    plt.savefig(buf, format="png", bbox_inches='tight')
-    plt.close(fig)
-    buf.seek(0)
-    image_base64 = base64.b64encode(buf.getvalue()).decode("utf-8")
+    final_pred_val = base_val + np.sum(shap_vals)
 
-    return explanation, image_base64
+    explanation = f"\n==== SHAP Explanation for Bot Attack Classification ====\n"
+    explanation += f"Base model prediction (Bot): {base_val:.2f}\n"
+    explanation += f"Final model output         : {final_pred_val:.2f}\n\n"
 
-@app.post('/v1/predict')
-def predict():
-    row = request.get_json()
-    X = prepare_features(row)
+    if positive_impacts:
+        explanation += "🔺 Factors that INCREASED Bot Likelihood:\n" + "\n".join(positive_impacts) + "\n\n"
 
-    pred_prob = model.predict_proba(X)[0][1]
-    pred_label = int(pred_prob >= 0.5)
+    if negative_impacts:
+        explanation += "🔻 Factors that DECREASED Bot Likelihood:\n" + "\n".join(negative_impacts) + "\n\n"
 
-    explanation, img_base64 = get_shap_explanation(X, encoders=encoders)
+    explanation += "📝 These features collectively explain the model's decision.\n"
+    return explanation
 
-    return jsonify({
-        "Prediction": "Bot Attack" if pred_label else "Legitimate",
-        "Bot Probability": round(float(pred_prob), 4),
-        "SHAP Explanation": explanation,
-        "SHAP Waterfall Plot": f"data:image/png;base64,{img_base64}"
-    })
+@app.post('/v1/predict')
+def predict():
+    try:
+        row = request.get_json()
+        X = prepare_features(row)
+        pred_prob = model.predict_proba(X)[0][1]
+        pred_label = int(pred_prob >= 0.5)
+
+        # Generate SHAP explanation
+        shap_values = explainer.shap_values(X)
+        explanation = generate_shap_bot_attack_paragraph(0, shap_values, X, encoders)
+
+        return jsonify({
+            "Prediction": "Bot Attack" if pred_label else "Legitimate",
+            "Bot Probability": round(float(pred_prob), 4),
+            "SHAP Explanation": explanation
+        })
+
+    except Exception as e:
+        return jsonify({"error": str(e)}), 500