Update data_manager.py

data_manager.py

@@ -1,4 +1,4 @@
-# data_manager.py (Updated to V7.9 - Re-balanced Spark Filter Weights)
+# data_manager.py (Updated to V9.3 - High Confidence ML Ranker)
 import os
 import asyncio
 import httpx
@@ -14,30 +14,29 @@ import pandas as pd
 try:
     import pandas_ta as ta
 except ImportError:
-    print("⚠️ مكتبة pandas_ta غير موجودة، فلتر الغربلة المتقدم سيفشل.")
+    print("⚠️ مكتبة pandas_ta غير موجودة. النظام سيفشل.")
     ta = None
 
+# (V9.1) استيراد العقل الحسابي (الذي يحتوي على V9.1 Features)
 from ml_engine.indicators import AdvancedTechnicalAnalyzer
 from ml_engine.monte_carlo import MonteCarloAnalyzer
-# (V8-MODIFICATION) استيراد المحرك الصحيح (V8)
 from ml_engine.patterns import ChartPatternAnalyzer
+# (V9.1) استيراد "العقل الذكي" الجديد
+from ml_engine.ranker import Layer1Ranker 
 
 logging.getLogger("httpx").setLevel(logging.WARNING)
 logging.getLogger("httpcore").setLevel(logging.WARNING)
 
 class DataManager:
-    # (V8-MODIFICATION) قبول r2_service
     def __init__(self, contracts_db, whale_monitor, r2_service=None):
         self.contracts_db = contracts_db or {}
         self.whale_monitor = whale_monitor
-        self.r2_service = r2_service # (V8-MODIFICATION) الإضافة الجديدة
+        self.r2_service = r2_service
         
         try:
             self.exchange = ccxt.kucoin({
-                'sandbox': False,
-                'enableRateLimit': True,
-                'timeout': 30000,
-                'verbose': False,
+                'sandbox': False, 'enableRateLimit': True,
+                'timeout': 30000, 'verbose': False,
             })
             print("✅ تم تهيئة اتصال KuCoin بنجاح")
         except Exception as e:
@@ -48,100 +47,88 @@ class DataManager:
         self.market_cache = {}
         self.last_market_load = None
 
+        # (V9.1) استيراد العقل الحسابي (الذي يحتوي على V9.1 Features)
         self.technical_analyzer = AdvancedTechnicalAnalyzer()
         self.monte_carlo_analyzer = MonteCarloAnalyzer()
-        self.pattern_analyzer = None # (V8-MODIFICATION) سيتم تهيئته في initialize
+        self.pattern_analyzer = None
+        # (V9.1) تهيئة "العقل الذكي" (النموذج)
+        self.layer1_ranker = None 
         
     async def initialize(self):
         self.http_client = httpx.AsyncClient(timeout=30.0)
         await self._load_markets()
         
-        # (V8-MODIFICATION) تهيئة محرك الأنماط V8 (ML-Based)
         print("   > [DataManager] تهيئة محرك الأنماط V8 (ML-Based)...")
         try:
             self.pattern_analyzer = ChartPatternAnalyzer(r2_service=self.r2_service)
-            await self.pattern_analyzer.initialize() # (تحميل النموذج والمقياس من R2)
+            await self.pattern_analyzer.initialize()
         except Exception as e:
             print(f"❌ [DataManager] فشل تهيئة محرك الأنماط V8: {e}")
-            # (العودة للوضع الآمن إذا فشل التحميل)
             self.pattern_analyzer = ChartPatternAnalyzer(r2_service=None)
-        # --- (نهاية الإضافة) ---
+            
+        # 🔴 --- START OF CHANGE (V9.1) --- 🔴
+        # (تهيئة "العقل الذكي" - نموذج الرانكر V9.1)
+        print("   > [DataManager] تهيئة الكاشف المصغر (Layer1 Ranker V9.1)...")
+        try:
+            # (نتوقع أن يكون النموذج موجوداً في هذا المسار)
+            model_file_path = "ml_models/layer1_ranker.lgbm"
+            self.layer1_ranker = Layer1Ranker(model_path=model_file_path)
+            await self.layer1_ranker.initialize()
+            if self.layer1_ranker.model is None:
+                print("   ⚠️ [DataManager V9.1] الرانكر في وضع 'وهمي' (Placeholder).")
+            else:
+                 print(f"   ✅ [DataManager V9.1] الرانكر {self.layer1_ranker.model_name} جاهز للعمل.")
+        except Exception as e:
+            print(f"❌ [DataManager V9.1] فشل تهيئة الرانكر V9.1: {e}")
+            self.layer1_ranker = None 
+        # 🔴 --- END OF CHANGE (V9.1) --- 🔴
 
-        print("✅ DataManager initialized - V7.9 (Re-balanced Spark Weights)")
+        print("✅ DataManager initialized - V9.3 (High Confidence Ranker)")
 
     async def _load_markets(self):
         try:
-            if not self.exchange:
-                return
-                
+            if not self.exchange: return
             print("🔄 جلب أحدث بيانات الأسواق من KuCoin...")
             self.exchange.load_markets()
             self.market_cache = self.exchange.markets
             self.last_market_load = datetime.now()
             print(f"✅ تم تحميل {len(self.market_cache)} سوق من KuCoin")
-            
-        except Exception as e:
-            print(f"❌ ف��ل تحميل بيانات الأسواق: {e}")
+        except Exception as e: print(f"❌ فشل تحميل بيانات الأسواق: {e}")
 
     async def close(self):
-        if self.http_client and not self.http_client.is_closed: 
-            await self.http_client.aclose()
-            print("   ✅ DataManager: http_client closed.")
-            
+        if self.http_client and not self.http_client.is_closed: await self.http_client.aclose()
         if self.exchange:
