Update backtest_engine.py

backtest_engine.py

@@ -1,9 +1,9 @@
 # ============================================================
-# 🧪 backtest_engine.py (V51.1 - GEM-Architect: Bug Fix)
+# 🧪 backtest_engine.py (V60.1 - GEM-Architect: Configurable Real-Deal)
 # ============================================================
 # التحديثات:
-# 1. إصلاح خطأ 'numpy.ndarray object has no attribute values'.
-# 2. تحسين التعامل مع المصفوفات لضمان استقرار الباكتست.
+# 1. إضافة متغير `BACKTEST_DAYS` للتحكم السهل في مدة البيانات.
+# 2. الحفاظ على المحرك الحقيقي (Real Models) بدون محاكاة.
 # ============================================================
 
 import asyncio
@@ -17,85 +17,120 @@ import shutil
 import concurrent.futures
 from typing import Dict, Any, List
 
-# استيراد خفيف لتجنب تضارب التوازي
-from ml_engine.processor import SystemLimits 
+# استيراد المحركات الحقيقية
+from ml_engine.processor import MLProcessor, SystemLimits 
 from ml_engine.data_manager import DataManager
 from learning_hub.adaptive_hub import StrategyDNA
+from r2 import R2Service
 
 logging.getLogger('ml_engine').setLevel(logging.WARNING)
-CACHE_DIR = "backtest_cache_grid"
+CACHE_DIR = "backtest_real_scores"
 
-class MassiveOptimizer:
-    def __init__(self, data_manager):
+class HeavyDutyBacktester:
+    def __init__(self, data_manager, processor):
         self.dm = data_manager
-        # 3 = سريع (تجربة) | 5 = متوسط (~3000) | 10 = دقيق (~1000)
+        self.proc = processor
         self.GRID_DENSITY = 10 
         
+        # 🔥🔥🔥 إعدادات التحكم في الوقت (غير هذا الرقم كما تشاء) 🔥🔥🔥
+        self.BACKTEST_DAYS = 7 # عدد الأيام التي سيتم فحصها
+        # ============================================================
+
         self.TARGET_COINS = [
             'BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'BNB/USDT', 'XRP/USDT',
-            'DOGE/USDT', 'ADA/USDT', 'AVAX/USDT', 'LINK/USDT', 'LTC/USDT',
-            'NEAR/USDT', 'RUNE/USDT', 'INJ/USDT', 'PEPE/USDT', 'SHIB/USDT'
+            'DOGE/USDT', 'ADA/USDT', 'AVAX/USDT'
         ]
         
         if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
-        print(f"🧪 [Backtest Engine V51.1] Grid Density set to: {self.GRID_DENSITY}")
+        print(f"🧪 [Backtest V60.1] Heavy Duty Mode (Real Models). Period: {self.BACKTEST_DAYS} Days.")
 
-    async def fetch_deep_history(self):
-        """تحميل البيانات وتجهيزها للمعالجة السريعة"""
-        print(f"\n⏳ [Data] Pre-fetching history for Grid Search...")
+    # ==============================================================
+    # PHASE 1: The Heavy Lift (Running Real AI Models)
+    # ==============================================================
+    async def generate_truth_data(self):
+        """
+        تشغيل النماذج الحقيقية على البيانات التاريخية وحفظ النتائج.
+        """
+        print(f"\n🚜 [Phase 1] Running REAL Models on History ({self.BACKTEST_DAYS} Days)...")
+        
         end_time_ms = int(time.time() * 1000)
-        start_time_ms = end_time_ms - (14 * 24 * 60 * 60 * 1000) 
+        # استخدام المتغير السهل هنا
+        start_time_ms = end_time_ms - (self.BACKTEST_DAYS * 24 * 60 * 60 * 1000) 
         
         for sym in self.TARGET_COINS:
             safe_sym = sym.replace('/', '_')
-            file_path = f"{CACHE_DIR}/{safe_sym}.pkl"
+            # نقوم بتضمين عدد الأيام في اسم الملف لكي لا يختلط ببيانات قديمة
+            scores_file = f"{CACHE_DIR}/{safe_sym}_scores_{self.BACKTEST_DAYS}d.pkl"
             
-            if os.path.exists(file_path): continue
+            if os.path.exists(scores_file):
+                print(f"   📂 {sym} scores already computed. Skipping.")
+                continue
 
