data_manager.py

# ml_engine/data_manager.py
# (V12.5 - Lazy Loading Fix + V15.6 App-Compat Fix)

import os
import asyncio
import httpx
import traceback
import time
from datetime import datetime
import ccxt.async_support as ccxt
import numpy as np
import logging
from typing import List, Dict, Any
import pandas as pd

try:
    import pandas_ta as ta
except ImportError:
    print("❌ [DataManager] مكتبة pandas_ta غير موجودة.")
    ta = None

from ml_engine.indicators import AdvancedTechnicalAnalyzer
from ml_engine.monte_carlo import MonteCarloAnalyzer
from ml_engine.ranker import Layer1Ranker
try:
    from ml_engine.patterns import ChartPatternAnalyzer
except ImportError:
    print("⚠️ [DataManager] لم يتم العثور على ml_engine/patterns.py")
    ChartPatternAnalyzer = None

logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
logging.getLogger("ccxt").setLevel(logging.WARNING)

class DataManager:
    def __init__(self, contracts_db, whale_monitor, r2_service=None):
        # ==================================================================
        # ⚙️ إعدادات التحكم المركزية (V12.3 Hybrid Thresholds)
        # ==================================================================
        # العتبة الرئيسية للنظام الهجين (Titan + Patterns + MC)
        self.HYBRID_ENTRY_THRESHOLD = 0.60
        # ==================================================================

        self.contracts_db = contracts_db or {}
        self.whale_monitor = whale_monitor
        self.r2_service = r2_service
        
        # تهيئة منصة التبادل (KuCoin كمثال)
        self.exchange = ccxt.kucoin({
            'enableRateLimit': True,
            'timeout': 30000,
        })
            
        self.http_client = None
        self.market_cache = {}

        self.technical_analyzer = AdvancedTechnicalAnalyzer()
        self.mc_analyzer = MonteCarloAnalyzer()
        
        self.layer1_ranker = None
        self.pattern_analyzer = None

    async def initialize(self):
        """تهيئة مدير البيانات والاتصالات"""
        self.http_client = httpx.AsyncClient(timeout=30.0)
        await self._load_markets()
        
        print("   > [DataManager] إنشاء النماذج المساندة (Lazy Load)...") # (تغيير الرسالة)
        try:
            self.layer1_ranker = Layer1Ranker(model_path="ml_models/layer1_ranker.lgbm")
            # [ 🔴🔴🔴 LAZY LOAD 🔴🔴🔴 ]
            # await self.layer1_ranker.initialize() # (تم الإلغاء - سيتم تحميله عند الحاجة)
            # [ 🔴🔴🔴 END LAZY LOAD 🔴🔴🔴 ]
            
            if ChartPatternAnalyzer:
                self.pattern_analyzer = ChartPatternAnalyzer(r2_service=self.r2_service)
                # [ 🔴🔴🔴 LAZY LOAD 🔴🔴🔴 ]
                # await self.pattern_analyzer.initialize() # (تم الإلغاء - سيتم تحميله عند الحاجة)
                # [ 🔴🔴🔴 END LAZY LOAD 🔴🔴🔴 ]
                
        except Exception as e:
            print(f"⚠️ [DataManager] تحذير أثناء إنشاء النماذج المساندة: {e}")

        print(f"✅ DataManager V12.5 initialized (Hybrid Threshold: {self.HYBRID_ENTRY_THRESHOLD})")

    async def _load_markets(self):
        """تحميل بيانات الأسواق وتخزينها مؤقتاً"""
        try:
            if self.exchange:
                await self.exchange.load_markets()
                self.market_cache = self.exchange.markets
        except Exception as e:
            print(f"❌ [DataManager] فشل تحميل الأسواق: {e}")

    async def close(self):
        """إغلاق جميع الاتصالات بأمان"""
        if self.http_client: await self.http_client.aclose()
        if self.exchange: await self.exchange.close()
        # تنظيف ذاكرة النماذج الفرعية إذا لزم الأمر
        if self.pattern_analyzer and hasattr(self.pattern_analyzer, 'clear_memory'):
            self.pattern_analyzer.clear_memory()
        if self.layer1_ranker and hasattr(self.layer1_ranker, 'clear_memory'):
            self.layer1_ranker.clear_memory()

    # ==================================================================
    # 🚀 [إضافة جديدة V15.6] دوال التوافق مع App
    # ==================================================================
    async def load_contracts_from_r2(self):
        """
        [جديد] يقوم بتحميل قاعدة بيانات العقود من R2 عند بدء التشغيل.
        (مطلوب بواسطة app.py V15.6)
        """
        print("   > [DataManager] Loading contracts database from R2...")
        if not self.r2_service:
            print("❌ [DataManager] R2Service not available. Cannot load contracts.")
            self.contracts_db = {}
            return
        
        try:
            self.contracts_db = await self.r2_service.load_contracts_db_async()
            print(f"✅ [DataManager] Contracts loaded. Total entries: {len(self.contracts_db)}")
        except Exception as e:
            print(f"❌ [DataManager] Failed to load contracts from R2: {e}")
            self.contracts_db = {}

    def get_contracts_db(self) -> Dict[str, Any]:
        """
        [جديد] إرجاع قاعدة بيانات العقود التي تم تحميلها.
        (مطلوب بواسطة app.py V15.6)
        """
        return self.contracts_db
    
    # ==================================================================
    # 🛡️ دوال الطبقة الأولى (Layer 1 Screening)
    # ==================================================================
    async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
        """
        الغربلة الأولية السريعة جداً بناءً على الحجم فقط.
        تختار أعلى 150 عملة سيولةً لتمريرها للتحليل الهجين المعمق.
        """
        print(f"🔍 [Layer 1] بدء الغربلة السريعة (Top Liquid Assets)...")
        volume_data = await self._get_volume_data_live()
        
        if not volume_data:
            print("⚠️ [Layer 1 Warning] لم يتم العثور على بيانات حجم تداول.")
            return []

        # اختيار أعلى 150 عملة من حيث حجم التداول الدولاري
        candidates = volume_data[:150]
        print(f"✅ [Layer 1] تم تمرير {len(candidates)} عملة للتحليل الهجين.")
        return candidates

    async def _get_volume_data_live(self):
        """جلب بيانات الحجم الحية لجميع الأزواج"""
        try:
            tickers = await self.exchange.fetch_tickers()
            data = []
            for symbol, ticker in tickers.items():
                # تصفية الأزواج: USDT فقط، وحجم تداول معقول (> 100k$) لتجنب العملات الميتة
                if symbol.endswith('/USDT') and ticker.get('quoteVolume') and ticker['quoteVolume'] > 100000:
                    data.append({
                        'symbol': symbol,
                        'dollar_volume': ticker['quoteVolume'],
                        'current_price': ticker['last']
                    })
            # الترتيب التنازلي حسب الحجم
            data.sort(key=lambda x: x['dollar_volume'], reverse=True)
            return data
        except Exception as e:
            print(f"❌ [DataManager] خطأ في جلب بيانات الحجم: {e}")
            return []

    # ==================================================================
    # 📊 دوال جلب البيانات (Data Fetching Pipeline)
    # ==================================================================
    async def stream_ohlcv_data(self, symbols: List[Dict], queue: asyncio.Queue):
        """
        مولد بيانات متدفق (Streaming Generator) يجلب شموع OHLCV
        لعدة إطارات زمنية لكل عملة، ويضع النتائج في طابور المعالجة.
        """
        timeframes = ['5m', '15m', '1h', '4h', '1d']
        limit = 500 # نحتاج تاريخ كافي للمؤشرات المعقدة

        for sym_data in symbols:
            symbol = sym_data['symbol']
            # جلب جميع الإطارات الزمنية بالتوازي للسرعة القصوى
            tasks = [self._fetch_ohlcv_live(symbol, tf, limit) for tf in timeframes]
            results = await asyncio.gather(*tasks, return_exceptions=False)
            
            ohlcv_packet = {}
            valid_packet = True
            for i, res in enumerate(results):
                tf = timeframes[i]
                # التحقق من جودة البيانات (على الأقل 200 شمعة للتحليل الدقيق)
                if res and isinstance(res, list) and len(res) >= 200:
                    ohlcv_packet[tf] = res
                else:
                    # إذا فقدنا إطاراً زمنياً حيوياً (مثل 5m أو 1h)، قد نعتبر الحزمة غير صالحة
                    if tf in ['5m', '1h']: valid_packet = False

            # إذا كانت الحزمة مكتملة بما يكفي، نرسلها للمعالجة
            if valid_packet and len(ohlcv_packet) >= 4:
                sym_data['ohlcv'] = ohlcv_packet
                await queue.put([sym_data]) # نرسل كقائمة لتوافق المعالج
            
            # فاصل زمني صغير جداً لتجنب تجاوز حدود الـ API بعنف
            await asyncio.sleep(0.05)

        # علامة نهاية التدفق
        await queue.put(None)

    async def _fetch_ohlcv_live(self, symbol, timeframe, limit):
        """دالة مساعدة لجلب الشموع مع معالجة الأخطاء البسيطة"""
        try:
            return await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
        except Exception:
            # نتجاهل الأخطاء الفردية لعدم إيقاف التدفق الكامل
            return None

    # ==================================================================
    # 🎯 دوال مساعدة للحارس والدماغ (Sentry & Brain Helpers)
    # ==================================================================
    async def get_latest_price_async(self, symbol: str) -> float:
        """جلب آخر سعر حقيقي (للتنفيذ والمراقبة)"""
        try:
            ticker = await self.exchange.fetch_ticker(symbol)
            return float(ticker['last'])
        except Exception as e:
            print(f"⚠️ [DataManager] Failed to fetch price for {symbol}: {e}")
            return 0.0

    # [NEW FIX V12.4] الدالة التي كانت مفقودة وتمت إضافتها
    async def get_latest_ohlcv(self, symbol: str, timeframe: str = '5m', limit: int = 100) -> List[List[float]]:
        """
        جلب عدد محدود من الشموع الأخيرة بسرعة.
        يستخدمها 'القناص' (Sniper) للمراقبة اللحظية الخفيفة.
        """
        try:
            candles = await self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
            if candles and len(candles) > 0:
                return candles
            return []
        except Exception as e:
            # print(f"⚠️ [DataManager] Failed rapid OHLCV fetch for {symbol}: {e}")
            return []