-            try:
-                await self.exchange.close()
-                print("   ✅ DataManager: ccxt.kucoin exchange closed.")
-            except Exception as e:
-                print(f"   ⚠️ DataManager: Error closing ccxt.kucoin: {e}")
-        
+            try: await self.exchange.close()
+            except Exception: pass
+
     async def get_market_context_async(self):
         try:
             sentiment_data = await self.get_sentiment_safe_async()
             price_data = await self._get_prices_with_fallback()
-            
-            bitcoin_price = price_data.get('bitcoin')
-            ethereum_price = price_data.get('ethereum')
-            
-            market_context = {
+            bitcoin_price = price_data.get('bitcoin'); ethereum_price = price_data.get('ethereum')
+            return {
                 'timestamp': datetime.now().isoformat(),
-                'bitcoin_price_usd': bitcoin_price,
-                'ethereum_price_usd': ethereum_price,
+                'bitcoin_price_usd': bitcoin_price, 'ethereum_price_usd': ethereum_price,
                 'fear_and_greed_index': sentiment_data.get('feargreed_value') if sentiment_data else None,
                 'sentiment_class': sentiment_data.get('feargreed_class') if sentiment_data else 'NEUTRAL',
                 'market_trend': self._determine_market_trend(bitcoin_price, sentiment_data),
                 'btc_sentiment': self._get_btc_sentiment(bitcoin_price),
                 'data_quality': 'HIGH' if bitcoin_price and ethereum_price else 'LOW'
             }
-            
-            return market_context
-                
-        except Exception as e:
-            return self._get_minimal_market_context()
+        except Exception as e: return self._get_minimal_market_context()
 
+    # ... (دوال get_sentiment_safe_async, _determine_market_trend... إلخ - لا تغيير) ...
     async def get_sentiment_safe_async(self):
         try:
             async with httpx.AsyncClient(timeout=10) as client:
                 response = await client.get("https://api.alternative.me/fng/")
-                response.raise_for_status()
-                data = response.json()
-                
-                if 'data' not in data or not data['data']:
-                    raise ValueError("بيانات المشاعر غير متوفرة")
-                
+                response.raise_for_status(); data = response.json()
+                if 'data' not in data or not data['data']: raise ValueError("بيانات المشاعر غير متوفرة")
                 latest_data = data['data'][0]
-                return {
-                    "feargreed_value": int(latest_data['value']), 
-                    "feargreed_class": latest_data['value_classification'],
-                    "source": "alternative.me",
-                    "timestamp": datetime.now().isoformat()
-                }
-        except Exception as e:
-            return None
-
+                return { "feargreed_value": int(latest_data['value']), "feargreed_class": latest_data['value_classification'], "source": "alternative.me", "timestamp": datetime.now().isoformat() }
+        except Exception as e: return None
     def _determine_market_trend(self, bitcoin_price, sentiment_data):
-        if bitcoin_price is None:
-            return "UNKNOWN"
+        if bitcoin_price is None: return "UNKNOWN"
         if bitcoin_price > 60000: score = 1
         elif bitcoin_price < 55000: score = -1
         else: score = 0
@@ -152,36 +139,27 @@ class DataManager:
         if score >= 1: return "bull_market"
         elif score <= -1: return "bear_market"
         else: return "sideways_market"
-
     def _get_btc_sentiment(self, bitcoin_price):
         if bitcoin_price is None: return 'UNKNOWN'
         elif bitcoin_price > 60000: return 'BULLISH'
         elif bitcoin_price < 55000: return 'BEARISH'
         else: return 'NEUTRAL'
-
     async def _get_prices_with_fallback(self):
         try:
             prices = await self._get_prices_from_kucoin_safe()
-            if prices.get('bitcoin') and prices.get('ethereum'):
-                return prices
+            if prices.get('bitcoin') and prices.get('ethereum'): return prices
             return await self._get_prices_from_coingecko()
-        except Exception as e:
-            return {'bitcoin': None, 'ethereum': None}
-
+        except Exception as e: return {'bitcoin': None, 'ethereum': None}
     async def _get_prices_from_kucoin_safe(self):
         if not self.exchange: return {'bitcoin': None, 'ethereum': None}
         try:
             prices = {'bitcoin': None, 'ethereum': None}
-            btc_ticker = self.exchange.fetch_ticker('BTC/USDT')
-            btc_price = float(btc_ticker.get('last', 0)) if btc_ticker.get('last') else None
+            btc_ticker = 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_price = float(eth_ticker.get('last', 0)) if eth_ticker.get('last') else None
+            eth_ticker = 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
             return prices
-        except Exception as e:
-            return {'bitcoin': None, 'ethereum': None}
-
+        except Exception as e: return {'bitcoin': None, 'ethereum': None}
     async def _get_prices_from_coingecko(self):
         try:
             await asyncio.sleep(0.5)
@@ -189,32 +167,17 @@ class DataManager:
             headers = {'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36', 'Accept': 'application/json'}
             async with httpx.AsyncClient(headers=headers) as client:
                 response = await client.get(url, timeout=10)
-                if response.status_code == 429:
-                    await asyncio.sleep(2)
-                    response = await client.get(url, timeout=10)
-                response.raise_for_status()
-                data = response.json()
-                btc_price = data.get('bitcoin', {}).get('usd')
-                eth_price = data.get('ethereum', {}).get('usd')
-                if btc_price and eth_price:
-                    return {'bitcoin': btc_price, 'ethereum': eth_price}
-                else:
-                    return {'bitcoin': None, 'ethereum': None}
-        except Exception as e:
-            return {'bitcoin': None, 'ethereum': None}
-
+                if response.status_code == 429: await asyncio.sleep(2); response = await client.get(url, timeout=10)
+                response.raise_for_status(); data = response.json()
+                btc_price = data.get('bitcoin', {}).get('usd'); eth_price = data.get('ethereum', {}).get('usd')
+                if btc_price and eth_price: return {'bitcoin': btc_price, 'ethereum': eth_price}
+                else: return {'bitcoin': None, 'ethereum': None}
+        except Exception as e: return {'bitcoin': None, 'ethereum': None}
     def _get_minimal_market_context(self):
-        return {
-            'timestamp': datetime.now().isoformat(),
-            'data_available': False,
-            'market_trend': 'UNKNOWN',
-            'btc_sentiment': 'UNKNOWN',
-            'data_quality': 'LOW'
-        }
+        return { 'timestamp': datetime.now().isoformat(), 'data_available': False, 'market_trend': 'UNKNOWN', 'btc_sentiment': 'UNKNOWN', 'data_quality': 'LOW' }
 
-    
     def _create_dataframe(self, candles: List) -> pd.DataFrame:
-        """(V7.1) دالة مساعدة لإنشاء DataFrame لتحليل 1H"""
+        """(V9.1) إنشاء DataFrame (تحتاج 200 شمعة للميزات)"""
         try:
             if not candles: return pd.DataFrame()
             df = pd.DataFrame(candles, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
@@ -224,583 +187,211 @@ class DataManager:
             df.sort_index(inplace=True)
             return df
         except Exception as e:
-            print(f"❌ خطأ في إنشاء DataFrame لمرشح 1H: {e}")
+            print(f"❌ خطأ في إنشاء DataFrame: {e}")
             return pd.DataFrame()
 
-    def _get_mc_score_for_filter(self, analysis: Dict) -> float:
-        """(V7.4) (دالة مساعدة) لحساب درجة مونت كارلو للفلتر"""
-        mc_distribution = analysis.get('monte_carlo_distribution')
-        monte_carlo_score = 0
-        
-        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)
-            
-            if current_price > 0:
-                normalized_var = var_95_value / current_price
-                risk_penalty = 1.0
-                if normalized_var > 0.05: risk_penalty = 0.5           
-                elif normalized_var > 0.03: risk_penalty = 0.8
-                
-                normalized_prob_score = max(0.0, (prob_gain - 0.5) * 2)
-                monte_carlo_score = normalized_prob_score * risk_penalty
-        return monte_carlo_score
-
-    def _calculate_1h_momentum_burst_score(self, analysis: Dict) -> float:
-        """
-        (محدث V7.7) - الفلتر الأول: "فلتر شمس منتصف الظهر" (الانفجار)
-        """
-        try:
-            ohlcv_candles = analysis.get('ohlcv_1h', {}).get('1h', [])
-            if not ohlcv_candles or len(ohlcv_candles) < 30: 
-                return 0.0 
-
-            closes_1h = [c[4] for c in ohlcv_candles]
-            if len(closes_1h) > 20:
-                std_dev = np.std(closes_1h[-20:])
-                if std_dev < 1e-5: 
-                    return 0.0 # (واقي العملات المستقرة)
-            
-            if ta is None: return 0.0 
-            df = self._create_dataframe(ohlcv_candles)
-            if df.empty: return 0.0
-
-            # 1. حساب مؤشرات الانفجار
-            volume = df['volume']
-            vol_ma = ta.sma(volume, length=20)
-            if vol_ma is None or vol_ma.empty: return 0.0
-            current_volume = volume.iloc[-1]
-            avg_volume = vol_ma.iloc[-1]
-            adx_data = ta.adx(df['high'], df['low'], df['close'], length=14)
-            current_adx = adx_data['ADX_14'].iloc[-1] if adx_data is not None and not adx_data.empty else 0
-            
-            # 2. جلب المؤشرات الأساسية (المحسوبة مسبقاً)
-            indicators = analysis.get('advanced_indicators', {}).get('1h', {})
-            rsi = indicators.get('rsi', 50)
-            
-            # 3. جلب درجة مونت كارلو (المحسوبة مسبقاً)
-            monte_carlo_score = self._get_mc_score_for_filter(analysis)
-            
-            # 4. جلب درجة الأنماط (المحسوبة مسبقاً)
-            pattern_confidence = analysis.get('pattern_analysis', {}).get('pattern_confidence', 0)
-
-            # --- (منطق فلترة الانفجار) ---
-            VOL_MULTIPLIER = 1.75  # (صارم)
-            ADX_THRESHOLD = 25.0   # (صارم)
-            RSI_MIN = 60           # (صارم)
-            RSI_MAX = 85           
-
-            vol_score = 0.0
-            if avg_volume > 0:
-                vol_score = min(1.0, max(0.0, (current_volume / avg_volume) / VOL_MULTIPLIER))
-            
-            adx_score = min(1.0, max(0.0, (current_adx - ADX_THRESHOLD) / 15.0)) 
-            
-            rsi_score = 0.0
-            if RSI_MIN <= rsi <= RSI_MAX:
-                rsi_score = 1.0
-            elif rsi > RSI_MAX:
-                rsi_score = 0.5 
-
-            # (الأوزان لفلتر الانفجار)
-            WEIGHT_VOL = 0.30
-            WEIGHT_ADX = 0.30
-            WEIGHT_RSI = 0.15
-            WEIGHT_MC = 0.15
-            WEIGHT_PATTERN = 0.10
-
-            final_score = (
-                (vol_score * WEIGHT_VOL) +
-                (adx_score * WEIGHT_ADX) +
-                (rsi_score * WEIGHT_RSI) +
-                (monte_carlo_score * WEIGHT_MC) +
-                (pattern_confidence * WEIGHT_PATTERN)
-            )
-            
-            return min(max(final_score, 0.0), 1.0)
-            
-        except Exception as e:
-            return 0.0
-
-    # 🔴 --- START OF CHANGE (V7.9) --- 🔴
-    def _calculate_1h_reversal_spark_score(self, analysis: Dict) -> float:
-        """
-        (محدث V7.9) - الفلتر الثاني: "فلتر شرارة الانعكاس" (أوزان معدلة)
-        """
-        try:
-            # (1. واقي العملات المستقرة)
-            ohlcv_candles = analysis.get('ohlcv_1h', {}).get('1h', [])
-            if not ohlcv_candles or len(ohlcv_candles) < 30: 
-                return 0.0 
-
-            closes_1h = [c[4] for c in ohlcv_candles]
-            if len(closes_1h) > 20:
-                std_dev = np.std(closes_1h[-20:])
-                if std_dev < 1e-5: 
-                    return 0.0 
-            
-            if ta is None: return 0.0 
-            df = self._create_dataframe(ohlcv_candles)
-            if df.empty: return 0.0
-
-            # 2. حساب مؤشرات "الشرارة"
-            volume = df['volume']
-            vol_ma = ta.sma(volume, length=20)
-            if vol_ma is None or vol_ma.empty: return 0.0
-            current_volume = volume.iloc[-1]
-            avg_volume = vol_ma.iloc[-1]
-            adx_data = ta.adx(df['high'], df['low'], df['close'], length=14)
-            current_adx = adx_data['ADX_14'].iloc[-1] if adx_data is not None and not adx_data.empty else 0
-            
-            # 3. جلب المؤشرات الأساسية (المحسوبة مسبقاً)
-            indicators = analysis.get('advanced_indicators', {}).get('1h', {})
-            rsi = indicators.get('rsi', 50)
-            
-            # 4. جلب درجة مونت كارلو (المحسوبة مسبقاً)
-            monte_carlo_score = self._get_mc_score_for_filter(analysis)
-            
-            # 5. جلب درجة الأنماط (المحسوبة مسبقاً)
-            pattern_analysis = analysis.get('pattern_analysis', {})
-            
-            # --- (منطق فلترة "الشرارة") ---
-            
-            pattern_score = 0.0
-            if pattern_analysis.get('predicted_direction') == 'up':
-                pattern_score = pattern_analysis.get('pattern_confidence', 0)
-
-            # (RSI في منطقة الاشتعال 45-60)
-            RSI_IGNITION_MIN = 45
-            RSI_IGNITION_MAX = 60
-            rsi_score = 0.0
-            if RSI_IGNITION_MIN <= rsi <= RSI_IGNITION_MAX:
-                rsi_score = 1.0 # (مثالي)
-            elif rsi > RSI_IGNITION_MAX:
-                rsi_score = 0.5 # (لا بأس)
-
-            # (ADX منخفض، يدل على "تجميع" أو "نطاق")
-            ADX_COIL_THRESHOLD = 20.0
-            adx_score = 0.0
-            if current_adx < ADX_COIL_THRESHOLD:
-                adx_score = 1.0 # (مثالي)
-            
-            # (الحجم بدأ في الارتفاع (أعلى من المتوسط))
-            vol_score = 0.0
-            if avg_volume > 0 and (current_volume / avg_volume) > 1.1:
-                vol_score = 0.5 # (مجرد اهتمام بسيط)
-
-            # (الأوزان الجديدة V7.9 - بناءً على طلبك)
-            WEIGHT_PATTERN = 0.10     # (تم الخفض)
-            WEIGHT_RSI_SPARK = 0.35   # (تم الرفع)
-            WEIGHT_ADX_COIL = 0.35    # (تم الرفع)
-            WEIGHT_MC = 0.10
-            WEIGHT_VOL = 0.10
-            # (المجموع = 1.0)
-
-            final_score = (
-                (pattern_score * WEIGHT_PATTERN) +
-                (rsi_score * WEIGHT_RSI_SPARK) +
-                (adx_score * WEIGHT_ADX_COIL) +
-                (monte_carlo_score * WEIGHT_MC) +
-                (vol_score * WEIGHT_VOL)
-            )
-            
-            return min(max(final_score, 0.0), 1.0)
-            
-        except Exception as e:
-            return 0.0
-    # 🔴 --- END OF CHANGE (V7.9) --- 🔴
-
+    # 🔴 --- (تم حذف جميع الفلاتر الغبية V7.x نهائياً) --- 🔴
 
+    # 🔴 --- START OF CHANGE (V9.3 - High Confidence ML Ranker) --- 🔴
     async def layer1_rapid_screening(self) -> List[Dict[str, Any]]:
         """
-        الطبقة 1: فحص سريع - (محدث بالكامل V7.9)
+        الطبقة 1: فحص سريع - (محدث V9.3)
+        - يستخدم "العقل الذكي" (V9.2 Model).
+        - يطبق عتبة ثقة صارمة (70%) لضمان الجودة.
         """
-        print("📊 الطبقة 1 (V7.9): بدء الغربلة (الكاشف المزدوج + أوزان الشرارة المعدلة)...")
+        print("📊 الطبقة 1 (V9.3 - High Confidence Ranker): بدء الغربلة...")
         
+        if not self.layer1_ranker:
+            print("❌ [V9.3] الرانكر غير مهيأ. إيقاف الغربلة.")
+            return []
+
         # الخطوة 1: جلب أفضل 100 عملة حسب الحجم
         volume_data = await self._get_volume_data_optimal()
+        if not volume_data: volume_data = await self._get_volume_data_direct_api()
         if not volume_data:
-            volume_data = await self._get_volume_data_direct_api()
-        
-        if not volume_data:
-            print("❌ فشل جلب بيانات الأحجام للطبقة 1")
+            print("❌ [V9.3] فشل جلب بيانات الأحجام.")
             return []
         
         volume_data.sort(key=lambda x: x['dollar_volume'], reverse=True)
         top_100_by_volume = volume_data[:100]
+        print(f"✅ [V9.3] تم تحديد أفضل {len(top_100_by_volume)} عملة. بدء حساب الميزات الذكية...")
         
-        print(f"✅ تم تحديد أفضل {len(top_100_by_volume)} عملة. بدء تشغيل الكاشف المزدوج (1H)...")
-        
-        final_candidates = []
-        
-        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]
-            batch_symbols = [s['symbol'] for s in batch_symbols_data]
-            
-            print(f"   🔄 معالجة دفعة {int(i/batch_size) + 1}/{(len(top_100_by_volume) + batch_size - 1) // batch_size} ({len(batch_symbols)} عملة)...")
-            
-            # الخطوة 2: جلب بيانات 1H بالتوازي
-            tasks = [self._fetch_1h_ohlcv_for_screening(symbol) for symbol in batch_symbols]
-            results_candles = await asyncio.gather(*tasks, return_exceptions=True)
+        final_candidates_with_scores = []
+        batch_symbols_data = top_100_by_volume
+        batch_symbols = [s['symbol'] for s in batch_symbols_data]
             
