ml_engine/ranker.py

# ml_engine/ranker.py (V11.1 - Layer 1 Smart Ranker with Memory Management)
import pandas as pd
import numpy as np
import os
import gc # استيراد لجمع القمامة
from typing import List, Dict, Any

try:
    import lightgbm as lgb
    LGBM_AVAILABLE = True
except ImportError:
    print("❌❌ [Ranker] مكتبة 'lightgbm' غير موجودة.")
    LGBM_AVAILABLE = False

class Layer1Ranker:
    def __init__(self, model_path: str = "ml_models/layer1_ranker.lgbm"):
        if not LGBM_AVAILABLE: raise ImportError("lightgbm required")
        self.model_path = model_path
        self.model_name = os.path.basename(model_path)
        self.model = None
        self.features_in_ = None

    async def initialize(self):
        """تحميل النموذج من الملف المحلي"""
        print(f"🔄 [Ranker] تحميل {self.model_name}...")
        try:
            if os.path.exists(self.model_path):
                self.model = lgb.Booster(model_file=self.model_path)
                self.features_in_ = self.model.feature_name()
                print(f"✅ [Ranker] تم تحميل النموذج بنجاح ({len(self.features_in_)} features).")
            else:
                print(f"❌ [Ranker] ملف النموذج غير موجود: {self.model_path}")
                self.model = None
        except Exception as e:
            print(f"❌ [Ranker] فشل تحميل النموذج: {e}")
            self.model = None

    def predict_proba(self, features_df: pd.DataFrame) -> np.ndarray:
        """التنبؤ باحتمالية الصعود"""
        if self.model is None: return np.zeros(len(features_df))
        try:
            # التأكد من ترتيب الأعمدة
            if not all(f in features_df.columns for f in self.features_in_):
                # محاولة تعبئة الميزات المفقودة بأصفار إذا لزم الأمر
                for f in self.features_in_:
                    if f not in features_df.columns: features_df[f] = 0.0
            
            aligned_df = features_df[self.features_in_]
            return self.model.predict(aligned_df)
        except Exception as e:
            print(f"❌ [Ranker] خطأ التنبؤ: {e}")
            return np.zeros(len(features_df))

    # 🔴 دالة جديدة لتنظيف الذاكرة
    def clear_memory(self):
        """تحرير نموذج LightGBM من الذاكرة"""
        self.model = None
        self.features_in_ = None
        gc.collect()
        print(f"🧹 [Ranker] تم تحرير النموذج من الذاكرة.")

print("✅ ML Module: Layer 1 Ranker V11.1 (Memory Managed) loaded.")