Update data_manager.py

data_manager.py

@@ -45,7 +45,9 @@ class DataManager:
                 return
                 
             print("🔄 جلب أحدث بيانات الأسواق من KuCoin...")
-            self.exchange.load_markets()
+            # استخدام run_in_executor للدوال المتزامنة
+            loop = asyncio.get_event_loop()
+            await loop.run_in_executor(None, self.exchange.load_markets)
             self.market_cache = self.exchange.markets
             self.last_market_load = datetime.now()
             print(f"✅ تم تحميل {len(self.market_cache)} سوق من KuCoin")
@@ -155,12 +157,15 @@ class DataManager:
         try:
             prices = {'bitcoin': None, 'ethereum': None}
             
-            btc_ticker = self.exchange.fetch_ticker('BTC/USDT')
+            # استخدام run_in_executor للدوال المتزامنة
+            loop = asyncio.get_event_loop()
+            
+            btc_ticker = await loop.run_in_executor(None, self.exchange.fetch_ticker, 'BTC/USDT')
             btc_price = float(btc_ticker.get('last', 0)) if btc_ticker.get('last') else None
             if btc_price and btc_price > 0:
                 prices['bitcoin'] = btc_price
             
-            eth_ticker = self.exchange.fetch_ticker('ETH/USDT')
+            eth_ticker = await loop.run_in_executor(None, self.exchange.fetch_ticker, 'ETH/USDT')
             eth_price = float(eth_ticker.get('last', 0)) if eth_ticker.get('last') else None
             if eth_price and eth_price > 0:
                 prices['ethereum'] = eth_price
@@ -168,6 +173,7 @@ class DataManager:
             return prices
             
         except Exception as e:
+            print(f"⚠️ فشل جلب الأسعار من KuCoin: {e}")
             return {'bitcoin': None, 'ethereum': None}
 
     async def _get_prices_from_coingecko(self):
@@ -176,7 +182,6 @@ class DataManager:
             await asyncio.sleep(0.5)
             url = "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin,ethereum&vs_currencies=usd"
             
