[Feature Added] > Model Drift Detection mechanism. This feature checks, if the model is confident in predicting the new uploaded images/videos as AI or not, if not then we need to retrain the model.

backend/app/ai/drift_monitor.py

@@ -0,0 +1,61 @@
+import logging
+import numpy as np
+from sqlalchemy.orm import Session
+from datetime import datetime, timedelta, UTC
+from app.models.file_model import File
+from scipy.stats import wasserstein_distance
+
+logger = logging.getLogger(__name__)
+
+class DriftMonitor:
+    def __init__(self, db: Session):
+        self.db = db
+        self.CONFIDENCE_DROP_THRESHOLD = 0.15
+        self.DISTRIBUTION_DRIFT_THRESHOLD = 0.20
+        self.reference_days = 30
+        self.recent_days = 7
+    
+    def _get_historical_data(self, days_ago_start: int, days_ago_end: int):
+        end_date = datetime.now(UTC) - timedelta(days=days_ago_end)
+        start_date = datetime.now(UTC) - timedelta(days=days_ago_start)
+
+        records = self.db.query(File.result, File.confidence).filter(
+            File.created_at >= start_date,
+            File.created_at <= end_date,
+            File.status == "Completed",
+            File.confidence != None
+        ).all()
+
+        scores = [r.confidence for r in records]
+        classifications = [1 if "AI" in r.result or "Suspicious" in r.result else 0 for r in records]
+
+        return scores, classifications
+    
+    def detect_drift(self):
+        ref_scores, ref_classes = self._get_historical_data(self.reference_days, self.recent_days)
+        recent_scores, recent_classes = self._get_historical_data(self.recent_days, 0)
+
+        if len(ref_scores) < 50 or len(recent_scores) < 20:
+            logger.info("Insufficient data to perform drink analysis.")
+            return False
+        
+        ref_avg_conf = np.mean(ref_scores)
+        recent_avg_conf = np.mean(recent_scores)
+
+        confidence_drop = ref_avg_conf - recent_avg_conf
+
+        if confidence_drop > self.CONFIDENCE_DROP_THRESHOLD:
+            logger.warning(f"DRIFT ALERT: Average AI confidence dropped significantly! (Historical: {ref_avg_conf:.2f}, Recent: {recent_avg_conf:.2f})")
+        
+        drift_distance = wasserstein_distance(ref_scores, recent_scores)
+
+        if drift_distance > self.DISTRIBUTION_DRIFT_THRESHOLD:
+            logger.warning(f"DRIFT ALERT: Huge shift in prediction patterns! Statistical distance: {drift_distance:.2f}")
+
+        ref_ai_ratio = sum(ref_classes) / len(ref_classes)
+        recent_ai_ratio = sum(recent_classes) / len(recent_classes)
+
+        if abs(ref_ai_ratio - recent_ai_ratio) > 0.30:
+            logger.warning(f"DRIFT ALERT: Abnormal spike/drop in AI classifications! (Historical Ratio: {ref_ai_ratio:.2f}, Recent: {recent_ai_ratio:.2f})")
+        
+        return True