مراقبة النماذج في الإنتاج

import random
from dataclasses import dataclass, field
from typing import List, Dict

@dataclass
class Prediction:
    text:       str
    predicted:  str
    confidence: float
    actual:     str = ""
    day:        int = 1

    @property
    def correct(self) -> bool:
        return self.predicted == self.actual if self.actual else True

class ModelMonitor:
    """مراقبة أداء النموذج في الإنتاج"""

    def __init__(self, model_name: str, baseline_acc: float = 0.94):
        self.model_name  = model_name
        self.baseline    = baseline_acc
        self.preds: List[Prediction] = []
        self._window     = 50

    def log(self, p: Prediction):
        self.preds.append(p)

    def _recent(self) -> List[Prediction]:
        return self.preds[-self._window:]

    def current_accuracy(self) -> float:
        with_labels = [p for p in self._recent() if p.actual]
        if not with_labels:
            return self.baseline
        return sum(1 for p in with_labels if p.correct) / len(with_labels)

    def avg_confidence(self) -> float:
        r = self._recent()
        return sum(p.confidence for p in r) / len(r) if r else 1.0

    def low_confidence_rate(self, threshold: float = 0.60) -> float:
        r = self._recent()
        if not r:
            return 0.0
        return sum(1 for p in r if p.confidence < threshold) / len(r)

    def detect_drift(self) -> Dict:
        curr_acc  = self.current_accuracy()
        curr_conf = self.avg_confidence()
        low_conf  = self.low_confidence_rate()
        acc_drop  = self.baseline - curr_acc

        alerts = []
        if acc_drop > 0.05:
            drop_str = f"{acc_drop:.1%}"
            alerts.append(f"Performance Drift: دقة انخفضت {drop_str}")
        if curr_conf < 0.75:
            conf_str = f"{curr_conf:.1%}"
            alerts.append(f"Low Confidence: متوسط الثقة {conf_str}")
        if low_conf > 0.20:
            rate_str = f"{low_conf:.1%}"
            alerts.append(f"High Uncertainty Rate: {rate_str} من التنبؤات منخفضة الثقة")

        return {
            "total":    len(self.preds),
            "accuracy": curr_acc,
            "baseline": self.baseline,
            "confidence": curr_conf,
            "alerts":   alerts,
            "status":   "ALERT" if alerts else "HEALTHY",
        }

    def report(self, title: str = ""):
        d    = self.detect_drift()
        icon = "🔴" if d["status"] == "ALERT" else "✅"
        label = title or self.model_name
        print(f"\n{icon} مراقبة: {label}")
        print("=" * 50)
        print(f"  Status         : {d['status']}")
        print(f"  التنبؤات       : {d['total']}")
        acc_str  = f"{d['accuracy']:.1%}"
        base_str = f"{d['baseline']:.1%}"
        conf_str = f"{d['confidence']:.1%}"
        print(f"  الدقة الحالية  : {acc_str}  (baseline: {base_str})")
        print(f"  متوسط الثقة    : {conf_str}")
        if d["alerts"]:
            print(f"\n  🚨 التنبيهات:")
            for a in d["alerts"]:
                print(f"     ⚠️  {a}")
        else:
            print(f"\n  ✅ لا انحرافات — النموذج صحيح")

# ─── محاكاة بيانات إنتاج ───────────────────────────────────
random.seed(99)
classes = ["positive", "negative", "neutral"]
monitor = ModelMonitor("SentimentClassifier-v2.3", baseline_acc=0.94)

print("📊 مراقبة النموذج في بيئة الإنتاج:")

# الأسبوع 1-2: أداء طبيعي
print("\n  الأسبوع 1-2: أداء طبيعي...")
for day in range(1, 15):
    for _ in range(5):
        actual = random.choice(classes)
        pred   = actual if random.random() < 0.94 else random.choice(classes)
        monitor.log(Prediction(
            text=f"review_{day}", predicted=pred,
            confidence=random.uniform(0.72, 0.99),
            actual=actual, day=day,
        ))

monitor.report("بعد أسبوعين")

# الأسبوع 3-4: بداية انجراف
print("\n\n  الأسبوع 3-4: بيانات جديدة مختلفة...")
for day in range(15, 29):
    for _ in range(5):
        actual = random.choice(classes)
        pred   = actual if random.random() < 0.82 else random.choice(classes)
        monitor.log(Prediction(
            text=f"new_review_{day}", predicted=pred,
            confidence=random.uniform(0.40, 0.72),  # ثقة منخفضة
            actual=actual, day=day,
        ))

monitor.report("بعد شهر")

# إجراءات التصحيح
print(f"\n\n💡 خطوات التصحيح عند الانجراف:")
steps = [
    "تحليل البيانات الجديدة — أين الاختلاف؟",
    "جمع بيانات تدريب جديدة تمثّل التوزيع الحالي",
    "إعادة التدريب مع MLflow لتتبع التغييرات",
    "A/B Testing بين النموذج القديم والجديد",
    "نشر تدريجي (Canary) بنسبة 10% ثم 100%",
]
for i, s in enumerate(steps, 1):
    print(f"  {i}. {s}")