Update learning_hub/adaptive_hub.py

learning_hub/adaptive_hub.py

@@ -1,6 +1,6 @@
 # ==============================================================================
 # 🧠 learning_hub/adaptive_hub.py 
-# (V54.1 - GEM-Architect: Fix Missing Method)
+# (V55.0 - GEM-Architect: The Judge Logic)
 # ==============================================================================
 
 import json
@@ -11,12 +11,22 @@ from typing import Dict, Any, List
 from ml_engine.processor import SystemLimits
 
 class StrategyDNA:
-    def __init__(self, name, model_weights, ob_settings, filters, guard_settings=None):
+    def __init__(self, name, model_weights, ob_settings, filters, guard_settings=None, backtest_performance=None):
         self.name = name
         self.model_weights = model_weights
         self.ob_settings = ob_settings
         self.filters = filters
         self.guard_settings = guard_settings if guard_settings else {}
+        
+        # ✅ الإضافة الجديدة: سجل أداء الباكتست الذي أنتج هذه الإعدادات
+        # هذا هو "السجل الرياضي" للبطل الحالي
+        self.backtest_performance = backtest_performance if backtest_performance else {
+            "net_profit": -9999.0, # قيمة منخفضة جداً للبداية
+            "win_rate": 0.0,
+            "total_trades": 0,
+            "date_recorded": "N/A"
+        }
+        
         self.stats = {"wins": 0, "losses": 0, "win_rate": 0.0}
 
     def to_dict(self):
@@ -26,6 +36,7 @@ class StrategyDNA:
             "ob_settings": self.ob_settings,
             "filters": self.filters,
             "guard_settings": self.guard_settings,
+            "backtest_performance": self.backtest_performance, # ✅ حفظ السجل
             "stats": self.stats
         }
 
@@ -35,7 +46,7 @@ class AdaptiveHub:
         self.dna_file_key = "learning/strategic_dna_v5_struct.json" 
         self.current_market_regime = "RANGE"
         self.strategies: Dict[str, StrategyDNA] = {}
-        print("🧠 [AdaptiveHub V54.1] Core Initialized (Method Restored).")
+        print("🧠 [AdaptiveHub V55.0] The Judge Core Initialized.")
 
     async def initialize(self):
         try:
@@ -54,7 +65,7 @@ class AdaptiveHub:
 
     def _create_default_dna(self):
         default_guards = {"hydra_crash": 0.85, "hydra_giveback": 0.70, "legacy_v2": 0.95, "legacy_v3": 0.95}
-        
+        # القيم الافتراضية ليس لها سجل باكتست، لذا ستكون سهلة الهزيمة
         self.strategies["BULL"] = StrategyDNA("BULL", {"titan": 0.50, "structure": 0.30, "sniper": 0.20}, {"wall_ratio_limit": 0.60, "imbalance_thresh": 0.5}, {"l1_min_score": 0.55, "l3_conf_thresh": 0.60}, default_guards)
         self.strategies["BEAR"] = StrategyDNA("BEAR", {"titan": 0.30, "structure": 0.40, "sniper": 0.30}, {"wall_ratio_limit": 0.30, "imbalance_thresh": 0.7}, {"l1_min_score": 0.75, "l3_conf_thresh": 0.75}, default_guards)
         self.strategies["RANGE"] = StrategyDNA("RANGE", {"titan": 0.40, "structure": 0.40, "sniper": 0.20}, {"wall_ratio_limit": 0.40, "imbalance_thresh": 0.6}, {"l1_min_score": 0.65, "l3_conf_thresh": 0.65}, default_guards)
@@ -62,36 +73,75 @@ class AdaptiveHub:
 
     def _load_from_dict(self, data):
         for key, val in data.get("strategies", {}).items():