-            # إضافة headers لتجنب rate limiting
             headers = {
                 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
                 'Accept': 'application/json'
@@ -201,6 +206,7 @@ class DataManager:
                     return {'bitcoin': None, 'ethereum': None}
                     
         except Exception as e:
+            print(f"⚠️ فشل جلب الأسعار من CoinGecko: {e}")
             return {'bitcoin': None, 'ethereum': None}
 
     def _get_minimal_market_context(self):
@@ -219,33 +225,27 @@ class DataManager:
         """
         print("📊 الطبقة 1: جلب أفضل 200 عملة حسب حجم التداول...")
         
-        # المحاولة 1: الطريقة المثلى - استخدام fetch_tickers
         volume_data = await self._get_volume_data_optimal()
         
         if not volume_data:
-            # المحاولة 2: الطريقة البديلة - استخدام API المباشر
             volume_data = await self._get_volume_data_direct_api()
         
         if not volume_data:
-            # المحاولة 3: الطريقة التقليدية (الاحتياطية)
             volume_data = await self._get_volume_data_traditional()
         
         if not volume_data:
             print("❌ فشل جميع محاولات جلب بيانات الأحجام")
             return []
         
-        # أخذ أفضل 200 عملة حسب الحجم فقط
         volume_data.sort(key=lambda x: x['dollar_volume'], reverse=True)
         top_200_by_volume = volume_data[:200]
         
         print(f"✅ تم اختيار أفضل {len(top_200_by_volume)} عملة حسب الحجم")
         
-        # المرحلة 2: تطبيق المؤشرات الأخرى على الـ200 فقط
         final_candidates = await self._apply_advanced_indicators(top_200_by_volume)
         
         print(f"🎯 تم تحليل {len(final_candidates)} عملة للطبقة 2")
         
-        # عرض أفضل 15 عملة
         print("🏆 أفضل 15 عملة من الطبقة 1:")
         for i, candidate in enumerate(final_candidates[:15]):
             score = candidate.get('layer1_score', 0)
@@ -261,32 +261,30 @@ class DataManager:
             if not self.exchange:
                 return []
                 
-            tickers = self.exchange.fetch_tickers()
+            # استخدام run_in_executor للدوال المتزامنة
+            loop = asyncio.get_event_loop()
+            tickers = await loop.run_in_executor(None, self.exchange.fetch_tickers)
             
             volume_data = []
             processed = 0
             
             for symbol, ticker in tickers.items():
-                # تصفية أزواج USDT النشطة فقط
                 if not symbol.endswith('/USDT'):
                     continue
                     
                 if not ticker.get('active', True):
                     continue
                 
-                # استخدام quoteVolume (الحجم بالدولار) إذا متوفر
                 current_price = ticker.get('last', 0)
                 quote_volume = ticker.get('quoteVolume', 0)
                 
                 if quote_volume > 0:
                     dollar_volume = quote_volume
                 else:
-                    # fallback: baseVolume * السعر
                     base_volume = ticker.get('baseVolume', 0)
                     dollar_volume = base_volume * current_price
                 
-                # فلترة أولية: تجاهل العملات ذات الحجم المنخفض جداً
-                if dollar_volume < 50000:  # أقل من 50K دولار
+                if dollar_volume < 50000:
                     continue
                 
                 volume_data.append({
@@ -302,6 +300,7 @@ class DataManager:
             return volume_data
             
         except Exception as e:
+            print(f"❌ خطأ في جلب بيانات الحجم المثلى: {e}")
             return []
 
     async def _get_volume_data_direct_api(self) -> List[Dict[str, Any]]:
@@ -323,15 +322,13 @@ class DataManager:
                 for ticker in tickers:
                     symbol = ticker['symbol']
                     
-                    # تصفية أزواج USDT فقط
                     if not symbol.endswith('USDT'):
                         continue
                     
-                    # تحويل الرمز للتنسيق القياسي (BTC-USDT → BTC/USDT)
                     formatted_symbol = symbol.replace('-', '/')
                     
                     try:
-                        dollar_volume = float(ticker['volValue'])  # الحجم بالدولار
+                        dollar_volume = float(ticker['volValue'])
                         current_price = float(ticker['last'])
                         price_change = float(ticker['changeRate']) * 100
                         
@@ -349,6 +346,7 @@ class DataManager:
                 return volume_data
                 
         except Exception as e:
+            print(f"❌ خطأ في جلب بيانات الحجم المباشر: {e}")
             return []
 
     async def _get_volume_data_traditional(self) -> List[Dict[str, Any]]:
@@ -365,7 +363,6 @@ class DataManager:
             volume_data = []
             processed = 0
             
-            # معالجة دفعات لتجنب rate limits
             batch_size = 50
             for i in range(0, len(usdt_symbols), batch_size):
                 batch = usdt_symbols[i:i + batch_size]
@@ -378,19 +375,22 @@ class DataManager:
                 
                 processed += len(batch)
                 
-                # انتظار قصير بين الدفعات
                 if i + batch_size < len(usdt_symbols):
                     await asyncio.sleep(1)
             
             return volume_data
             
         except Exception as e:
+            print(f"❌ خطأ في جلب بيانات الحجم التقليدية: {e}")
             return []
 
     async def _process_single_symbol(self, symbol: str) -> Dict[str, Any]:
         """معالجة رمز واحد لجلب بيانات الحجم"""
         try:
-            ticker = self.exchange.fetch_ticker(symbol)
+            # استخدام run_in_executor للدوال المتزامنة
+            loop = asyncio.get_event_loop()
+            ticker = await loop.run_in_executor(None, self.exchange.fetch_ticker, symbol)
+            
             if not ticker:
                 return None
             
@@ -424,15 +424,12 @@ class DataManager:
             try:
                 symbol = symbol_data['symbol']
                 
-                # جلب بيانات إضافية للرمز
                 detailed_data = await self._get_detailed_symbol_data(symbol)
                 if not detailed_data:
                     continue
                 
-                # دمج البيانات
                 symbol_data.update(detailed_data)
                 
-                # حساب الدرجة النهائية
                 score = self._calculate_advanced_score(symbol_data)
                 symbol_data['layer1_score'] = score
                 
@@ -441,14 +438,16 @@ class DataManager:
             except Exception as e:
                 continue
         
-        # ترتيب المرشحين حسب الدرجة النهائية
         candidates.sort(key=lambda x: x.get('layer1_score', 0), reverse=True)
         return candidates
 
     async def _get_detailed_symbol_data(self, symbol: str) -> Dict[str, Any]:
         """جلب بيانات تفصيلية للرمز"""
         try:
-            ticker = self.exchange.fetch_ticker(symbol)
+            # استخدام run_in_executor للدوال المتزامنة
+            loop = asyncio.get_event_loop()
+            ticker = await loop.run_in_executor(None, self.exchange.fetch_ticker, symbol)
+            
             if not ticker:
                 return None
             
@@ -458,7 +457,6 @@ class DataManager:
             open_price = ticker.get('open', 0)
             price_change_24h = (ticker.get('percentage', 0) or 0) * 100
             
-            # حساب المؤشرات المتقدمة
             volatility = self._calculate_volatility(high_24h, low_24h, current_price)
             price_strength = self._calculate_price_strength(current_price, open_price, price_change_24h)
             momentum = self._calculate_momentum(price_change_24h)
@@ -485,19 +483,11 @@ class DataManager:
         price_strength = symbol_data.get('price_strength', 0)
         momentum = symbol_data.get('momentum', 0)
         
-        # 1. درجة الحجم (40%) - الأهم
         volume_score = self._calculate_volume_score(dollar_volume)
-        
-        # 2. درجة الزخم (25%)
         momentum_score = momentum
-        
-        # 3. درجة التقلب (20%)
         volatility_score = self._calculate_volatility_score(volatility)
-        
-        # 4. درجة قوة السعر (15%)
         strength_score = price_strength
         
-        # الدرجة النهائية
         final_score = (
             volume_score * 0.40 +
             momentum_score * 0.25 +
@@ -505,7 +495,6 @@ class DataManager:
             strength_score * 0.15
         )
         
-        # تحديث أسباب الترشيح
         reasons = []
         if volume_score >= 0.7:
             reasons.append('high_volume')
@@ -520,19 +509,19 @@ class DataManager:
 
     def _calculate_volume_score(self, dollar_volume: float) -> float:
         """حساب درجة الحجم"""
-        if dollar_volume >= 10000000:    # 10M+
+        if dollar_volume >= 10000000:
             return 1.0
-        elif dollar_volume >= 5000000:   # 5M+
+        elif dollar_volume >= 5000000:
             return 0.9
-        elif dollar_volume >= 2000000:   # 2M+
+        elif dollar_volume >= 2000000:
             return 0.8
-        elif dollar_volume >= 1000000:   # 1M+
+        elif dollar_volume >= 1000000:
             return 0.7
-        elif dollar_volume >= 500000:    # 500K+
+        elif dollar_volume >= 500000:
             return 0.6
-        elif dollar_volume >= 250000:    # 250K+
+        elif dollar_volume >= 250000:
             return 0.5
-        elif dollar_volume >= 100000:    # 100K+
+        elif dollar_volume >= 100000:
             return 0.4
         else:
             return 0.3
@@ -545,13 +534,13 @@ class DataManager:
 
     def _calculate_volatility_score(self, volatility: float) -> float:
         """حساب درجة التقلب"""
-        if 0.02 <= volatility <= 0.15:   # تقلب مثالي 2%-15%
+        if 0.02 <= volatility <= 0.15:
             return 1.0
-        elif 0.01 <= volatility <= 0.20: # مقبول 1%-20%
+        elif 0.01 <= volatility <= 0.20:
             return 0.8
-        elif volatility <= 0.01:         # قليل جداً
+        elif volatility <= 0.01:
             return 0.4
-        elif volatility > 0.20:          # عالي جداً
+        elif volatility > 0.20:
             return 0.3
         else:
             return 0.5
@@ -561,7 +550,6 @@ class DataManager:
         if open_price == 0:
             return 0.5
         
-        # قوة السعر تعتمد على المسافة من سعر الافتتاح ونسبة التغير
         distance_from_open = abs(current_price - open_price) / open_price
         change_strength = min(abs(price_change) / 50, 1.0)
         
@@ -569,21 +557,21 @@ class DataManager:
 
     def _calculate_momentum(self, price_change: float) -> float:
         """حساب الزخم"""
-        if price_change >= 15:           # +15%+
+        if price_change >= 15:
             return 1.0
-        elif price_change >= 10:         # +10%+
+        elif price_change >= 10:
             return 0.9
-        elif price_change >= 5:          # +5%+
+        elif price_change >= 5:
             return 0.8
-        elif price_change >= 2:          # +2%+
+        elif price_change >= 2:
             return 0.7
-        elif price_change >= 0:          # موجب
+        elif price_change >= 0:
             return 0.6
-        elif price_change >= -5:         # حتى -5%
+        elif price_change >= -5:
             return 0.5
-        elif price_change >= -10:        # حتى -10%
+        elif price_change >= -10:
             return 0.4
-        else:                            # أكثر من -10%
+        else:
             return 0.3
 
     async def get_ohlcv_data_for_symbols(self, symbols: List[str]) -> List[Dict[str, Any]]:
@@ -592,8 +580,7 @@ class DataManager:
         """
         print(f"📊 جلب بيانات OHLCV كاملة لـ {len(symbols)} عملة بشكل متوازي...")
         
-        # تقسيم الرموز إلى دفعات لتجنب rate limits
-        batch_size = 15  # تقليل حجم الدفعة لتحسين الاستقرار
+        batch_size = 15
         batches = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)]
         
         all_results = []
