Update ml_engine/processor.py

ml_engine/processor.py

@@ -18,138 +18,104 @@ class MLProcessor:
         self.data_manager = data_manager
         self.learning_engine = learning_engine
         
-        # يقوم المعالج الرئيسي بتهيئة جميع الأدوات التي يحتاجها
         self.technical_analyzer = AdvancedTechnicalAnalyzer()
         self.strategy_engine = MultiStrategyEngine(data_manager, learning_engine)
         self.monte_carlo_analyzer = MonteCarloAnalyzer()
         self.pattern_analyzer = ChartPatternAnalyzer()
 
-        # 🔴 --- إضافة جديدة: منظم سرعة مخصص لبيانات الحيتان --- 🔴
-        # السماح بطلبين متزامنين كحد أقصى لبيانات الحيتان عبر النظام بأكمله
-        # يمكنك تجربة قيمة 1 أو 2 هنا
-        self.whale_data_semaphore = asyncio.Semaphore(2) 
-        # 🔴 --- نهاية الإضافة --- 🔴
+        # 🔴 تم إزالة منظم سرعة الحيتان من هنا
+        # self.whale_data_semaphore = asyncio.Semaphore(1) 
 
-    async def process_and_score_symbol_enhanced(self, raw_data):
-        """المعالجة المحسنة للرموز مع كل التحليلات المتقدمة"""
+    # 🔴 --- تعديل جوهري: إضافة معلمة جديدة واستخدامها --- 🔴
+    async def process_and_score_symbol_enhanced(self, raw_data, preloaded_whale_data: dict = None):
+        """
+        (معدلة) المعالجة المحسنة للرموز
+        - تتوقع بيانات الحيتان كمعلمة (preloaded_whale_data)
+        - لا تقوم بجلب بيانات الحيتان بنفسها
+        """
         try:
             if not raw_data or not raw_data.get('ohlcv'): 
-                # print(f"❌ بيانات غير صالحة للرمز {raw_data.get('symbol', 'unknown')}") # Reduced logging
                 return None
             
             symbol = raw_data['symbol']
-            # print(f"🔍 معالجة الرمز {symbol} بالتحليلات المتقدمة...") # Reduced logging
             
             base_analysis = await self.process_and_score_symbol(raw_data)
             if not base_analysis: 
                 return None
             
             try:
+                # حساب المؤشرات (لا تغيير)
                 advanced_indicators = {}
-                ohlcv_available = raw_data.get('ohlcv', {}) # Get OHLCV data
-                for timeframe, candles in ohlcv_available.items(): # Use ohlcv_available
-                    if candles and len(candles) >= 20: # Ensure candles is not None
+                ohlcv_available = raw_data.get('ohlcv', {}) 
+                for timeframe, candles in ohlcv_available.items(): 
+                    if candles and len(candles) >= 20: 
                         dataframe = self._create_dataframe(candles)
-                        # 1. استخدام محلل المؤشرات (سريع نسبياً)
                         indicators = self.technical_analyzer.calculate_all_indicators(dataframe, timeframe)
                         advanced_indicators[timeframe] = indicators
-                
                 base_analysis['advanced_indicators'] = advanced_indicators
                 
-                # 2. استخدام محلل مونت كارلو (حسابي، لا يستهلك شبكة)
-                monte_carlo_probability = await self.monte_carlo_analyzer.predict_1h_probability(ohlcv_available) # Use ohlcv_available
+                # حساب مونت كارلو (لا تغيير)
+                monte_carlo_probability = await self.monte_carlo_analyzer.predict_1h_probability(ohlcv_available) 
                 if monte_carlo_probability is not None:
                     base_analysis['monte_carlo_probability'] = monte_carlo_probability
                     base_analysis['monte_carlo_details'] = self.monte_carlo_analyzer.simulation_results
                 
