Update app.py

app.py

@@ -17,26 +17,25 @@ import json
 import state
 import re
 
-# إعدادات النظام
+from whale_news_data import whale_monitor_global
+from helpers import safe_float_conversion, _apply_patience_logic
+
 TOP_N_SYMBOLS = 100
 OPPORTUNITY_COUNT = 10
 CHUNK_SIZE = 5
 
-# المتغيرات العامة للنظام
 r2_service_global = None
 data_manager_global = None
 llm_service_global = None
 learning_engine_global = None
 realtime_monitor = None
 
-# Real-time trade monitoring with enhanced risk management
 class RealTimeTradeMonitor:
     def __init__(self):
         self.monitoring_tasks = {}
         self.is_running = False
         
     async def start_monitoring(self):
-        """بدء مراقبة جميع الصفقات المفتوحة"""
         self.is_running = True
         print("🔍 Starting real-time trade monitoring...")
         
@@ -62,7 +61,6 @@ class RealTimeTradeMonitor:
                 await asyncio.sleep(30)
     
     async def _monitor_single_trade(self, trade):
-        """مراقبة صفقة فردية في الوقت الحقيقي"""
         symbol = trade['symbol']
         strategy = trade.get('strategy', 'GENERIC')
         print(f"📊 Starting real-time monitoring for {symbol} (Strategy: {strategy})")
@@ -121,13 +119,11 @@ class RealTimeTradeMonitor:
                 await asyncio.sleep(30)
     
     def stop_monitoring(self):
-        """إيقاف جميع مهام المراقبة"""
         self.is_running = False
         self.monitoring_tasks.clear()
         print("🛑 Real-time trade monitoring stopped")
 
 async def monitor_market_async():
-    """Background task to continuously monitor market health"""
     global data_manager_global
     
     init_attempts = 0
@@ -161,7 +157,6 @@ async def monitor_market_async():
             is_critical = whale_analysis.get('critical_alert', False)
             total_volume = whale_analysis.get('total_volume_usd', 0)
 
-            # 🆕 استخدام البيانات المحسنة من data_manager
             netflow_analysis = whale_analysis.get('netflow_analysis', {})
             net_flow = netflow_analysis.get('net_flow', 0)
             flow_direction = netflow_analysis.get('flow_direction', 'BALANCED')
@@ -177,7 +172,6 @@ async def monitor_market_async():
             should_halt_trading = False
             halt_reason = ""
             
-            # 🆕 تحسين شروط إيقاف التداول بناءً على البيانات المحسنة
             if is_critical:
                 should_halt_trading = True
                 halt_reason = f"CRITICAL whale activity detected: {whale_analysis.get('description')}"
@@ -216,7 +210,6 @@ async def monitor_market_async():
             await asyncio.sleep(60)
 
 async def get_fallback_market_context():
-    """Fallback function when main market context fails"""
     return {
         'timestamp': datetime.now().isoformat(),
         'general_whale_activity': {
@@ -235,22 +228,7 @@ async def get_fallback_market_context():
         'fear_and_greed_index': 50
     }
 
-def safe_float_conversion(value, default=0.0):
-    """تحويل آمن للقيم إلى أرقام"""
-    try:
-        if value is None:
-            return default
-        if isinstance(value, (int, float)):
-            return float(value)
-        if isinstance(value, str):
-            cleaned = ''.join(character for character in value if character.isdigit() or character in '.-')
-            return float(cleaned) if cleaned else default
-        return default
-    except (ValueError, TypeError):
-        return default
-
 async def validate_candidate_data_enhanced(candidate):
-    """✨ تحسين التحقق من جودة المرشحين"""
     try:
         required_fields = ['symbol', 'current_price', 'final_score', 'enhanced_final_score']
         
@@ -281,7 +259,6 @@ async def validate_candidate_data_enhanced(candidate):
         if 'recommended_strategy' not in candidate:
             candidate['recommended_strategy'] = 'unknown'
         
-        # ✅ الإصلاح: التأكد من وجود استراتيجية مستهدفة صالحة
         if 'target_strategy' not in candidate or not candidate['target_strategy'] or candidate['target_strategy'] == 'unknown':
             candidate['target_strategy'] = 'GENERIC'
             
@@ -292,9 +269,7 @@ async def validate_candidate_data_enhanced(candidate):
         return False
 
 async def analyze_market_strategy(market_context):
-    """تحديد الاستراتيجية المثلى بناءً على ظروف السوق المحسنة"""
     try:
-        # 🆕 استخدام البيانات المحسنة من data_manager
         whale_analysis = market_context.get('general_whale_activity', {})
         netflow_analysis = whale_analysis.get('netflow_analysis', {})
         trading_signals = whale_analysis.get('trading_signals', [])
@@ -347,7 +322,6 @@ async def analyze_market_strategy(market_context):
             json_match = re.search(r'\{.*\}', response, re.DOTALL)
             strategy_data = json.loads(json_match.group())
         except:
-            # 🆕 الاستراتيجية الافتراضية المحسنة بناءً على تحليل صافي التدفق
             net_flow = netflow_analysis.get('net_flow', 0)
             if net_flow > 1000000:
                 fallback_strategy = "AGGRESSIVE_GROWTH"
@@ -381,16 +355,13 @@ async def analyze_market_strategy(market_context):
         }
 
 async def find_strategy_specific_candidates(strategy, scan_count):