@@ -601,16 +588,13 @@ class DataManager:
         for batch_num, batch in enumerate(batches):
             print(f"   🔄 معالجة الدفعة {batch_num + 1}/{len(batches)} ({len(batch)} عملة)...")
             
-            # إنشاء مهام للدفعة الحالية بشكل متوازي
             batch_tasks = []
             for symbol in batch:
                 task = asyncio.create_task(self._fetch_complete_ohlcv_parallel(symbol))
                 batch_tasks.append(task)
             
-            # انتظار انتهاء جميع مهام الدفعة الحالية
             batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True)
             
-            # معالجة نتائج الدفعة
             successful_count = 0
             for i, result in enumerate(batch_results):
                 symbol = batch[i]
@@ -619,7 +603,6 @@ class DataManager:
                 elif result is not None:
                     all_results.append(result)
                     successful_count += 1
-                    # طباعة رسالة واحدة فقط لكل عملة توضح عدد الأطر الزمنية
                     timeframes_count = result.get('successful_timeframes', 0)
                     print(f"      ✅ {symbol}: {timeframes_count}/6 أطر زمنية")
                 else:
@@ -627,9 +610,8 @@ class DataManager:
         
             print(f"      ✅ اكتملت الدفعة {batch_num + 1}: {successful_count}/{len(batch)} ناجحة")
             