-                # 3. استخدام محلل الأنماط (حسابي، لا يستهلك شبكة)
-                pattern_analysis = await self.pattern_analyzer.detect_chart_patterns(ohlcv_available) # Use ohlcv_available
+                # حساب الأنماط (لا تغيير)
+                pattern_analysis = await self.pattern_analyzer.detect_chart_patterns(ohlcv_available) 
                 base_analysis['pattern_analysis'] = pattern_analysis
                 
-                # 🔴 --- تعديل: استخدام منظم السرعة لجلب بيانات الحيتان --- 🔴
-                whale_data = None
-                # print(f"   🐋 [ML-{symbol}] انتظار للحصول على قفل بيانات الحيتان...") # Optional debug logging
-                async with self.whale_data_semaphore:
-                    # print(f"   🐋 [ML-{symbol}] حصل على القفل، بدء جلب بيانات الحيتان...") # Optional debug logging
-                    try:
-                        # هذه هي العملية التي تستهلك الشبكة بشكل كبير
-                        whale_data = await self.data_manager.get_whale_data_for_symbol(symbol)
-                        if whale_data:
-                            base_analysis['whale_data'] = whale_data
-                        # print(f"   🐋 [ML-{symbol}] اكتمل جلب بيانات الحيتان.") # Optional debug logging
-                    except Exception as whale_error:
-                         print(f"   ⚠️ [ML-{symbol}] خطأ أثناء جلب بيانات الحيتان: {whale_error}")
-                         base_analysis['whale_data'] = {'data_available': False, 'error': str(whale_error)} # Store error info
-                # print(f"   🐋 [ML-{symbol}] تحرير قفل بيانات الحيتان.") # Optional debug logging
+                # 🔴 --- تعديل: استخدام بيانات الحيتان المحملة مسبقاً --- 🔴
+                # لا يوجد استدعاء شبكي هنا
+                if preloaded_whale_data:
+                    base_analysis['whale_data'] = preloaded_whale_data.get(symbol, {'data_available': False, 'reason': 'Not preloaded'})
+                else:
+                     base_analysis['whale_data'] = {'data_available': False, 'reason': 'Preloading disabled'}
                 # 🔴 --- نهاية التعديل --- 🔴
 
-                # 4. استخدام محرك الاستراتيجيات (يعتمد على البيانات المحسوبة)
+                # حساب الاستراتيجيات والنتيجة النهائية (لا تغيير في المنطق هنا)
                 strategy_scores, base_scores = await self.strategy_engine.evaluate_all_strategies(base_analysis, self.market_context)
                 base_analysis['strategy_scores'] = strategy_scores
                 base_analysis['base_strategy_scores'] = base_scores
                 
-                # (بقية الكود كما هو)
                 if base_scores:
+                    # ... (الكود الخاص بتحديد أفضل استراتيجية كما هو) ...
                     best_strategy = max(base_scores.items(), key=lambda x: x[1])
                     best_strategy_name = best_strategy[0]
                     best_strategy_score = best_strategy[1]
                     base_analysis['recommended_strategy'] = best_strategy_name
                     base_analysis['strategy_confidence'] = best_strategy_score
-                    
-                    if best_strategy_score > 0.3: 
-                        base_analysis['target_strategy'] = best_strategy_name
-                    else: 
-                        base_analysis['target_strategy'] = 'GENERIC'
-                        
-                    # print(f"🎯 أفضل استراتيجية لـ {symbol}: {best_strategy_name} (ثقة: {best_strategy_score:.2f})") # Reduced logging
+                    base_analysis['target_strategy'] = best_strategy_name if best_strategy_score > 0.3 else 'GENERIC'
+
                 
                 enhanced_score = self._calculate_enhanced_final_score(base_analysis)
                 base_analysis['enhanced_final_score'] = enhanced_score
                 