-    """✨ نظام فلترة ذكي يستخدم الاستراتيجيات المتخصصة - مع دعم البيانات المحسنة"""
     try:
-        # 1. جلب قائمة المرشحين الأولية
         all_candidates = await data_manager_global.find_high_potential_candidates(scan_count * 2)
         
         if not all_candidates:
             print(f"⚠️ الماسح العام لم يجد أي مرشحين أوليين.")
             return []
 
-        # 2. تحديث market_context قبل المعالجة
         market_context = await data_manager_global.get_market_context_async()
         if not market_context:
             print("❌ Failed to get market context for strategy analysis")
@@ -399,21 +370,18 @@ async def find_strategy_specific_candidates(strategy, scan_count):
         feature_processor = FeatureProcessor(market_context, data_manager_global, learning_engine_global)
         
         processed_candidates = []
-        for candidate in all_candidates[:30]:  # ⬇️ تخفيض من 50 إلى 30 للأداء
+        for candidate in all_candidates[:30]:
             try:
-                # تحويل البيانات الخام إلى بيانات معالجة
                 symbol_with_reasons = [{'symbol': candidate['symbol'], 'reasons': candidate.get('reasons', [])}]
                 ohlcv_data = await data_manager_global.get_fast_pass_data_async(symbol_with_reasons)
                 
                 if ohlcv_data and ohlcv_data[0]:
-                    # ✅ تحديث market_context قبل كل معالجة لمنع الخطأ
                     try:
                         updated_market_context = await data_manager_global.get_market_context_async()
                         if updated_market_context:
                             feature_processor.market_context = updated_market_context
                     except Exception as e:
                         print(f"⚠️ Failed to update market context for {candidate['symbol']}: {e}")
-                        # الاستمرار بالسياق القديم إذا فشل التحديث
                     
                     processed = await feature_processor.process_and_score_symbol_enhanced(ohlcv_data[0])
                     if processed:
@@ -425,34 +393,28 @@ async def find_strategy_specific_candidates(strategy, scan_count):
             print("⚠️ لم يتم معالجة أي مرشح بنجاح")
             return []
         
-        # 3. فرز المرشحين حسب الاستراتيجية المطلوبة
         if strategy != 'GENERIC':
-            # تحديد أفضل المرشحين للاستراتيجية المحددة
             strategy_candidates = []
             for candidate in processed_candidates:
-                # ✅ استخدام الدرجات الأساسية بدلاً من المرجحة
                 base_scores = candidate.get('base_strategy_scores', {})
                 strategy_score = base_scores.get(strategy, 0)
                 
-                # ✅ تخفيض كبير في عتبة القبول
-                if strategy_score > 0.2:  # ⬇️ تخفيض من 0.4 إلى 0.2
+                if strategy_score > 0.2:
                     candidate['strategy_match_score'] = strategy_score
                     strategy_candidates.append(candidate)
                     print(f"✅ {candidate['symbol']} مناسب لـ {strategy} (درجة: {strategy_score:.3f})")
             
-            # فرز حسب تطابق الاستراتيجية
             sorted_candidates = sorted(strategy_candidates, 
                                     key=lambda x: x.get('strategy_match_score', 0), 
                                     reverse=True)
-            top_candidates = sorted_candidates[:15]  # ⬇️ تخفيض من 20 إلى 15
+            top_candidates = sorted_candidates[:15]
             
             print(f"✅ تم اختيار {len(top_candidates)} مرشحًا لاستراتيجية {strategy}")
         else:
