Update learning_hub/statistical_analyzer.py

learning_hub/statistical_analyzer.py

@@ -1,11 +1,13 @@
 # learning_hub/statistical_analyzer.py
-# (V13.0 - Real-Time Weight Adaptation)
+# (V13.1 - GEM-Architect: Real-Time Weight Adaptation - Fixed Imports)
 
 import json
 import asyncio
 import traceback
 from datetime import datetime
 from collections import deque
+# ✅ [FIX] إضافة استيراد الأنواع المفقودة
+from typing import Dict, Any
 
 class StatisticalAnalyzer:
     def __init__(self, r2_service, data_manager):
@@ -20,7 +22,7 @@ class StatisticalAnalyzer:
             "monte_carlo": deque(maxlen=self.window_size)
         }
         
-        # الأوزان الحالية
+        # الأوزان الحالية (الافتراضية)
         self.current_weights = {
             "titan": 0.5,
             "patterns": 0.4,
@@ -31,12 +33,12 @@ class StatisticalAnalyzer:
         print("✅ Learning Hub: Statistical Analyzer (Live Tuner) loaded")
 
     async def initialize(self):
-        # محاولة تحميل الأوزان السابقة من R2
+        # في النسخ المستقبلية، يمكننا تحميل آخر أوزان محفوظة من R2 هنا
         self.initialized = True
 
     async def update_live_performance(self, trade_object: Dict[str, Any]):
         """
-        تحديث سجلات الأداء اللحظية وتعديل الأوزان.
+        تحديث سجلات الأداء اللحظية وتعديل الأوزان فوراً.
         """
         try:
             pnl = trade_object.get('pnl_percent', 0)
@@ -48,57 +50,76 @@ class StatisticalAnalyzer:
             # 1. تقييم كل مكون على حدة
             # هل كان المكون يتوقع الصعود (>0.6)؟ وهل نجح؟
             
-            # Titan
+            # Titan Check
             t_score = comps.get('titan_score', 0.5)
             t_correct = 1 if (t_score > 0.6 and is_win) or (t_score < 0.4 and not is_win) else 0
             self.history["titan"].append(t_correct)
             
-            # Patterns
+            # Patterns Check
             p_score = comps.get('patterns_score', 0.5)
             p_correct = 1 if (p_score > 0.6 and is_win) or (p_score < 0.4 and not is_win) else 0
             self.history["patterns"].append(p_correct)
             
-            # Monte Carlo
+            # Monte Carlo Check
             m_score = comps.get('mc_score', 0.5)
             m_correct = 1 if (m_score > 0.6 and is_win) or (m_score < 0.4 and not is_win) else 0
             self.history["monte_carlo"].append(m_correct)
             
-            # 2. إعادة حساب الأوزان (فقط إذا اكتملت العينة)
+            # 2. إعادة حساب الأوزان (فقط إذا تجمعت عينة كافية >= 5)
             if len(self.history["titan"]) >= 5:
                 await self._rebalance_weights()
                 
         except Exception as e:
-            print(f"❌ [Stats] Update failed: {e}")
+            print(f"❌ [StatsAnalyzer] Update failed: {e}")
+            traceback.print_exc()
 
     async def _rebalance_weights(self):
         """إعادة توزيع الأوزان بناءً على الدقة الحديثة"""
-        # حساب المتوسط (Accuracy) لكل مكون
-        acc_t = sum(self.history["titan"]) / len(self.history["titan"])
-        acc_p = sum(self.history["patterns"]) / len(self.history["patterns"])
-        acc_m = sum(self.history["monte_carlo"]) / len(self.history["monte_carlo"])
-        
-        # منع الأوزان الصفرية
-        acc_t = max(acc_t, 0.1)
-        acc_p = max(acc_p, 0.1)
-        acc_m = max(acc_m, 0.1)
-        
-        total = acc_t + acc_p + acc_m
-        
-        # الأوزان الجديدة
-        new_w = {
-            "titan": acc_t / total,
-            "patterns": acc_p / total,
-            "monte_carlo": acc_m / total
-        }
-        
-        # تطبيق التغيير بنعومة (Smoothing) لعدم القفز
-        # New = 90% Old + 10% Calculated
-        self.current_weights = {
-            k: (self.current_weights[k] * 0.9) + (new_w[k] * 0.1) 
-            for k in new_w
-        }
-        
-        # print(f"⚖️ [Stats] Weights Tuned: Titan={self.current_weights['titan']:.2f}, Patt={self.current_weights['patterns']:.2f}")
+        try:
+            # حساب المتوسط (Accuracy) لكل مكون
+            acc_t = sum(self.history["titan"]) / len(self.history["titan"])
+            acc_p = sum(self.history["patterns"]) / len(self.history["patterns"])
+            acc_m = sum(self.history["monte_carlo"]) / len(self.history["monte_carlo"])
+            
+            # منع الأوزان الصفرية (Smoothing)
+            acc_t = max(acc_t, 0.1)
+            acc_p = max(acc_p, 0.1)
+            acc_m = max(acc_m, 0.1)
+            
+            total = acc_t + acc_p + acc_m
+            
+            # الأوزان الجديدة المقترحة
+            new_w = {
+                "titan": acc_t / total,
+                "patterns": acc_p / total,
+                "monte_carlo": acc_m / total
+            }
+            
+            # تطبيق التغيير بنعومة (Smoothing): 90% قديم + 10% جديد
+            # هذا يمنع التقلب الشديد في الأوزان مع كل صفقة
+            self.current_weights = {
+                k: (self.current_weights[k] * 0.9) + (new_w[k] * 0.1) 
+                for k in new_w
+            }
+            
+            # (اختياري) طباعة التعديل للمراقبة
+            # print(f"⚖️ [Stats] Weights Adapted: Titan={self.current_weights['titan']:.2f}")
 
-    def get_live_weights(self):
-        return self.current_weights
+        except Exception as e:
+            print(f"❌ [StatsAnalyzer] Rebalance failed: {e}")
+
+    def get_live_weights(self) -> Dict[str, float]:
+        """إرجاع الأوزان الحالية لاستخدامها في المعالج"""
+        return self.current_weights
+
+    async def save_weights_to_r2(self):
+        """حفظ الأوزان الحالية في R2 (للاستعادة عند إعادة التشغيل)"""
+        try:
+            data = {
+                "weights": self.current_weights,
+                "last_updated": datetime.now().isoformat()
+            }
+            # نفترض وجود دالة upload_json_async في r2_service كما أضفناها سابقاً
+            await self.r2_service.upload_json_async(data, "learning_dynamic_weights.json")
+        except Exception:
+            pass