-                # Reduced logging for success case
-                # print(f"✅ اكتمل التحليل المتقدم لـ {symbol}:")
-                # print(f"   📊 النهائي: {enhanced_score:.3f}")
-                # if monte_carlo_probability is not None:
-                #     print(f"   🎯 مونت كارلو: {monte_carlo_probability:.3f}")
-                # print(f"   🎯 نمط: {pattern_analysis.get('pattern_detected')} (ثقة: {pattern_analysis.get('pattern_confidence', 0):.2f})")
-                # if whale_data and whale_data.get('data_available'):
-                #     print(f"   🐋 حيتان: {whale_data.get('trading_signal', {}).get('action', 'HOLD')} (ثقة: {whale_data.get('trading_signal', {}).get('confidence', 0):.2f})")
-                
                 return base_analysis
                 
             except Exception as strategy_error:
                 print(f"❌ خطأ في التحليل المتقدم لـ {symbol}: {strategy_error}")
-                # Ensure base_analysis is returned even if advanced steps fail
                 return base_analysis 
                 
         except Exception as error:
             print(f"❌ خطأ فادح في المعالجة المحسنة للرمز {raw_data.get('symbol', 'unknown')}: {error}")
-            # Fallback to basic processing might hide the real issue, returning None is safer
-            return None # Changed from fallback to None
+            return None
+
+    # ...(بقية الدوال في الملف كما هي بدون تغيير، بما في ذلك _calculate_enhanced_final_score)...
 
-    # ...(بقية الدوال في الملف كما هي بدون تغيير)...
-    
     def _create_dataframe(self, candles):
         """إنشاء DataFrame من بيانات الشموع مع DatetimeIndex مرتب"""
         try:
