Update data_manager.py

data_manager.py

@@ -1,4 +1,4 @@
-# data_manager.py (Updated to V7.1 - 1H Detector Screening)
+# data_manager.py (Updated to V7.2 - Fixed Task Bug & Added Stablecoin Guard)
 import os
 import asyncio
 import httpx
@@ -11,19 +11,15 @@ import logging
 from typing import List, Dict, Any
 import pandas as pd
 
-# 🔴 --- START OF CHANGE (V7.1) --- 🔴
-# (استيراد الأدوات اللازمة لـ "الكاشف المصغر")
 try:
     import pandas_ta as ta
 except ImportError:
     print("⚠️ مكتبة pandas_ta غير موجودة، فلتر الغربلة المتقدم سيفشل.")
     ta = None
 
-# (استيراد الوحدات الأساسية من محرك ML لتشغيل الكاشف المصغر)
 from ml_engine.indicators import AdvancedTechnicalAnalyzer
 from ml_engine.monte_carlo import MonteCarloAnalyzer
 from ml_engine.patterns import ChartPatternAnalyzer
-# 🔴 --- END OF CHANGE --- 🔴
 
 logging.getLogger("httpx").setLevel(logging.WARNING)
 logging.getLogger("httpcore").setLevel(logging.WARNING)
@@ -49,17 +45,14 @@ class DataManager:
         self.market_cache = {}
         self.last_market_load = None
 
-        # 🔴 --- START OF CHANGE (V7.1) --- 🔴
-        # (تهيئة أدوات الكاشف المصغر)
         self.technical_analyzer = AdvancedTechnicalAnalyzer()
         self.monte_carlo_analyzer = MonteCarloAnalyzer()
         self.pattern_analyzer = ChartPatternAnalyzer()
-        # 🔴 --- END OF CHANGE --- 🔴
         
     async def initialize(self):
         self.http_client = httpx.AsyncClient(timeout=30.0)
         await self._load_markets()
-        print("✅ DataManager initialized - V7.1 (1H Detector Screening)")
+        print("✅ DataManager initialized - V7.2 (Stablecoin Guard Active)")
 
     async def _load_markets(self):
         try:
@@ -205,10 +198,10 @@ class DataManager:
             'data_quality': 'LOW'
         }
 
-    # 🔴 --- START OF REFACTOR (V7.1 - 1H Detector Screening) --- 🔴
+    # 🔴 --- START OF REFACTOR (V7.2 - Fixed Bugs) --- 🔴
     
     def _create_dataframe(self, candles: List) -> pd.DataFrame:
-        """(جديد V7.1) دالة مساعدة لإنشاء DataFrame لتحليل 1H"""
+        """(V7.1) دالة مساعدة لإنشاء DataFrame لتحليل 1H"""
         try:
             if not candles: return pd.DataFrame()
             df = pd.DataFrame(candles, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
@@ -223,11 +216,22 @@ class DataManager:
 
     def _calculate_1h_filter_score(self, analysis: Dict) -> float:
         """
-        (جديد V7.1) - "الكاشف المصغر"
-        مقتبس من processor.py لحساب درجة (0-1) بناءً على بيانات 1H فقط.
-        يستخدم (المؤشرات، مونت كارلو، الأنماط) فقط.
+        (محدث V7.2) - "الكاشف المصغر"
+        يحتوي الآن على "واقي العملات المستقرة"
         """
         try:
+            # 🔴 --- (FIX V7.2) واقي العملات المستقرة --- 🔴
+            # (يتم تشغيله قبل أي حساب آخر)
+            if 'ohlcv_1h' in analysis and '1h' in analysis['ohlcv_1h']:
+                closes_1h = [c[4] for c in analysis['ohlcv_1h']['1h']]
+                if len(closes_1h) > 20:
+                    # (نستخدم آخر 20 شمعة للتحقق من التقلب)
+                    std_dev = np.std(closes_1h[-20:])
+                    if std_dev < 1e-5: # (تقلب شبه صفري)
+                        print(f"      - {analysis.get('symbol', 'N/A')}: تم الاستبعاد (عملة مستقرة)")
+                        return 0.0 # (إعطاء درجة صفر فوراً)
+            # 🔴 --- نهاية الإصلاح --- 🔴
+
             # 1. درجة الأنماط (Pattern Score)
             pattern_confidence = analysis.get('pattern_analysis', {}).get('pattern_confidence', 0)
             
@@ -235,7 +239,7 @@ class DataManager:
             mc_distribution = analysis.get('monte_carlo_distribution')
             monte_carlo_score = 0
             
-            if mc_distribution:
+            if mc_distribution and mc_distribution.get('error') is None: # (تحقق من عدم وجود خطأ)
                 prob_gain = mc_distribution.get('probability_of_gain', 0)
                 var_95_value = mc_distribution.get('risk_metrics', {}).get('VaR_95_value', 0)
                 current_price = analysis.get('current_price', 1)
@@ -246,12 +250,10 @@ class DataManager:
                     if normalized_var > 0.05: risk_penalty = 0.5           
                     elif normalized_var > 0.03: risk_penalty = 0.8
                     
-                    # (تطبيع الاحتمالية: 0.5 -> 0, 1.0 -> 1.0)
                     normalized_prob_score = max(0.0, (prob_gain - 0.5) * 2)
                     monte_carlo_score = normalized_prob_score * risk_penalty
             
             # 3. درجة المؤشرات (Indicator Score)
-            # (تقييم بسيط للمؤشرات الرئيسية لـ 1H)
             indicator_score = 0
             indicators = analysis.get('advanced_indicators', {}).get('1h', {})
             if indicators:
@@ -268,7 +270,6 @@ class DataManager:
                     indicator_score = min(0.4 + (35 - rsi) / 35, 0.8)
 
             # 4. حساب النتيجة النهائية (بدون استراتيجيات أو حيتان)
-            # الأوزان: مونت كارلو (40%)، الأنماط (30%)، المؤشرات (30%)
             components = []
             weights = []
             
@@ -290,14 +291,9 @@ class DataManager:
 
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
         """
-        الطبقة 1: فحص سريع - (محدث بالكامل V7.1)
-        1. جلب أفضل 100 عملة حسب الحجم (24 ساعة).
-        2. جلب 100 شمعة (1H) لهذه الـ 100 عملة.
-        3. تشغيل "الكاشف المصغر" (مؤشرات، مونت كارلو، أنماط) على 1H.
-        4. استبعاد أي عملة درجتها < 0.20.
-        5. إرجاع العملات الناجحة فقط للطبقة 2.
+        الطبقة 1: فحص سريع - (محدث بالكامل V7.2)
         """
-        print("📊 الطبقة 1 (V7.1): بدء الغربلة (الكاشف المصغر 1H)...")
+        print("📊 الطبقة 1 (V7.2): بدء الغربلة (الكاشف المصغر 1H)...")
         
         # الخطوة 1: جلب أفضل 100 عملة حسب الحجم
         volume_data = await self._get_volume_data_optimal()
@@ -315,7 +311,6 @@ class DataManager:
         
         final_candidates = []
         
-        # (استخدام دفعات لمعالجة الـ 100 عملة بكفاءة)
         batch_size = 20
         for i in range(0, len(top_100_by_volume), batch_size):
             batch_symbols_data = top_100_by_volume[i:i + batch_size]
@@ -327,7 +322,6 @@ class DataManager:
             tasks = [self._fetch_1h_ohlcv_for_screening(symbol) for symbol in batch_symbols]
             results_candles = await asyncio.gather(*tasks, return_exceptions=True)
             
-            # الخطوة 3 و 4: تطبيق الكاشف المصغر
             analysis_tasks = []
             valid_symbol_data = []
 
@@ -340,43 +334,45 @@ class DataManager:
                 
                 # (إعداد البيانات للمحللات)
                 ohlcv_1h_only = {'1h': candles}
-                symbol_data['ohlcv_1h'] = ohlcv_1h_only # (تخزين مؤقت)
-                symbol_data['current_price'] = candles[-1][4] # آخر سعر إغلاق
+                symbol_data['ohlcv_1h'] = ohlcv_1h_only 
+                symbol_data['current_price'] = candles[-1][4]
                 analysis_tasks.append(self._run_mini_detector(symbol_data))
                 valid_symbol_data.append(symbol_data)
 
             if not analysis_tasks:
                 continue
                 
-            # تشغيل التحليلات (MC, Patterns, Indicators) بالتوازي
             analysis_results = await asyncio.gather(*analysis_tasks, return_exceptions=True)
 
             for j, (analysis_output) in enumerate(analysis_results):
-                symbol_data = valid_symbol_data[j] # (استرجاع بيانات الرمز المطابقة)
+                symbol_data = valid_symbol_data[j]
                 symbol = symbol_data['symbol']
 
                 if isinstance(analysis_output, Exception):
                     print(f"      - {symbol}: فشل الكاشف المصغر ({analysis_output})")
                     continue
                 
+                # (تمرير ohlcv_1h إلى دالة حساب النقاط من أجل واقي العملات المستقرة)
+                analysis_output['ohlcv_1h'] = symbol_data['ohlcv_1h']
+                analysis_output['symbol'] = symbol
+                
                 # الخطوة 4: حساب النتيجة النهائية للفلتر
                 filter_score = self._calculate_1h_filter_score(analysis_output)
                 
                 # الخطوة 5: تطبيق العتبة (Threshold)
                 if filter_score >= 0.20:
                     print(f"      ✅ {symbol}: نجح (الدرجة: {filter_score:.2f})")
-                    # (إضافة البيانات الأولية للطبقة 2)
                     symbol_data['layer1_score'] = filter_score
                     symbol_data['reasons_for_candidacy'] = [f'1H_DETECTOR_PASS']
                     
-                    # (تنظيف البيانات قبل إرسالها)
                     if 'ohlcv_1h' in symbol_data: del symbol_data['ohlcv_1h'] 
                     
                     final_candidates.append(symbol_data)
                 # else:
+                #    (تم إيقاف طباعة الفشل لتقليل التشويش)
                 #    print(f"      - {symbol}: فشل (الدرجة: {filter_score:.2f})")
 
-        print(f"🎯 اكتملت الغربلة (V7.1). تم تأهيل {len(final_candidates)} عملة من أصل 100 لل��بقة 2.")
+        print(f"🎯 اكتملت الغربلة (V7.2). تم تأهيل {len(final_candidates)} عملة من أصل 100 للطبقة 2.")
         
         print("🏆 المرشحون الناجحون:")
         for k, candidate in enumerate(final_candidates[:15]):
@@ -387,30 +383,24 @@ class DataManager:
         return final_candidates
 
     async def _run_mini_detector(self, symbol_data: Dict) -> Dict:
-        """(جديد V7.1) يشغل المحللات الأساسية بالتوازي على بيانات 1H فقط"""
+        """(V7.1) يشغل المحللات الأساسية بالتوازي على بيانات 1H فقط"""
         ohlcv_1h = symbol_data.get('ohlcv_1h')
         current_price = symbol_data.get('current_price')
         
-        # (إنشاء DataFrame مرة واحدة)
         df = self._create_dataframe(ohlcv_1h.get('1h'))
         if df.empty:
             raise ValueError("DataFrame فارغ لتحليل 1H")
             
         analysis_dict = {'current_price': current_price}
 
-        # إنشاء المهام
-        # (لا حاجة لـ asyncio.create_task لأن هذه الدوال ليست async)
         # (ملاحظة: جعلنا دوال التحليل async في ملفاتها لتكون متوافقة)
-        
         task_indicators = self.technical_analyzer.calculate_all_indicators(df, '1h')
         task_mc = self.monte_carlo_analyzer.generate_1h_price_distribution(ohlcv_1h)
         task_pattern = self.pattern_analyzer.detect_chart_patterns(ohlcv_1h)
 
-        # (تشغيل المهام التي هي async)
         results = await asyncio.gather(task_mc, task_pattern, return_exceptions=True)
         
-        # (معالجة النتائج)
-        analysis_dict['advanced_indicators'] = {'1h': task_indicators} # (هذه دالة متزامنة)
+        analysis_dict['advanced_indicators'] = {'1h': task_indicators}
         
         if not isinstance(results[0], Exception):
             analysis_dict['monte_carlo_distribution'] = results[0]
@@ -421,20 +411,16 @@ class DataManager:
 
 
     async def _fetch_1h_ohlcv_for_screening(self, symbol: str) -> List:
-        """(جديد V7.1) جلب 100 شمعة لإطار الساعة (1H) للغربلة السريعة"""
+        """(V7.1) جلب 100 شمعة لإطار الساعة (1H) للغربلة السريعة"""
         try:
-            # 100 شمعة كافية لحساب (RSI 14, EMA 21, MACD 26, BB 20)
             ohlcv_data = self.exchange.fetch_ohlcv(symbol, '1h', limit=100)
             
-            if not ohlcv_data or len(ohlcv_data) < 50: # (الحد الأدنى 50)
+            if not ohlcv_data or len(ohlcv_data) < 50:
                 return None
             return ohlcv_data
         except Exception:
             return None
 
-    # 🔴 --- END OF REFACTOR (V7.1) --- 🔴
-
-
     # (دوال جلب الحجم تبقى كما هي لأنها فعالة)
     async def _get_volume_data_optimal(self) -> List[Dict[str, Any]]:
         try:
@@ -453,8 +439,13 @@ class DataManager:
                     if base_volume is None: continue
                     dollar_volume = base_volume * current_price
                 if dollar_volume is None or dollar_volume < 50000: continue
-                price_change_24h = (ticker.get('percentage', 0) or 0) * 100
+                
+                # 🔴 --- (FIX V7.2) إصلاح نسبة التغير المئوية --- 🔴
+                # (ccxt 'percentage' هي النسبة الفعلية، لا تحتاج للضرب في 100)
+                price_change_24h = ticker.get('percentage', 0) or 0
                 if price_change_24h is None: price_change_24h = 0
+                # 🔴 --- نهاية الإصلاح --- 🔴
+
                 volume_data.append({
                     'symbol': symbol, 'dollar_volume': dollar_volume,
                     'current_price': current_price, 'volume_24h': ticker.get('baseVolume', 0) or 0,
@@ -488,7 +479,10 @@ class DataManager:
                         if vol_value is None or last_price is None or change_rate is None or vol is None: continue
                         dollar_volume = float(vol_value) if vol_value else 0
                         current_price = float(last_price) if last_price else 0
+                        
+                        # (الإصلاح هنا أيضاً: 'changeRate' هو النسبة العشرية، لذا نضرب في 100)
                         price_change = (float(change_rate) * 100) if change_rate else 0
+                        
                         volume_24h = float(vol) if vol else 0
                         if dollar_volume >= 50000 and current_price > 0:
                             volume_data.append({
@@ -504,8 +498,9 @@ class DataManager:
             return []
 
     # (دالة تدفق الشموع تبقى كما هي - لا تغيير)
-    async def stream_ohlcv_data(self, symbols: List[str], queue: asyncio.Queue):
+    async def stream_ohlcv_data(self, symbols: List[Dict[str, Any]], queue: asyncio.Queue):
         """
+        (محدث V7.2)
         جلب بيانات OHLCV كاملة (6 أطر زمنية) للعملات الناجحة فقط
         """
         print(f"📊 بدء تدفق بيانات OHLCV (الكاملة) لـ {len(symbols)} عملة (الناجحين من الغربلة)...")
@@ -519,28 +514,37 @@ class DataManager:
             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)
             
+            # 🔴 --- (FIX V7.2) إصلاح تمرير الوسيط --- 🔴
+            for symbol_data in batch:
+                symbol_str = symbol_data['symbol'] # (استخراج النص)
+                # (تمرير النص فقط إلى الدالة التي تتوقع نصاً)
+                task = asyncio.create_task(self._fetch_complete_ohlcv_parallel(symbol_str))
+                batch_tasks.append(task)
+            # 🔴 --- نهاية الإصلاح --- 🔴
+
             batch_results = await asyncio.gather(*batch_tasks, return_exceptions=True)
             
             successful_data_for_batch = []
             successful_count = 0
             for i, result in enumerate(batch_results):
-                symbol = batch[i]
+                
+                # 🔴 --- (FIX V7.2) مطابقة النتائج بالبيانات الأصلية --- 🔴
+                original_symbol_data = batch[i]
+                symbol_str = original_symbol_data['symbol']
+                # 🔴 --- نهاية الإصلاح --- 🔴
+
                 if isinstance(result, Exception):
-                    print(f"      ❌ [المنتج] فشل جلب {symbol}: {result}")
+                    print(f"      ❌ [المنتج] فشل جلب {symbol_str}: {result}")
                 elif result is not None:
-                    # (مهم: نمرر بيانات المرشح الأولية التي تحتوي على الدرجة)
-                    original_data = next((s for s in symbols if s['symbol'] == symbol), {'symbol': symbol})
-                    result.update(original_data) # (دمج الدرجة الأولية مع البيانات الكاملة)
+                    # (دمج الدرجة الأولية مع البيانات الكاملة)
+                    result.update(original_symbol_data) 
                     successful_data_for_batch.append(result)
                     successful_count += 1
                     timeframes_count = result.get('successful_timeframes', 0)
-                    print(f"      ✅ [المنتج] {symbol}: {timeframes_count}/6 أطر زمنية")
+                    print(f"      ✅ [المنتج] {symbol_str}: {timeframes_count}/6 أطر زمنية")
                 else:
-                    print(f"      ⚠️ [المنتج] {symbol}: بيانات غير كافية، تم التجاهل")
+                    print(f"      ⚠️ [المنتج] {symbol_str}: بيانات غير كافية، تم التجاهل")
 
             print(f"      📦 [المنتج] اكتملت الدفعة {batch_num + 1}: {successful_count}/{len(batch)} ناجحة")
             
@@ -565,7 +569,7 @@ class DataManager:
 
 
     async def _fetch_complete_ohlcv_parallel(self, symbol: str) -> Dict[str, Any]:
-        """جلب بيانات OHLCV كاملة مع جميع الإطارات الزمنية بشكل متوازي"""
+        """(V7.2) جلب بيانات OHLCV كاملة - يتوقع 'symbol' كنص"""
         try:
             ohlcv_data = {}
             
@@ -576,6 +580,7 @@ class DataManager:
             
             timeframe_tasks = []
             for timeframe, limit in timeframes:
+                # (هنا 'symbol' هو نص، وهو صحيح)
                 task = asyncio.create_task(self._fetch_single_timeframe_improved(symbol, timeframe, limit))
                 timeframe_tasks.append(task)
             
@@ -614,11 +619,12 @@ class DataManager:
         except Exception as e: return None
 
     async def _fetch_single_timeframe_improved(self, symbol: str, timeframe: str, limit: int):
-        """جلب بيانات إطار زمني واحد مع تحسين التعامل مع الأخطاء"""
+        """(V7.2) جلب بيانات إطار زمني واحد - يتوقع 'symbol' كنص"""
         max_retries = 3
         retry_delay = 2
         for attempt in range(max_retries):
             try:
+                # (هنا 'symbol' هو نص، وهو صحيح)
                 ohlcv_data = self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
                 if ohlcv_data and len(ohlcv_data) > 0:
                     return ohlcv_data
@@ -631,7 +637,7 @@ class DataManager:
                     return []
 
     async def get_latest_price_async(self, symbol):
-        """جلب السعر الحالي لعملة محددة"""
+        """(V7.2) جلب السعر الحالي - يتوقع 'symbol' كنص"""
         try:
             if not self.exchange: return None
             if not symbol or '/' not in symbol: return None
@@ -660,4 +666,4 @@ class DataManager:
         except Exception as e:
             return {'action': 'HOLD', 'confidence': 0.3, 'reason': f'Error: {str(e)}', 'source': 'whale_analysis'}
 
-print("✅ DataManager loaded - V7.1 (1H Detector Screening)")
+print("✅ DataManager loaded - V7.2 (Stablecoin Guard Active + Task Bug Fixed)")