-            self.strategies[key] = StrategyDNA(val["name"], val["model_weights"], val["ob_settings"], val["filters"], val.get("guard_settings", {}))
+            self.strategies[key] = StrategyDNA(
+                val["name"], 
+                val["model_weights"], 
+                val["ob_settings"], 
+                val["filters"], 
+                val.get("guard_settings", {}),
+                val.get("backtest_performance", None) # ✅ تحميل السجل
+            )
         self.current_market_regime = data.get("current_regime", "RANGE")
 
-    # ✅ الدالة المفقودة تمت استعادتها هنا
+    # 🔥🔥🔥 الدالة الجديدة: الحكم (The Judge) 🔥🔥🔥
+    def submit_challenger(self, regime: str, new_config: dict, new_stats: dict) -> bool:
+        """
+        تقارن بين المتحدي الجديد والبطل الحالي.
+        تعيد True إذا فاز الجديد وتم التحديث، و False إذا تم رفضه.
+        """
+        if regime not in self.strategies: return False
+        
+        champion = self.strategies[regime]
+        old_stats = champion.backtest_performance
+        
+        # معايير التحكيم:
+        # 1. الربح الصافي هو الملك.
+        # 2. إذا تساوى الربح، نختار نسبة الفوز الأعلى.
+        
+        new_profit = new_stats.get('net_profit', -100)
+        old_profit = old_stats.get('net_profit', -9999)
+        
+        print(f"⚖️ [JUDGE] Comparing {regime} DNA:")
+        print(f"   👑 Champion: Profit ${old_profit:.2f} | WinRate {old_stats.get('win_rate', 0):.1f}%")
+        print(f"   🥊 Challenger: Profit ${new_profit:.2f} | WinRate {new_stats.get('win_rate', 0):.1f}%")
+        
+        is_winner = False
+        
+        # القاعدة 1: الربح الصافي يجب أن يكون أعلى بوضوح (أو النظام جديد كلياً)
+        if new_profit > old_profit:
+            is_winner = True
+        # القاعدة 2: إذا الربح متقارب جداً، نفضل نسبة الفوز الأعلى
+        elif abs(new_profit - old_profit) < 0.5 and new_stats.get('win_rate', 0) > old_stats.get('win_rate', 0):
+            is_winner = True
+            
+        if is_winner:
+            print(f"   ✅ [JUDGE] Challenger WINS! Updating DNA.")
+            # تحديث الجينات
+            champion.model_weights['titan'] = new_config['w_titan']
+            champion.model_weights['structure'] = new_config['w_struct']
+            champion.filters['l1_min_score'] = new_config['thresh']
+            # تحديث سجل البطل
+            champion.backtest_performance = new_stats
+            return True
+        else:
+            print(f"   🛡️ [JUDGE] Champion retains title. Challenger rejected.")
+            return False
+
     async def register_trade_outcome(self, trade_data: Dict[str, Any]):
         try:
-            # نتأكد من وجود البيانات، أحياناً قد تأتي كائن Trade وأحياناً Dict
-            # إذا كانت كائن، نحولها لـ dict أو نصل للخاصية مباشرة
             pnl = 0.0
             if isinstance(trade_data, dict):
                 pnl = trade_data.get('profit_pct', 0.0)
-                # دعم للتسميات المختلفة
                 if 'pnl_percent' in trade_data: pnl = trade_data['pnl_percent']
             else:
-                # في حال كان كائناً (Object)
                 pnl = getattr(trade_data, 'profit_pct', 0.0)
 
             is_win = pnl > 0
-            
             if self.current_market_regime in self.strategies:
                 active_dna = self.strategies[self.current_market_regime]
                 if is_win: active_dna.stats["wins"] += 1
                 else: active_dna.stats["losses"] += 1
-                
-                # طباعة خفيفة للتأكد
-                # print(f"🧠 [AdaptiveHub] Learned: {self.current_market_regime} | Win? {is_win}")
-
-        except Exception as e:
-            print(f"❌ [AdaptiveHub] Trade Analysis Error: {e}")
-            traceback.print_exc()
+        except Exception: traceback.print_exc()
 
     def _inject_current_parameters(self):
         if self.current_market_regime not in self.strategies: return