Create models​​​/anomaly.py

models​​​/anomaly.py

@@ -0,0 +1,25 @@
+import pandas as pd
+from transformers import pipeline
+import logging
+
+logger = logging.getLogger(__name__)
+
+def detect_anomalies(df):
+    """Detect anomalies in log data using a Hugging Face model."""
+    logger.info("Detecting anomalies...")
+    try:
+        detector = pipeline(
+            "text-classification",
+            model="prajjwal1/bert-tiny",
+            tokenizer="prajjwal1/bert-tiny",
+            clean_up_tokenization_spaces=True
+        )
+        df["text"] = df["status"] + " Usage:" + df["usage_count"].astype(str)
+        results = detector(df["text"].tolist())
+        df["anomaly"] = [r["label"] for r in results]
+        anomalies = df[df["anomaly"] == "POSITIVE"]
+        logger.info(f"Detected {len(anomalies)} anomalies.")
+        return anomalies
+    except Exception as e:
+        logger.error(f"Failed to detect anomalies: {e}")
+        raise