-            analysis_tasks = []
-            valid_symbol_data = []
-
-            for j, (candles) in enumerate(results_candles):
-                symbol_data = batch_symbols_data[j]
-                symbol = symbol_data['symbol']
-
-                if isinstance(candles, Exception) or not candles or len(candles) < 50:
-                    continue
-                
-                ohlcv_1h_only = {'1h': candles}
-                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
-                
-            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 = symbol_data['symbol']
-
-                if isinstance(analysis_output, Exception):
-                    print(f"      - {symbol}: فشل الكاشف المصغر ({analysis_output})")
-                    continue
-                
-                analysis_output['ohlcv_1h'] = symbol_data['ohlcv_1h']
-                analysis_output['symbol'] = symbol
-                
-                # (1. محاولة فلتر "الانفجار" أولاً)
-                burst_score = self._calculate_1h_momentum_burst_score(analysis_output)
-                
-                final_score = burst_score
-                candidacy_reason = "1H_MOMENTUM_BURST" # (السبب الافتراضي)
-
-                # (2. إذا فشل فلتر الانفجار، جرب فلتر "الشرارة")
-                if burst_score < 0.50:
-                    # (استخدام V7.9 بالأوزان الجديدة)
-                    spark_score = self._calculate_1h_reversal_spark_score(analysis_output)
-                    
-                    final_score = spark_score # (استخدام درجة الشرارة)
-                    candidacy_reason = "1H_REVERSAL_SPARK" # (تغيير السبب)
-
-                # (3. التحقق من العتبة النهائية 0.50)
-                if final_score >= 0.50:
-                    print(f"      ✅ {symbol}: نجح ({candidacy_reason} | الدرجة: {final_score:.2f})")
-                    symbol_data['layer1_score'] = final_score
-                    symbol_data['reasons_for_candidacy'] = [candidacy_reason]
-                    
-                    if 'ohlcv_1h' in symbol_data: del symbol_data['ohlcv_1h'] 
-                    
-                    final_candidates.append(symbol_data)
-
-        print(f"🎯 اكتملت الغربلة (V7.9). تم تأهيل {len(final_candidates)} عملة من أصل 100 للطبقة 2.")
-        
-        if final_candidates:
-            print("🏆 المرشحون الناجحون:")
-            for k, candidate in enumerate(final_candidates[:15]):
-                score = candidate.get('layer1_score', 0)
-                reason = candidate.get('reasons_for_candidacy', ['N/A'])[0]
-                print(f"   {k+1:2d}. {candidate['symbol']}: (الدرجة: {score:.2f}) | السبب: {reason}")
+        # الخطوة 2: جلب بيانات 1H (200 شمعة)
+        tasks = [self._fetch_1h_ohlcv_for_screening(symbol, limit=200) for symbol in batch_symbols]
+        results_candles = await asyncio.gather(*tasks, return_exceptions=True)
         