-            # انتظار قصير بين الدفعات لتجنب rate limits
             if batch_num < len(batches) - 1:
-                await asyncio.sleep(1)  # زيادة وقت الانتظار
+                await asyncio.sleep(1)
         
         print(f"✅ تم تجميع بيانات OHLCV كاملة لـ {len(all_results)} عملة من أصل {len(symbols)}")
         return all_results
@@ -639,7 +621,6 @@ class DataManager:
         try:
             ohlcv_data = {}
             
-            # جلب 200 شمعة لكل إطار زمني مع تحسين التعامل مع الأخطاء
             timeframes = [
                 ('5m', 200),
                 ('15m', 200),  
@@ -649,40 +630,34 @@ class DataManager:
                 ('1w', 200),
             ]
             
-            # إنشاء مهام لجميع الإطارات الزمنية بشكل متوازي
             timeframe_tasks = []
             for timeframe, limit in timeframes:
                 task = asyncio.create_task(self._fetch_single_timeframe_improved(symbol, timeframe, limit))
                 timeframe_tasks.append(task)
             
-            # انتظار جميع المهام
             timeframe_results = await asyncio.gather(*timeframe_tasks, return_exceptions=True)
             
-            # تحسين: قبول البيانات حتى لو كانت غير مكتملة
             successful_timeframes = 0