-            # للاستراتيجية العامة، استخدم النقاط المحسنة
             sorted_candidates = sorted(processed_candidates, 
                                     key=lambda x: x.get('enhanced_final_score', 0), 
                                     reverse=True)
-            top_candidates = sorted_candidates[:15]  # ⬇️ تخفيض من 20 إلى 15
+            top_candidates = sorted_candidates[:15]
             print(f"✅ تم اختيار {len(top_candidates)} مرشحًا للاستراتيجية العامة")
 
         return top_candidates
@@ -463,7 +425,6 @@ async def find_strategy_specific_candidates(strategy, scan_count):
         return []
 
 async def find_new_opportunities_async():
-    """✨ NEW: المسح المحسن باستراتيجية مسبقة مع عتبات مخفضة ودعم البيانات المحسنة"""
     print("🔍 Scanning for new opportunities with enhanced data analysis...")
     try:
         await r2_service_global.save_system_logs_async({
@@ -491,7 +452,6 @@ async def find_new_opportunities_async():
         
         if not high_potential_candidates:
             print("🔄 لا توجد مرشحين متخصصين، جلب مرشحين عامين...")
-            # ✅ استرجاع مرشحين عامين كبديل
             high_potential_candidates = await data_manager_global.find_high_potential_candidates(20)
             if high_potential_candidates:
                 for candidate in high_potential_candidates:
@@ -512,10 +472,9 @@ async def find_new_opportunities_async():
             chunk_data = await data_manager_global.get_fast_pass_data_async(chunk)
             
             print(f"⏳ Processing and scoring chunk {index//CHUNK_SIZE + 1}...")
-            # ✅ تحديث market_context قبل معالجة كل شريحة
             updated_market_context = await data_manager_global.get_market_context_async()
             if not updated_market_context:
-                updated_market_context = market_context  # استخدام السياق القديم كبديل
+                updated_market_context = market_context
                 
             feature_processor = FeatureProcessor(updated_market_context, data_manager_global, learning_engine_global)
             
@@ -530,7 +489,6 @@ async def find_new_opportunities_async():
             print("❌ No candidates were processed successfully.")
             return
 
-        # ✅ استخدام السياق المحدث للتصفية النهائية
         updated_market_context = await data_manager_global.get_market_context_async()
         if not updated_market_context:
             updated_market_context = market_context
@@ -578,11 +536,9 @@ async def find_new_opportunities_async():
                     continue
                 
                 if llm_analysis_data.get('action') in ["BUY", "SELL"]:
-                    # ✅ التحقق النهائي من الاستراتيجية
                     final_strategy = llm_analysis_data.get('strategy')
                     candidate_strategy = candidate.get('target_strategy', 'GENERIC')
                     
-                    # إذا كانت استراتيجية LLM غير صالحة، استخدم استراتيجية المرشح
                     if not final_strategy or final_strategy == 'unknown' or final_strategy == 'GENERIC':
                         final_strategy = candidate_strategy
                         llm_analysis_data['strategy'] = final_strategy
@@ -633,7 +589,6 @@ async def find_new_opportunities_async():
         return None
 
 async def re_analyze_open_trade_async(trade_data):
-    """Re-analyzes an open trade with enhanced strategy preservation and improved data integration"""
     symbol = trade_data.get('symbol')
     
     try:
@@ -643,7 +598,6 @@ async def re_analyze_open_trade_async(trade_data):
         
         print(f"⏳ Re-analyzing trade: {symbol} (held for {hold_minutes:.1f} minutes)")
         
-        # ✅ الإصلاح المحسن: الحفاظ على الاستراتيجية الأصلية مع التحقق الشامل
         original_strategy = trade_data.get('strategy')
         if not original_strategy or original_strategy == 'unknown':
             original_strategy = trade_data.get('decision_data', {}).get('strategy', 'GENERIC')
@@ -673,7 +627,6 @@ async def re_analyze_open_trade_async(trade_data):
             return None
         
         raw_data = ohlcv_data_list[0]
-        # ✅ تحديث market_context قبل المعالجة
         try:
             updated_market_context = await data_manager_global.get_market_context_async()
             if updated_market_context:
@@ -708,7 +661,6 @@ async def re_analyze_open_trade_async(trade_data):
         
         final_decision = _apply_patience_logic(re_analysis_decision, hold_minutes, trade_data, processed_data)
         
-        # ✅ الإصلاح النهائي: التأكد من وجود الاستراتيجية في القرار النهائي
         if not final_decision.get('strategy') or final_decision['strategy'] == 'unknown':
             final_decision['strategy'] = original_strategy
             print(f"🔧 Final re-analysis strategy fix for {symbol}: {original_strategy}")
@@ -743,29 +695,7 @@ async def re_analyze_open_trade_async(trade_data):
         })
         return None
 