-        return final_candidates
-
-    async def _run_mini_detector(self, symbol_data: Dict) -> Dict:
-        """(محدث V7.8) يشغل المحللات + يطبع نتيجة الأنماط V8"""
-        ohlcv_1h = symbol_data.get('ohlcv_1h')
-        current_price = symbol_data.get('current_price')
-        symbol = symbol_data.get('symbol', 'UNKNOWN') # (للطباعة)
-        
-        df = self._create_dataframe(ohlcv_1h.get('1h'))
-        if df.empty:
-            raise ValueError("DataFrame فارغ لتحليل 1H")
-            
-        analysis_dict = {'current_price': current_price}
-
-        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)
+        valid_symbol_data_for_ranking = []
+        for j, (candles) in enumerate(results_candles):
+            symbol_data = batch_symbols_data[j]
+            if isinstance(candles, Exception) or not candles or len(candles) < 200: continue
+            symbol_data['ohlcv_1h_raw'] = candles 
+            valid_symbol_data_for_ranking.append(symbol_data)
+
+        if not valid_symbol_data_for_ranking:
+            print("❌ [V9.3] لا توجد عملات صالحة (تحتاج 200 شمعة 1H).")
+            return []
 
-        results = await asyncio.gather(task_mc, task_pattern, return_exceptions=True)
+        print(f"   🔄 [V9.3] حساب الميزات الذكية لـ {len(valid_symbol_data_for_ranking)} عملة...")
         
