Update ml_engine/processor.py

ml_engine/processor.py

@@ -22,11 +22,8 @@ class MLProcessor:
         self.strategy_engine = MultiStrategyEngine(data_manager, learning_engine)
         self.monte_carlo_analyzer = MonteCarloAnalyzer()
         self.pattern_analyzer = ChartPatternAnalyzer()
+        self.whale_data_semaphore = asyncio.Semaphore(2) 
 
-        # 🔴 تم إزالة منظم سرعة الحيتان من هنا
-        # self.whale_data_semaphore = asyncio.Semaphore(1) 
-
-    # 🔴 --- تعديل جوهري: إضافة معلمة جديدة واستخدامها --- 🔴
     async def process_and_score_symbol_enhanced(self, raw_data, preloaded_whale_data: dict = None):
         """
         (معدلة) المعالجة المحسنة للرموز
@@ -44,7 +41,7 @@ class MLProcessor:
                 return None
             
             try:
-                # حساب المؤشرات (لا تغيير)
+                # حساب المؤشرات
                 advanced_indicators = {}
                 ohlcv_available = raw_data.get('ohlcv', {}) 
                 for timeframe, candles in ohlcv_available.items(): 
@@ -54,38 +51,34 @@ class MLProcessor:
                         advanced_indicators[timeframe] = indicators
                 base_analysis['advanced_indicators'] = advanced_indicators
                 
-                # حساب مونت كارلو (لا تغيير)
+                # حساب مونت كارلو
                 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
                 
-                # حساب الأنماط (لا تغيير)
+                # حساب الأنماط
                 pattern_analysis = await self.pattern_analyzer.detect_chart_patterns(ohlcv_available) 
                 base_analysis['pattern_analysis'] = pattern_analysis
                 
-                # 🔴 --- تعديل: استخدام بيانات الحيتان المحملة مسبقاً --- 🔴
-                # لا يوجد استدعاء شبكي هنا
+                # استخدام بيانات الحيتان المحملة مسبقاً
                 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'}
-                # 🔴 --- نهاية التعديل --- 🔴
 
-                # حساب الاستراتيجيات والنتيجة النهائية (لا تغيير في المنطق هنا)
+                # حساب الاستراتيجيات والنتيجة النهائية
                 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
                     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
@@ -100,8 +93,6 @@ class MLProcessor:
             print(f"❌ خطأ فادح في المعالجة المحسنة للرمز {raw_data.get('symbol', 'unknown')}: {error}")
             return None
 
-    # ...(بقية الدوال في الملف كما هي بدون تغيير، بما في ذلك _calculate_enhanced_final_score)...
-
     def _create_dataframe(self, candles):
         """إنشاء DataFrame من بيانات الشموع مع DatetimeIndex مرتب"""
         try:
@@ -199,7 +190,45 @@ class MLProcessor:
                  print(f"      🐋 حيتان: {signal.get('action', 'HOLD')} (ثقة: {signal.get('confidence', 0):.2f})")
         return top_candidates
 
-# (safe_json_parse و process_multiple_symbols_parallel كما هي، لكن process_multiple_symbols_parallel لم تعد تُستخدم مباشرة في التدفق الجديد)
+    # 🔴 --- بدء الإصلاح: نقل الدالة إلى داخل الكلاس --- 🔴
+    # تأكد من أن هذه الدالة بنفس مستوى المسافة الب��دئة مثل الدوال الأخرى في الكلاس
+    async def process_multiple_symbols_parallel(self, symbols_data_list, preloaded_whale_data: dict, max_concurrent=5):
+        """(معدلة) معالجة متعددة للرموز بشكل متوازٍ باستخدام بيانات الحيتان المحملة مسبقًا"""
+        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 # Return the exception itself
+
+        try:
+            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) # Capture results/exceptions manually
+                
+            successful_results = []
+            for result in batch_results:
+                if isinstance(result, Exception):
+                    # Propagate the exception to be handled by the caller (process_batch_parallel in app.py)
+                    # This makes error reporting more consistent.
+                    raise result 
+                # Check result is a dict before accessing keys
+                if isinstance(result, dict): # Removed score check here, done by caller
+                    successful_results.append(result)
+                # else: # Optional: Handle unexpected return types if necessary
+                #    print(f"Warning: Unexpected type returned by process_symbol_with_semaphore: {type(result)}")
+
+            return successful_results # Return list of results (dicts)
+            
+        except Exception as error: # Catch errors raised from gather or process_symbol_with_semaphore
+            # Re-raise the error to be caught by process_batch_parallel in app.py
+            raise error 
+    # 🔴 --- نهاية الإصلاح --- 🔴
+
+# (safe_json_parse يمكن أن تبقى خارج الكلاس لأنها دالة مساعدة عامة)
 def safe_json_parse(json_string):
     if not json_string: return None
     try:
@@ -215,40 +244,4 @@ def safe_json_parse(json_string):
             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
-
-    try:
-        # print(f"🚀 بدء المعالجة المتوازية لـ {len(symbols_data_list)} رمز (ML بحد أقصى {max_concurrent})...") # Reduced logging
-        
-        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)
-        
-        # print(f"🎯 اكتملت المعالجة المتوازية للمجموعة: {len(successful_results)}/{len(symbols_data_list)} ناجحة") # Reduced logging
-        return successful_results
-        
-    except Exception as error:
-        print(f"❌ خطأ في المعالجة المتوازية لـ ML: {error}")
-        return []
-
-
-print("✅ ML Processor loaded - Non-Blocking Whale Data Integration")
+print("✅ ML Processor loaded - Non-Blocking Whale Data Integration & Fixed Method Definition")