-def _apply_patience_logic(decision, hold_minutes, trade_data, processed_data):
-    """Apply patience logic to prevent premature selling decisions"""
-    action = decision.get('action')
-    
-    if action == "CLOSE_TRADE" and hold_minutes < 20:
-        current_price = processed_data.get('current_price', 0)
-        entry_price = trade_data.get('entry_price', 0)
-        
-        try:
-            profit_loss_percent = ((current_price - entry_price) / entry_price) * 100
-        except (TypeError, ZeroDivisionError):
-            profit_loss_percent = 0
-        
-        if profit_loss_percent < 2:
-            print(f"🛑 Blocked premature selling! Only {hold_minutes:.1f} minutes held, PnL: {profit_loss_percent:.2f}%")
-            decision['action'] = "HOLD"
-            decision['reasoning'] = f"Patience Filter: Blocked premature sell. Held for {hold_minutes:.1f}m. Giving trade more time."
-            return decision
-    
-    return decision
-        
 async def run_bot_cycle_async():
-    """The main asynchronous bot cycle with enhanced strategy validation and improved data integration"""
     print(f"\n{'='*70}")
     print(f"⏳ New cycle initiated at: {datetime.now().isoformat()}")
     print(f"{'='*70}")
@@ -785,7 +715,6 @@ async def run_bot_cycle_async():
             open_trades = await r2_service_global.get_open_trades_async()
             print(f"✅ Found {len(open_trades)} open trade(s).")
             
-            # ✅ الإصلاح المحسن: فحص وإصلاح الاستراتيجيات الفارغة في الصفقات المفتوحة
             trades_fixed = 0
             for trade in open_trades:
                 if not trade.get('strategy') or trade['strategy'] == 'unknown':
@@ -832,19 +761,15 @@ async def run_bot_cycle_async():
                 portfolio_state = await r2_service_global.get_portfolio_state_async()
                 current_capital = portfolio_state.get("current_capital_usd", 0)
                 
-                # ✅ الإصلاح الحاسم: التحقق من رأس المال المتاح بشكل صحيح
                 print(f"💰 Current available capital: ${current_capital:.2f}")
                 
-                # إذا كان رأس المال 0، نتحقق مما إذا كان هناك خطأ في الحساب
                 if current_capital <= 0:
                     print("⚠️ Current capital is 0. Checking for potential calculation errors...")
                     
-                    # التحقق من وجود صفقات مفتوحة
                     if len(open_trades) == 0:
                         print("🔄 No open trades but capital is 0. This might be an error.")
                         print("💡 Attempting to recover capital state...")
                         
-                        # محاولة استعادة رأس المال من الحالة الأولية
                         initial_capital = portfolio_state.get("initial_capital_usd", 10.0)
                         if initial_capital > 0:
                             portfolio_state["current_capital_usd"] = initial_capital
@@ -859,7 +784,6 @@ async def run_bot_cycle_async():
                     if new_opportunity:
                         print(f"✅ Opportunity for {new_opportunity['symbol']} confirmed! Saving trade. Strategy: {new_opportunity.get('strategy')}")
                         
-                        # ✅ التحقق النهائي قبل الحفظ
                         if not new_opportunity['decision'].get('strategy'):
                             new_opportunity['decision']['strategy'] = new_opportunity.get('strategy', 'GENERIC')
                             print(f"🔧 Final pre-save strategy fix: {new_opportunity['decision']['strategy']}")
@@ -912,14 +836,11 @@ async def lifespan(application: FastAPI):
         data_manager_global = DataManager(contracts_database)
         await data_manager_global.initialize()
 
-        # ✅ تهيئة نظام التعلم مع تمرير data_manager
         learning_engine_global = LearningEngine(r2_service_global, data_manager_global)