-            # Added check for empty list
-            if not candles:
-                return pd.DataFrame()
+            if not candles: return pd.DataFrame()
             df = pd.DataFrame(candles, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
             df[['open', 'high', 'low', 'close', 'volume']] = df[['open', 'high', 'low', 'close', 'volume']].astype(float)
-            
-            # تحويل timestamp إلى datetime وضبطه كـ index
             df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
             df.set_index('timestamp', inplace=True)
-            
-            # التأكد من أن الفهرس مرتب
             df.sort_index(inplace=True)
-            
             return df
         except Exception as e:
             print(f"❌ خطأ في إنشاء DataFrame: {e}")
             return pd.DataFrame()
-    
+
     def _calculate_enhanced_final_score(self, analysis):
         """حساب الدرجة النهائية المحسنة"""
         try:
@@ -157,198 +123,132 @@ class MLProcessor:
             monte_carlo_score = analysis.get('monte_carlo_probability', 0)
             pattern_confidence = analysis.get('pattern_analysis', {}).get('pattern_confidence', 0)
             strategy_confidence = analysis.get('strategy_confidence', 0)
-            # 🔴 إضافة: عامل لبيانات الحيتان (إذا كانت متاحة وذات ثقة)
             whale_confidence = 0
             whale_data = analysis.get('whale_data')
             if whale_data and whale_data.get('data_available'):
                  signal = whale_data.get('trading_signal', {})
-                 # نعطي وزناً فقط للإشارات غير المحايدة وذات الثقة المتوسطة فأعلى
                  if signal.get('action') != 'HOLD' and signal.get('confidence', 0) >= 0.5:
                       whale_confidence = signal.get('confidence', 0)
 
-
             components = []
             weights = []
             
-            if base_score > 0:
-                components.append(base_score)
-                weights.append(0.20) # تقليل الوزن قليلاً
-            
-            if monte_carlo_score > 0:
-                components.append(monte_carlo_score)
-                weights.append(0.25) # تقليل الوزن قليلاً
-            
-            if pattern_confidence > 0:
-                components.append(pattern_confidence)
-                weights.append(0.25) # الحفاظ على الوزن
-            
-            if strategy_confidence > 0:
-                components.append(strategy_confidence)
-                weights.append(0.15) # تقليل الوزن قليلاً
+            if base_score > 0: components.append(base_score); weights.append(0.20)
+            if monte_carlo_score > 0: components.append(monte_carlo_score); weights.append(0.25)
+            if pattern_confidence > 0: components.append(pattern_confidence); weights.append(0.25)
+            if strategy_confidence > 0: components.append(strategy_confidence); weights.append(0.15)
+            if whale_confidence > 0: components.append(whale_confidence); weights.append(0.15)
 
-            # 🔴 إضافة: وزن لثقة الحيتان
-            if whale_confidence > 0:
-                components.append(whale_confidence)
-                weights.append(0.15) # إعطاء وزن معقول لبيانات الحيتان
-
-            if not components:
-                return 0  # ❌ لا توجد بيانات صالحة
+            if not components: return 0
             
-            # حساب المتوسط المرجح
             total_weight = sum(weights)
-            if total_weight == 0:
-                return 0
+            if total_weight == 0: return 0
                 
             enhanced_score = sum(comp * weight for comp, weight in zip(components, weights)) / total_weight
             
-            return min(max(enhanced_score, 0.0), 1.0) # Ensure score is between 0 and 1
+            return min(max(enhanced_score, 0.0), 1.0)
             
         except Exception as e:
             print(f"❌ خطأ في حساب الدرجة المحسنة: {e}")
-            return analysis.get('final_score', 0) # Fallback to base score
-    
+            return analysis.get('final_score', 0)
+
     async def process_and_score_symbol(self, raw_data):
         """المعالجة الأساسية للرمز"""
         try:
             symbol = raw_data['symbol']
-            ohlcv_data = raw_data.get('ohlcv') # Use get with default None
+            ohlcv_data = raw_data.get('ohlcv')
             
-            # Added check for ohlcv_data validity
-            if not ohlcv_data: 
-                # print(f"⚠️ بيانات OHLCV مفقودة للمعالجة الأساسية لـ {symbol}") # Reduced logging
-                return None
+            if not ohlcv_data: return None
             
             current_price = raw_data.get('current_price', 0)
             layer1_score = raw_data.get('layer1_score', 0)
             reasons = raw_data.get('reasons_for_candidacy', [])
-            
             final_score = layer1_score
             
             return {
-                'symbol': symbol,
-                'current_price': current_price,
-                'final_score': final_score,
-                'enhanced_final_score': final_score, # Initialize enhanced score
-                'reasons_for_candidacy': reasons,
-                'layer1_score': layer1_score,
-                # 🔴 إضافة: التأكد من تمرير بيانات OHLCV هنا أيضاً
-                'ohlcv': ohlcv_data
+                'symbol': symbol, 'current_price': current_price, 'final_score': final_score,
+                'enhanced_final_score': final_score, 'reasons_for_candidacy': reasons,
+                'layer1_score': layer1_score, 'ohlcv': ohlcv_data
             }
-            
         except Exception as error:
             print(f"❌ خطأ في المعالجة الأساسية للرمز {raw_data.get('symbol', 'unknown')}: {error}")
             return None
-    
+
     def filter_top_candidates(self, candidates, number_of_candidates=10):
         """تصفية أفضل المرشحين"""
-        valid_candidates = [candidate for candidate in candidates if candidate is not None and isinstance(candidate, dict)] # Added type check
-        
-        if not valid_candidates:
-            print("❌ لا توجد مرشحات صالحة للتصفية")
-            return []
-        
-        # Sort by 'enhanced_final_score', safely handling missing keys
-        sorted_candidates = sorted(valid_candidates, 
-                                 key=lambda candidate: candidate.get('enhanced_final_score', 0), 
-                                 reverse=True)
+        valid_candidates = [c for c in candidates if c is not None and isinstance(c, dict)]
+        if not valid_candidates: print("❌ لا توجد مرشحات صالحة للتصفية"); return []
         
+        sorted_candidates = sorted(valid_candidates, key=lambda c: c.get('enhanced_final_score', 0), reverse=True)
         top_candidates = sorted_candidates[:number_of_candidates]
         
         print(f"🎖️ أفضل {len(top_candidates)} مرشح:")
-        for i, candidate in enumerate(top_candidates):
-             # Safely get values using .get()
-            score = candidate.get('enhanced_final_score', 0)
-            strategy = candidate.get('recommended_strategy', 'GENERIC')
-            mc_score = candidate.get('monte_carlo_probability') # Allow None
-            pattern_analysis = candidate.get('pattern_analysis', {})
-            pattern = pattern_analysis.get('pattern_detected', 'no_pattern')
-            symbol = candidate.get('symbol', 'UNKNOWN') # Get symbol safely
-            
-            print(f"   {i+1}. {symbol}:")
-            print(f"      📊 النهائي: {score:.3f}")
-            if mc_score is not None: # Check if mc_score exists and is not None
-                print(f"      🎯 مونت كارلو: {mc_score:.3f}")
+        for i, c in enumerate(top_candidates):
+            score = c.get('enhanced_final_score', 0)
+            strategy = c.get('recommended_strategy', 'GENERIC')
+            mc_score = c.get('monte_carlo_probability')
+            pattern = c.get('pattern_analysis', {}).get('pattern_detected', 'no_pattern')
+            symbol = c.get('symbol', 'UNKNOWN')
+            
+            print(f"   {i+1}. {symbol}: 📊 {score:.3f}")
+            if mc_score is not None: print(f"      🎯 مونت كارلو: {mc_score:.3f}")
             print(f"      🎯 استراتيجية: {strategy} | نمط: {pattern}")
-            # 🔴 إضافة: عرض معلومات الحيتان إن وجدت
-            whale_data = candidate.get('whale_data')
+            whale_data = c.get('whale_data')
             if whale_data and whale_data.get('data_available'):
                  signal = whale_data.get('trading_signal', {})
-                 whale_action = signal.get('action', 'HOLD')
-                 whale_conf = signal.get('confidence', 0)
-                 print(f"      🐋 حيتان: {whale_action} (ثقة: {whale_conf:.2f})")
-
+                 print(f"      🐋 حيتان: {signal.get('action', 'HOLD')} (ثقة: {signal.get('confidence', 0):.2f})")
         return top_candidates
 
-# (safe_json_parse و process_multiple_symbols_parallel كما هي)
+# (safe_json_parse و process_multiple_symbols_parallel كما هي، لكن process_multiple_symbols_parallel لم تعد تُستخدم مباشرة في التدفق الجديد)
 def safe_json_parse(json_string):
-    """تحليل JSON آمن مع معالجة الأخطاء"""
-    # Added check for None or empty string
-    if not json_string:
-        return None
+    if not json_string: return None
     try:
         return json.loads(json_string)
     except json.JSONDecodeError as e:
         try:
-            # More robust cleaning attempts
-            json_string = str(json_string).replace("'", '"') # Ensure it's a string first
-            # Try removing potential invalid characters or sequences before regex
-            json_string = re.sub(r'\\"', '"', json_string) # Handle escaped quotes if needed
-            json_string = re.sub(r'[\n\t]', ' ', json_string) # Replace newlines/tabs
-            
-            # Make regex less greedy and more specific
-            json_string = re.sub(r'(?<!")(\b\w+\b)(?=\s*:)', r'"\1"', json_string) # Add quotes around keys like word:
-            json_string = re.sub(r':\s*(\btrue\b|\bfalse\b|\bnull\b)(?=[,\s}])', r': \1', json_string) # Ensure space after colon for bool/null
-            
-            # Handle trailing commas more carefully
-            json_string = re.sub(r',\s*([}\]])', r'\1', json_string)
-            
-            return json.loads(json_string)
-        except json.JSONDecodeError:
-            # print(f"❌ فشل تحليل JSON بعد الإصلاح: {e} - String: {json_string[:200]}...") # Reduced logging
-            return None
+            s = str(json_string).replace("'", '"')
+            s = re.sub(r'\\"', '"', s)
+            s = re.sub(r'[\n\t]', ' ', s)
+            s = re.sub(r'(?<!")(\b\w+\b)(?=\s*:)', r'"\1"', s)
+            s = re.sub(r':\s*(\btrue\b|\bfalse\b|\bnull\b)(?=[,\s}])', r': \1', s)
+            s = re.sub(r',\s*([}\]])', r'\1', s)
+            return json.loads(s)
+        except json.JSONDecodeError: return None
+
+async def process_multiple_symbols_parallel(self, symbols_data_list, preloaded_whale_data: dict, max_concurrent=5):
+    """(معدلة) معالجة متعددة للرموز بشكل متوازٍ باستخدام بيانات الحيتان المحملة مسبقًا"""
+    # 🔴 استخدام Semaphore داخلي للتحكم في مهام ML نفسها (وليس الحيتان)
+    semaphore = asyncio.Semaphore(max_concurrent)
+    tasks_results = []
 
+    async def process_symbol_with_semaphore(symbol_data):
+        async with semaphore:
+            try:
+                # تمرير بيانات الحيتان المحملة مسبقًا
+                return await self.process_and_score_symbol_enhanced(symbol_data, preloaded_whale_data)
+            except Exception as e:
+                return e
 
-async def process_multiple_symbols_parallel(self, symbols_data_list, max_concurrent=20):
-    """معالجة متعددة للرموز بشكل متوازي مع التحكم في التزامن"""
     try:
-        print(f"🚀 بدء المعالجة المتوازية لـ {len(symbols_data_list)} رمز (بحد أقصى {max_concurrent} متزامنة)...")
+        # print(f"🚀 بدء المعالجة المتوازية لـ {len(symbols_data_list)} رمز (ML بحد أقصى {max_concurrent})...") # Reduced logging
         
-        batches = [symbols_data_list[i:i + max_concurrent] 
-                  for i in range(0, len(symbols_data_list), max_concurrent)]
+        batch_tasks = [asyncio.create_task(process_symbol_with_semaphore(sd)) for sd in symbols_data_list]
+        batch_results = await asyncio.gather(*batch_tasks, return_exceptions=False)
+            
+        successful_results = []
+        for result in batch_results:
+            if isinstance(result, Exception):
+                 # print(f"   ⚠️ خطأ في مهمة ML متوازية: {result}") # Reduced logging
+                 continue
+            if isinstance(result, dict) and result.get('enhanced_final_score', 0) > 0.4: 
+                successful_results.append(result)
         
-        all_results = []
-        
-        for batch_num, batch in enumerate(batches):
-            print(f"   🔄 معالجة الدفعة {batch_num + 1}/{len(batches)} ({len(batch)} رمز)...")
-            
-            batch_tasks = []
-            for symbol_data in batch:
-                # 🔴 استخدام المهمة المعدلة مع منظم سرعة الحيتان
-                task = asyncio.create_task(self.process_and_score_symbol_enhanced(symbol_data))
-                batch_tasks.append(task)
-            
-            batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True)
-            
-            successful_results = []
-            for result in batch_results:
-                if isinstance(result, Exception):
-                    # print(f"   ⚠️ خطأ في مهمة متوازية: {result}") # Reduced logging
-                    continue
-                # Check result is a dict and has the score key
-                if isinstance(result, dict) and result.get('enhanced_final_score', 0) > 0.4: 
-                    successful_results.append(result)
-            
-            all_results.extend(successful_results)
-            print(f"   ✅ اكتملت الدفعة {batch_num + 1}: {len(successful_results)}/{len(batch)} ناجحة")
-            
-            if batch_num < len(batches) - 1:
-                await asyncio.sleep(1)
-        
-        print(f"🎯 اكتملت المعالجة المتوازية: {len(all_results)}/{len(symbols_data_list)} رمز تم تحليلها بنجاح")
-        return all_results
+        # print(f"🎯 اكتملت المعالجة المتوازية ل��مجموعة: {len(successful_results)}/{len(symbols_data_list)} ناجحة") # Reduced logging
+        return successful_results
         
     except Exception as error:
-        print(f"❌ خطأ في المعالجة المتوازية: {error}")
+        print(f"❌ خطأ في المعالجة المتوازية لـ ML: {error}")
         return []
 
-print("✅ ML Processor loaded - Whale Semaphore Added & Scoring Adjusted")
+
+print("✅ ML Processor loaded - Non-Blocking Whale Data Integration")