-        analysis_dict['advanced_indicators'] = {'1h': task_indicators}
+        # الخطوة 3: حساب "الميزات الذكية V9.1"
+        all_features_list = []
+        symbols_in_order = []
+        for symbol_data in valid_symbol_data_for_ranking:
+            try:
+                df = self._create_dataframe(symbol_data['ohlcv_1h_raw'])
+                if df.empty: continue
+                smart_features = self.technical_analyzer.calculate_v9_smart_features(df)
+                if smart_features:
+                    all_features_list.append(smart_features)
+                    symbols_in_order.append(symbol_data)
+            except Exception: pass
         
-        if not isinstance(results[0], Exception):
-            analysis_dict['monte_carlo_distribution'] = results[0]
+        if not all_features_list:
+            print("❌ [V9.3] فشل حساب الميزات الذكية.")
+            return []
+
+        # الخطوة 4: التنبؤ (التصنيف)
+        print(f"   🧠 [V9.3] إرسال {len(all_features_list)} عملة إلى نموذج الرانكر...")
+        features_dataframe = pd.DataFrame(all_features_list)
+        probabilities = self.layer1_ranker.predict_proba(features_dataframe)
         
-        # (إضافة طباعة تشخيصية لنموذج الأنماط V8)
-        if not isinstance(results[1], Exception):
-            pattern_result = results[1]
-            analysis_dict['pattern_analysis'] = pattern_result
+        # الخطوة 5: تجميع النتائج (مع عتبة الثقة العالية 70%)
+        for i, (symbol_data) in enumerate(symbols_in_order):
+            score = probabilities[i]
             