-        await learning_engine_global.initialize_enhanced()  # ✅ استخدام التهيئة المحسنة
+        await learning_engine_global.initialize_enhanced()
         
-        # ✅ إجبار تحديث الاستراتيجيات من البيانات الحالية
         await learning_engine_global.force_strategy_learning()
         
-        # ✅ التحقق من أن الأوزان يتم تحميلها
         if learning_engine_global.initialized:
             weights = await learning_engine_global.get_optimized_strategy_weights("bull_market")
             print(f"🎯 الأوزان المحملة: {weights}")
@@ -954,18 +875,15 @@ application = FastAPI(lifespan=lifespan)
 
 @application.get("/run-cycle")
 async def run_cycle_api():
-    """API endpoint to trigger the bot cycle."""
     asyncio.create_task(run_bot_cycle_async())
     return {"message": "Bot cycle initiated in the background."}
 
 @application.get("/health")
 async def health_check():
-    """Detailed health check."""
     learning_metrics = {}
     if learning_engine_global and learning_engine_global.initialized:
         learning_metrics = await learning_engine_global.calculate_performance_metrics()
     
-    # 🆕 الحصول على إحصائيات استخدام API المحسنة
     api_stats = {}
     if data_manager_global:
         api_stats = data_manager_global.get_performance_stats()
@@ -982,12 +900,11 @@ async def health_check():
         },
         "market_state_ok": state.MARKET_STATE_OK,
         "learning_engine": learning_metrics,
-        "api_usage_stats": api_stats.get('api_usage', {})  # 🆕 إضافة إحصائيات API
+        "api_usage_stats": api_stats.get('api_usage', {})
     }
 
 @application.get("/stats")
 async def get_performance_stats():
-    """Get performance statistics for all services."""
     try:
         market_context = await data_manager_global.get_market_context_async() if data_manager_global else {}
         
@@ -997,14 +914,13 @@ async def get_performance_stats():
             learning_stats = await learning_engine_global.calculate_performance_metrics()
             improvement_suggestions = await learning_engine_global.suggest_improvements()
         
-        # 🆕 الحصول على إحصائيات API المحسنة
         api_stats = {}
         if data_manager_global:
             api_stats = data_manager_global.get_performance_stats()
         
         stats = {
             "timestamp": datetime.now().isoformat(),
-            "data_manager": api_stats,  # 🆕 استخدام الإحصائيات المحسنة
+            "data_manager": api_stats,
             "market_state": {
                 "is_healthy": state.MARKET_STATE_OK,
                 "description": "Market is healthy for trading" if state.MARKET_STATE_OK else "Market conditions are unfavorable",
@@ -1016,7 +932,7 @@ async def get_performance_stats():
             },
             "learning_engine": learning_stats,
             "improvement_suggestions": improvement_suggestions,
-            "enhanced_features": {  # 🆕 قسم الميزات المحسنة
+            "enhanced_features": {
                 "netflow_analysis": True,
                 "enhanced_whale_tracking": True,
                 "dynamic_strategy_selection": True
@@ -1028,7 +944,6 @@ async def get_performance_stats():
 
 @application.get("/logs/status")
 async def get_logs_status():
-    """Get status of logging system."""
     try:
         open_trades = await r2_service_global.get_open_trades_async()
         portfolio_state = await r2_service_global.get_portfolio_state_async()
@@ -1039,13 +954,12 @@ async def get_logs_status():
             "current_capital": portfolio_state.get("current_capital_usd", 0),
             "total_trades": portfolio_state.get("total_trades", 0),
             "timestamp": datetime.now().isoformat(),
-            "enhanced_analysis": True  # 🆕 إشارة إلى استخدام التحليل المحسن
+            "enhanced_analysis": True
         }
     except Exception as error:
         raise HTTPException(status_code=500, detail=f"Failed to get logs status: {str(error)}")
 
 async def cleanup_on_shutdown():
-    """Cleanup function for graceful shutdown."""
     global r2_service_global, data_manager_global, realtime_monitor, learning_engine_global
     print("\n🛑 Shutdown signal received. Cleaning up...")
     
@@ -1077,7 +991,6 @@ async def cleanup_on_shutdown():
     print("✅ Cleanup completed.")
 
 def signal_handler(signum, frame):
-    """Handle shutdown signals."""
     print(f"\n⚠️ Received signal {signum}")
     asyncio.create_task(cleanup_on_shutdown())
     sys.exit(0)