-            print(f"   ⬇️  Downloading {sym}...", end="", flush=True) 
-            try:
-                candles = await self.dm.exchange.fetch_ohlcv(sym, '15m', since=start_time_ms, limit=1000)
-                if candles:
-                    df = pd.DataFrame(candles, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
-                    df = df.drop_duplicates(subset=['timestamp']).sort_values('timestamp')
-                    for col in ['open', 'high', 'low', 'close', 'volume']: df[col] = df[col].astype(float)
-                    
-                    # حساب المؤشرات مسبقاً (Vectorized)
-                    df['ema50'] = df['close'].ewm(span=50).mean()
-                    
-                    # Scanner Proxies
-                    # RSI Manual Calculation for speed
-                    delta = df['close'].diff()
-                    gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
-                    loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
-                    rs = gain / loss
-                    df['rsi'] = 100 - (100 / (1 + rs))
+            print(f"   ⚙️ Processing {sym} with ML Engine...", end="", flush=True)
+            
+            # جلب الشموع (نطلب شموع أكثر قليلاً لضمان وجود بيانات كافية للمؤشرات)
+            candles = await self.dm.exchange.fetch_ohlcv(sym, '15m', since=start_time_ms, limit=2000)
+            if not candles:
+                print(" ❌ No Data")
+                continue
+                
+            df = pd.DataFrame(candles, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
+            df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
+            df.set_index('datetime', inplace=True)
+            
+            ai_results = []
+            
+            # محاكاة التداول شمعة بشمعة (Real Inference Loop)
+            # نبدأ من الشمعة 100 لتوفير بيانات كافية
+            start_idx = 100 if len(df) > 100 else 0
+            
+            for i in range(start_idx, len(df)):
+                # تجهيز النافذة الزمنية كما يراها المعالج في الوقت الحي
+                window = df.iloc[i-100:i+1] 
+                current_price = window['close'].iloc[-1]
+                
+                # تحويل البيانات لصيغة OHLCV list
+                ohlcv_15m = window.reset_index()[['timestamp', 'open', 'high', 'low', 'close', 'volume']].values.tolist()
+                
+                # تجهيز الحزمة للمعالج
+                raw_data = {
+                    'symbol': sym,
+                    'current_price': current_price,
+                    'ohlcv': {'15m': ohlcv_15m, '1h': ohlcv_15m} # استخدام 15m كبديل لـ 1h للسرعة مع النماذج
+                }
+                
+                # 🔥 استدعاء المعالج الحقيقي (Titan + Patterns) 🔥
+                result = await self.proc.process_compound_signal(raw_data)
+                
+                if result:
+                    titan_real = result.get('titan_score', 0.5)
+                    pattern_real = result.get('patterns_score', 0.5)
                     
-                    # BB
-                    df['ma20'] = df['close'].rolling(20).mean()
-                    df['std20'] = df['close'].rolling(20).std()
-                    df['bb_upper'] = df['ma20'] + (df['std20'] * 2)
+                    # استدعاء الكاشف الحقيقي (Scanner)
+                    scanner_res = self.dm._apply_scanner_strategies(window, sym)
                     
-                    df.to_pickle(file_path)
-                    print(" ✅")
-                else:
-                    print(" ⚠️ Empty")
-            except: print(" ❌ Error")
-            await asyncio.sleep(0.5)
+                    ai_results.append({
+                        'timestamp': window.index[-1],
+                        'close': current_price,
+                        'real_titan': titan_real,
+                        'real_pattern': pattern_real,
+                        'real_scanner_data': scanner_res
+                    })
+            
+            if ai_results:
+                pd.DataFrame(ai_results).to_pickle(scores_file)
+                print(f" ✅ Done ({len(ai_results)} candles)")
+            else:
+                print(" ⚠️ Empty Results")
 
     # ==============================================================
-    # 🧠 The Worker Logic (Isolated for Speed & Accuracy)
+    # PHASE 2: The Grid Optimizer (Fast Math on Real Scores)
     # ==============================================================
     @staticmethod
-    def _worker_evaluate_batch(combinations_batch, market_data_files):
-        """
-        يقوم هذا العامل بتقييم مجموعة من التوليفات (Batch) دفعة واحدة.
-        """
+    def _worker_optimize(combinations_batch, scores_files):
         results = []
-        
-        # تحميل البيانات للذاكرة (يتم مرة واحدة لكل Worker)
         dfs = []
-        for fp in market_data_files:
+        for fp in scores_files:
             try: dfs.append(pd.read_pickle(fp))
             except: pass
             
@@ -104,130 +139,105 @@ class MassiveOptimizer:
             total_trades = 0
             
             w_titan = config['w_titan']
-            w_scanner = config['w_scanner']
+            w_scanner = config['w_scanner'] 
             entry_thresh = config['thresh']
             
             for df in dfs:
-                # ---------------------------------------------------
-                # ⚡ Vectorized Signal Logic
-                # ---------------------------------------------------
-                # 1. Titan Score (Simulated)
-                titan_score = np.where(df['close'] > df['ema50'], 0.9, 0.3)
-                
-                # 2. Scanner Score (Simulated)
-                rsi_cond = np.where(df['rsi'] < 60, 1.0, 0.4) 
-                bb_cond = np.where(df['close'] > df['bb_upper'], 1.0, 0.0) 
-                
-                scanner_score = (rsi_cond * 0.7) + (bb_cond * 0.3)
-                
-                # 3. Final Weighted Score
-                final_score = (titan_score * w_titan) + (scanner_score * w_scanner)
-                final_score = final_score / (w_titan + w_scanner)
-                
-                # 4. Generate Entries (Boolean Numpy Array)
-                signals = (final_score > entry_thresh)
-                
-                # 5. Fast Loop for PnL
-                prices = df['close'].values
-                
-                # 🔥 FIX: signals هو أصلاً numpy array، لا نحتاج .values
-                sigs = signals 
-                
                 in_pos = False
                 entry_p = 0.0
                 
-                for i in range(len(prices)-1):
-                    if not in_pos and sigs[i]:
+                for idx, row in df.iterrows():
+                    # 1. حساب Scanner Score
+                    s_data = row['real_scanner_data']
+                    active_cnt = sum([1 for k,v in s_data.items() if v['active']])
+                    scanner_score = (active_cnt * 100) / 4 
+                    scanner_score /= 100.0 
+                    
+                    real_titan = row['real_titan']
+                    
+                    # 2. المعادلة الموزونة
+                    final_score = (real_titan * w_titan) + (scanner_score * w_scanner)
+                    final_score /= (w_titan + w_scanner)
+                    
+                    # 3. محاكاة الدخول
+                    if not in_pos and final_score >= entry_thresh:
                         in_pos = True
-                        entry_p = prices[i]
+                        entry_p = row['close']
                     elif in_pos:
-                        curr = prices[i]
-                        pnl = (curr - entry_p) / entry_p
+                        # 4. محاكاة الخروج (TP/SL)
+                        pnl = (row['close'] - entry_p) / entry_p
                         
-                        if pnl > 0.03 or pnl < -0.015: 
+                        # شروط الخروج (يمكن تعديلها هنا أيضاً)
+                        if pnl > 0.03 or pnl < -0.02:
                             total_pnl += pnl
                             total_trades += 1
                             in_pos = False
                             
-            if total_trades > 5: 
+            if total_trades > 3: # تصفية النتائج القليلة جداً
                 results.append({
                     'config': config,
                     'pnl': total_pnl,
                     'trades': total_trades,
-                    'score': total_pnl * np.log(total_trades) 
+                    'score': total_pnl 
                 })
-                
         return results
 
-    # ==============================================================
-    # 🚀 The Grid Generator
-    # ==============================================================
     async def run_optimization(self):
-        market_files = [os.path.join(CACHE_DIR, f) for f in os.listdir(CACHE_DIR) if f.endswith('.pkl')]
-        if not market_files:
-            await self.fetch_deep_history()
-            market_files = [os.path.join(CACHE_DIR, f) for f in os.listdir(CACHE_DIR) if f.endswith('.pkl')]
+        # 1. تشغيل النماذج الحقيقية
+        await self.generate_truth_data()
         
-        print(f"🧩 [Optimizer] Generating Grid with Density={self.GRID_DENSITY}...")
+        # 2. تجهيز الشبكة
+        score_files = [os.path.join(CACHE_DIR, f) for f in os.listdir(CACHE_DIR) if f.endswith(f'_scores_{self.BACKTEST_DAYS}d.pkl')]
+        if not score_files:
+            print("❌ No AI scores found. Phase 1 failed?")
+            return
+
+        print(f"\n🧩 [Phase 2] Running Grid Search on REAL AI SCORES...")
         
-        w_titan_range = np.linspace(0.2, 0.9, num=self.GRID_DENSITY)
-        w_scanner_range = np.linspace(0.1, 0.8, num=self.GRID_DENSITY)
+        w_titan_range = np.linspace(0.1, 0.9, num=self.GRID_DENSITY)
+        w_scanner_range = np.linspace(0.1, 0.9, num=self.GRID_DENSITY)
         thresh_range = np.linspace(0.50, 0.80, num=self.GRID_DENSITY)
         
         combinations = []
         for wt, ws, th in itertools.product(w_titan_range, w_scanner_range, thresh_range):
-            combinations.append({
-                'w_titan': round(float(wt), 2), 
-                'w_scanner': round(float(ws), 2), 
-                'thresh': round(float(th), 2)
-            })
+            combinations.append({'w_titan': round(wt, 2), 'w_scanner': round(ws, 2), 'thresh': round(th, 2)})
             
-        print(f"   📊 Total Unique Combinations: {len(combinations):,}")
-        print(f"   🚀 Est. Processing Time: {len(combinations)/2000:.1f} minutes (on parallel cores)")
+        print(f"   📊 Combinations: {len(combinations):,}")
         
-        start_time = time.time()
         final_results = []
+        batch_size = max(50, len(combinations) // (os.cpu_count() * 2))
+        batches = [combinations[i:i+batch_size] for i in range(0, len(combinations), batch_size)]
         
-        batch_size = max(100, len(combinations) // (os.cpu_count() * 4))
-        batches = [combinations[i:i + batch_size] for i in range(0, len(combinations), batch_size)]
-        
-        print(f"   🔥 Firing up {os.cpu_count()} CPU Cores for {len(batches)} batches...")
-        
-        loop = asyncio.get_running_loop()
         with concurrent.futures.ProcessPoolExecutor() as executor:
-            futures = [executor.submit(self._worker_evaluate_batch, batch, market_files) for batch in batches]
-            
+            futures = [executor.submit(self._worker_optimize, batch, score_files) for batch in batches]
             for future in concurrent.futures.as_completed(futures):
-                try:
-                    res = future.result()
-                    final_results.extend(res)
-                except Exception as e: print(f"Batch Error: {e}")
+                try: final_results.extend(future.result())
+                except Exception as e: print(f"Grid Error: {e}")
 
-        elapsed = time.time() - start_time
-        print(f"✅ Optimization Finished in {elapsed:.2f}s")
-        
         if not final_results:
-            print("⚠️ No profitable strategies found (Check Data or lowered thresholds).")
+            print("⚠️ No profitable config found.")
             return None
             
-        best_result = sorted(final_results, key=lambda x: x['score'], reverse=True)[0]
+        best = sorted(final_results, key=lambda x: x['pnl'], reverse=True)[0]
         
         print("\n" + "="*60)
-        print(f"🏆 GRAND CHAMPION (From {len(combinations):,} options):")
-        print(f"   💰 Total Score (PnL): {best_result['pnl']:.2f}")
-        print(f"   📊 Trades: {best_result['trades']}")
-        print(f"   🧬 DNA: {best_result['config']}")
+        print(f"🏆 REAL-MODEL CHAMPION ({self.BACKTEST_DAYS} Days):")
+        print(f"   💰 PnL: {best['pnl']:.2f}")
+        print(f"   📊 Trades: {best['trades']}")
+        print(f"   🧬 Config: {best['config']}")
         print("="*60)
         
-        return best_result['config']
+        return best['config']
 
 async def run_strategic_optimization_task():
-    print("\n🧪 [STRATEGIC BACKTEST V51.1] Starting Massive Grid Search...")
-    from r2 import R2Service
+    print("\n🧪 [STRATEGIC BACKTEST V60.1] Starting Heavy Duty Optimization...")
     r2 = R2Service()
     dm = DataManager(None, None, r2)
+    proc = MLProcessor(dm)
+    await dm.initialize()
+    await proc.initialize() 
     
-    optimizer = MassiveOptimizer(dm)
+    optimizer = HeavyDutyBacktester(dm, proc)
     best_config = await optimizer.run_optimization()
     
     if best_config:
@@ -235,19 +245,17 @@ async def run_strategic_optimization_task():
         hub = AdaptiveHub(r2)
         await hub.initialize()
         
-        regime = "RANGE" 
+        regime = "RANGE"
         if regime in hub.strategies:
-            print(f"💉 Injecting new DNA into {regime} Strategy...")
+            print(f"💉 Injecting REAL DNA into {regime}...")
             st = hub.strategies[regime]
-            
             st.model_weights['titan'] = best_config['w_titan']
-            # نستخدم 'patterns' لحفظ وزن الـ Scanner مؤقتاً أو كما اتفقنا سابقاً
             st.model_weights['patterns'] = best_config['w_scanner'] 
             st.filters['l1_min_score'] = best_config['thresh'] * 100
             
             await hub._save_state_to_r2()
             hub._inject_current_parameters()
-            print("✅ [System] DNA Updated & Active.")
+            print("✅ [System] DNA Updated.")
 
     await dm.close()