-            pattern_name = pattern_result.get('pattern_detected', 'no_clear_pattern')
-            confidence = pattern_result.get('pattern_confidence', 0)
-            tf = pattern_result.get('timeframe', 'N/A')
+            # 🔴 (V9.3: عتبة الثقة الصارمة - طلب المستخدم) 🔴
+            if score >= 0.70: 
+                symbol = symbol_data['symbol']
+                print(f"      ✅ {symbol}: نجح بامتياز (الاحتمالية: {score:.3f})")
+                symbol_data['layer1_score'] = float(score)
+                symbol_data['reasons_for_candidacy'] = ["V9_SMART_RANKER_HIGH_CONFIDENCE"]
+                if 'ohlcv_1h_raw' in symbol_data: del symbol_data['ohlcv_1h_raw'] 
+                final_candidates_with_scores.append(symbol_data)
 
-            if pattern_name != "Neutral / No Pattern" and confidence > 0.5:
-                print(f"      💡 [ML Pattern] {symbol}: نمط مُكتشف! "
-                      f"'{pattern_name}' (الثقة: {confidence:.2f}) على إطار {tf}")
-            elif pattern_name == "Neutral / No Pattern":
-                pass 
-            else:
-                print(f"      - [ML Pattern] {symbol}: نمط ضعيف. "
-                      f"'{pattern_name}' (الثقة: {confidence:.2f}) على إطار {tf}")
+        print(f"🎯 اكتملت الغربلة (V9.3). تم تأهيل {len(final_candidates_with_scores)} عملة (ثقة >= 70%).")
         
+        if final_candidates_with_scores:
+            final_candidates_with_scores.sort(key=lambda x: x['layer1_score'], reverse=True)
+            print("🏆 المرشحون الناجحون (Top Candidates):")
+            for k, candidate in enumerate(final_candidates_with_scores[:5]):
+                print(f"   {k+1}. {candidate['symbol']}: (Score: {candidate.get('layer1_score'):.3f})")
         else:
-            print(f"      ❌ [ML Pattern] {symbol}: فشل تحليل الأنماط ({results[1]})")
-            analysis_dict['pattern_analysis'] = {} 
-            
-        return analysis_dict
+            print("⚠️ [V9.3] لم تنجح أي عملة في تجاوز عتبة الثقة 70%. (هذا جيد، النظام حذر)")
 
+        return final_candidates_with_scores[:20]
+    # 🔴 --- END OF CHANGE (V9.3) --- 🔴
 
-    async def _fetch_1h_ohlcv_for_screening(self, symbol: str) -> List:
-        """(V7.1) جلب 100 شمعة لإطار الساعة (1H) للغربلة السريعة"""
+    async def _fetch_1h_ohlcv_for_screening(self, symbol: str, limit: int = 200) -> List:
         try:
-            ohlcv_data = self.exchange.fetch_ohlcv(symbol, '1h', limit=100)
-            
-            if not ohlcv_data or len(ohlcv_data) < 50:
-                return None
+            ohlcv_data = self.exchange.fetch_ohlcv(symbol, '1h', limit=limit)
+            if not ohlcv_data or len(ohlcv_data) < limit: return None
             return ohlcv_data
-        except Exception:
-            return None
+        except Exception: return None
 
+    # ... (باقي الدوال _get_volume_data_optimal, stream_ohlcv_data... إلخ - لا تغيير) ...
     async def _get_volume_data_optimal(self) -> List[Dict[str, Any]]:
         try:
             if not self.exchange: return []
-            tickers = self.exchange.fetch_tickers()
-            volume_data = []
+            tickers = self.exchange.fetch_tickers(); volume_data = []
             for symbol, ticker in tickers.items():
                 if not symbol.endswith('/USDT') or not ticker.get('active', True): continue
-                current_price = ticker.get('last', 0)
-                quote_volume = ticker.get('quoteVolume', 0)
+                current_price = ticker.get('last', 0); quote_volume = ticker.get('quoteVolume', 0)
                 if current_price is None or current_price <= 0: continue
-                if quote_volume is not None and quote_volume > 0:
-                    dollar_volume = quote_volume
+                if quote_volume is not None and quote_volume > 0: dollar_volume = quote_volume
                 else:
                     base_volume = ticker.get('baseVolume', 0)
                     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