-            min_required_timeframes = 2  # تخفيف الشرط من 3 إلى 2 أطر زمنية
+            min_required_timeframes = 2
             
-            # معالجة النتائج
             for i, (timeframe, limit) in enumerate(timeframes):
                 result = timeframe_results[i]
                 
                 if isinstance(result, Exception):
                     continue
                 
-                if result and len(result) >= 10:  # تخفيف الشرط من 50 إلى 10 شموع
+                if result and len(result) >= 10:
                     ohlcv_data[timeframe] = result
                     successful_timeframes += 1
             
-            # تحسين: قبول العملة إذا كان لديها عدد كافٍ من الأطر الزمنية
             if successful_timeframes >= min_required_timeframes and ohlcv_data:
                 try:
-                    # ✅ الحصول على السعر الحالي مباشرة
                     current_price = await self.get_latest_price_async(symbol)
                     
                     result_data = {
                         'symbol': symbol,
                         'ohlcv': ohlcv_data,
-                        'raw_ohlcv': ohlcv_data,  # ✅ إضافة البيانات الخام مباشرة
+                        'raw_ohlcv': ohlcv_data,
                         'current_price': current_price,
                         'timestamp': datetime.now().isoformat(),
                         'candles_count': {tf: len(data) for tf, data in ohlcv_data.items()},
@@ -708,7 +683,9 @@ class DataManager:
         
         for attempt in range(max_retries):
             try:
-                ohlcv_data = self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
+                # استخدام run_in_executor للدوال المتزامنة
+                loop = asyncio.get_event_loop()
+                ohlcv_data = await loop.run_in_executor(None, self.exchange.fetch_ohlcv, symbol, timeframe, limit)
                 
                 if ohlcv_data and len(ohlcv_data) > 0:
                     return ohlcv_data
@@ -722,14 +699,15 @@ class DataManager:
                     return []
 
     async def get_latest_price_async(self, symbol):
-        """جلب السعر الحالي لعملة محددة"""
+        """جلب السعر الحالي لعملة محددة - الإصلاح الرئيسي هنا"""
         try:
             if not self.exchange:
                 print(f"❌ Exchange غير مهيأ لـ {symbol}")
                 return None
 
-            # ✅ الإصلاح الرئيسي: إزالة asyncio.create_task واستخدام fetch_ticker مباشرة
-            ticker = self.exchange.fetch_ticker(symbol)
+            # استخدام run_in_executor للدوال المتزامنة بدلاً من asyncio.create_task
+            loop = asyncio.get_event_loop()
+            ticker = await loop.run_in_executor(None, self.exchange.fetch_ticker, symbol)
             
             if not ticker:
                 print(f"❌ لم يتم العثور على ticker لـ {symbol}")
@@ -780,8 +758,6 @@ class DataManager:
         except Exception as e:
             return False
 
-    # === الدوال الجديدة لدعم بيانات الحيتان ===
-    
     async def get_whale_data_for_symbol(self, symbol):
         """جلب بيانات الحيتان لعملة محددة"""
         try:
@@ -813,4 +789,4 @@ class DataManager:
                 'source': 'whale_analysis'
             }
 
-print("✅ DataManager loaded - Parallel OHLCV Fetching System with Whale Data Support & Improved Error Handling")
+print("✅ DataManager loaded - Fixed Async Wrapper for ccxt")