-                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,
-                    'price_change_24h': price_change_24h
-                })
-            print(f"✅ تم معالجة {len(volume_data)} عملة في الطريقة المثلى (لجلب الحجم)")
+                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, 'price_change_24h': price_change_24h })
+            print(f"✅ تم معالجة {len(volume_data)} عملة في الطريقة المثلى")
             return volume_data
-        except Exception as e:
-            print(f"❌ خطأ في جلب بيانات الحجم المثلى: {e}")
-            return []
-
+        except Exception as e: print(f"❌ خطأ في جلب بيانات الحجم المثلى: {e}"); return []
     async def _get_volume_data_direct_api(self) -> List[Dict[str, Any]]:
         try:
             url = "https://api.kucoin.com/api/v1/market/allTickers"
             async with httpx.AsyncClient(timeout=15) as client:
-                response = await client.get(url)
-                response.raise_for_status()
-                data = response.json()
+                response = await client.get(url); response.raise_for_status(); data = response.json()
                 if data.get('code') != '200000': raise ValueError(f"استجابة API غير متوقعة: {data.get('code')}")
-                tickers = data['data']['ticker']
-                volume_data = []
+                tickers = data['data']['ticker']; volume_data = []
                 for ticker in tickers:
                     symbol = ticker['symbol']
                     if not symbol.endswith('USDT'): continue
                     formatted_symbol = symbol.replace('-', '/')
                     try:
-                        vol_value = ticker.get('volValue')
-                        last_price = ticker.get('last')
-                        change_rate = ticker.get('changeRate')
-                        vol = ticker.get('vol')
+                        vol_value = ticker.get('volValue'); last_price = ticker.get('last'); change_rate = ticker.get('changeRate'); vol = ticker.get('vol')
                         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
-                        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({
-                                'symbol': formatted_symbol, 'dollar_volume': dollar_volume,
-                                'current_price': current_price, 'volume_24h': volume_24h,
-                                'price_change_24h': price_change
-                            })
-                    except (ValueError, TypeError, KeyError) as e: continue
-                print(f"✅ تم معالجة {len(volume_data)} عملة في الطريقة المباشرة (لجلب الحجم)")
+                        dollar_volume = float(vol_value) if vol_value else 0; current_price = float(last_price) if last_price else 0; 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({ 'symbol': formatted_symbol, 'dollar_volume': dollar_volume, 'current_price': current_price, 'volume_24h': volume_24h, 'price_change_24h': price_change })
+                    except (ValueError, TypeError, KeyError): continue
+                print(f"✅ تم معالجة {len(volume_data)} عملة في الطريقة المباشرة")
                 return volume_data
-        except Exception as e:
-            print(f"❌ خطأ في جلب بيانات الحجم المباشر: {e}")
-            return []
-
+        except Exception as e: print(f"❌ خطأ ف�� جلب بيانات الحجم المباشر: {e}"); return []
     async def stream_ohlcv_data(self, symbols: List[Dict[str, Any]], queue: asyncio.Queue):
-        """
-        (محدث V7.2)
-        جلب بيانات OHLCV كاملة (6 أطر زمنية) للعملات الناجحة فقط
-        """
-        print(f"📊 بدء تدفق بيانات OHLCV (الكاملة) لـ {len(symbols)} عملة (الناجحين من الغربلة)...")
-        
-        batch_size = 15
-        batches = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)]
-        
-        total_successful = 0
-        
+        print(f"📊 بدء تدفق بيانات OHLCV (الكاملة) لـ {len(symbols)} عملة (مصنفة)...")
+        batch_size = 15; batches = [symbols[i:i + batch_size] for i in range(0, len(symbols), batch_size)]; total_successful = 0
         for batch_num, batch in enumerate(batches):
             print(f"   🔄 [المنتج] جلب الدفعة {batch_num + 1}/{len(batches)} ({len(batch)} عملة)...")
-            
-            batch_tasks = []
-            
-            # (V7.2 FIX)
-            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_tasks = []; for symbol_data in batch: task = asyncio.create_task(self._fetch_complete_ohlcv_parallel(symbol_data['symbol'])); 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):
-                
-                original_symbol_data = batch[i]
-                symbol_str = original_symbol_data['symbol']
-
-                if isinstance(result, Exception):
-                    print(f"      ❌ [المنتج] فشل جلب {symbol_str}: {result}")
+                if isinstance(result, Exception): print(f"      ❌ [المنتج] فشل جلب {batch[i]['symbol']}: {result}")
                 elif result is not None:
-                    result.update(original_symbol_data) 
-                    successful_data_for_batch.append(result)
-                    successful_count += 1
-                    timeframes_count = result.get('successful_timeframes', 0)
-                    print(f"      ✅ [المنتج] {symbol_str}: {timeframes_count}/6 أطر زمنية")
-                else:
-                    print(f"      ⚠️ [المنتج] {symbol_str}: بيانات غير كافية، تم التجاهل")
-
-            print(f"      📦 [المنتج] اكتملت الدفعة {batch_num + 1}: {successful_count}/{len(batch)} ناجحة")
-            
+                    result.update(batch[i]); successful_data_for_batch.append(result)
+                    print(f"      ✅ [المنتج] {batch[i]['symbol']}: {result.get('successful_timeframes', 0)}/6 أطر زمنية")
             if successful_data_for_batch:
-                try:
-                    await queue.put(successful_data_for_batch)
-                    print(f"      📬 [المنتج] تم إرسال {len(successful_data_for_batch)} عملة إلى طابور المعالجة")
-                    total_successful += len(successful_data_for_batch)
-                except Exception as q_err:
-                    print(f"      ❌ [المنتج] فشل إرسال الدفعة للطابور: {q_err}")
-
-            if batch_num < len(batches) - 1:
-                await asyncio.sleep(1)
-        
-        print(f"✅ [المنتج] اكتمل تدفق بيانات OHLCV (الكاملة). تم إرسال {total_successful} عملة للمعالجة.")
-        
-        try:
-            await queue.put(None)
-            print("      📬 [المنتج] تم إرسال إشارة الإنهاء (None) إلى الطابور.")
-        except Exception as q_err:
-            print(f"      ❌ [المنتج] فشل إرسال إشارة الإنهاء (None) للطابور.")
-
+                try: await queue.put(successful_data_for_batch); total_successful += len(successful_data_for_batch)
+                except Exception as q_err: print(f"      ❌ [المنتج] فشل إرسال الدفعة للطابور: {q_err}")
+            if batch_num < len(batches) - 1: await asyncio.sleep(1)
+        print(f"✅ [المنتج] اكتمل التدفق. تم إرسال {total_successful} عملة."); await queue.put(None)
 
     async def _fetch_complete_ohlcv_parallel(self, symbol: str) -> Dict[str, Any]:
-        """(V7.2) جلب بيانات OHLCV كاملة - يتوقع 'symbol' كنص"""
         try:
-            ohlcv_data = {}
-            
-            timeframes = [
-                ('5m', 200), ('15m', 200), ('1h', 200),
-                ('4h', 200), ('1d', 200), ('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
-            
+            ohlcv_data = {}; timeframes = [('5m', 200), ('15m', 200), ('1h', 200), ('4h', 200), ('1d', 200), ('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
             for i, (timeframe, limit) in enumerate(timeframes):
                 result = timeframe_results[i]
                 if isinstance(result, Exception): continue
-                # (V8-MODIFICATION) زيادة الحد الأدنى من الشموع لاستيعاب المؤشرات
-                if result and len(result) >= 200: # (كان 10)
-                    ohlcv_data[timeframe] = result
-                    successful_timeframes += 1
-            
-            # (V8-MODIFICATION) زيادة الحد الأدنى للأطر الزمنية
-            if successful_timeframes >= 3 and ohlcv_data: # (كان 2)
+                if result and len(result) >= 200: ohlcv_data[timeframe] = result; successful_timeframes += 1
+            if successful_timeframes >= 3 and ohlcv_data:
                 try:
                     current_price = await self.get_latest_price_async(symbol)
                     if current_price is None:
-                        for timeframe_data in ohlcv_data.values():
-                            if timeframe_data and len(timeframe_data) > 0:
-                                last_candle = timeframe_data[-1]
-                                if len(last_candle) >= 5:
-                                    current_price = last_candle[4]; break
+                        for tf_data in ohlcv_data.values():
+                            if tf_data and len(tf_data) > 0: current_price = tf_data[-1][4]; break
                     if current_price is None: return None
-                    
-                    result_data = {
-                        'symbol': symbol, '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()},
-                        'successful_timeframes': successful_timeframes
-                    }
-                    return result_data
-                except Exception as price_error: return None
+                    return { 'symbol': symbol, '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()}, 'successful_timeframes': successful_timeframes }
+                except Exception: return None
             else: return None
-        except Exception as e: return None
-
+        except Exception: return None
     async def _fetch_single_timeframe_improved(self, symbol: str, timeframe: str, limit: int):
-        """(V7.2) جلب بيانات إطار زمني واحد - يتوقع 'symbol' كنص"""
-        max_retries = 3
-        retry_delay = 2
+        max_retries = 3; retry_delay = 2
         for attempt in range(max_retries):
             try:
                 ohlcv_data = self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
-                if ohlcv_data and len(ohlcv_data) > 0:
-                    return ohlcv_data
-                else:
-                    return []
-            except Exception as e:
-                if attempt < max_retries - 1:
-                    await asyncio.sleep(retry_delay * (attempt + 1))
-                else:
-                    return []
-
+                if ohlcv_data and len(ohlcv_data) > 0: return ohlcv_data
+                else: return []
+            except Exception:
+                if attempt < max_retries - 1: await asyncio.sleep(retry_delay * (attempt + 1))
+                else: 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
             ticker = self.exchange.fetch_ticker(symbol)
-            if not ticker: return None
-            current_price = ticker.get('last')
-            if current_price is None: return None
-            return float(current_price)
-        except Exception as e: return None
-
+            return float(ticker['last']) if ticker and ticker.get('last') else None
+        except Exception: return None
     async def get_whale_data_for_symbol(self, symbol):
-        try:
-            if self.whale_monitor:
-                whale_data = await self.whale_monitor.get_symbol_whale_activity(symbol)
-                return whale_data
-            else: return None
-        except Exception as e: return None
-
+        try: return await self.whale_monitor.get_symbol_whale_activity(symbol) if self.whale_monitor else None
+        except Exception: return None
     async def get_whale_trading_signal(self, symbol, whale_data, market_context):
-        try:
-            if self.whale_monitor:
-                return await self.whale_monitor.generate_whale_trading_signal(symbol, whale_data, market_context)
-            else:
-                return {'action': 'HOLD', 'confidence': 0.3, 'reason': 'Whale monitor not available', 'source': 'whale_analysis'}
-        except Exception as e:
-            return {'action': 'HOLD', 'confidence': 0.3, 'reason': f'Error: {str(e)}', 'source': 'whale_analysis'}
+        try: return await self.whale_monitor.generate_whale_trading_signal(symbol, whale_data, market_context) if self.whale_monitor else {'action': 'HOLD', 'confidence': 0.3, 'reason': 'Whale monitor not available', 'source': 'whale_analysis'}
+        except Exception as e: return {'action': 'HOLD', 'confidence': 0.3, 'reason': f'Error: {str(e)}', 'source': 'whale_analysis'}
 
-print("✅ DataManager loaded - V7.9 (Re-balanced Spark Weights)")
+print("✅ DataManager loaded - V9.3 (High Confidence Ranker)")