Update backtest_engine.py

backtest_engine.py

@@ -1,5 +1,5 @@
 # ============================================================
-# 🧪 backtest_engine.py (V71.1 - GEM-Architect: 6-Frame Turbo)
+# 🧪 backtest_engine.py (V71.2 - GEM-Architect: Memory-Safe Turbo)
 # ============================================================
 
 import asyncio
@@ -9,6 +9,7 @@ import time
 import logging
 import itertools 
 import os
+import gc # ✅ مكتبة تنظيف الذاكرة
 import concurrent.futures
 from typing import Dict, Any, List
 
@@ -43,7 +44,7 @@ class HeavyDutyBacktester:
         ]
         
         if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
-        print(f"🧪 [Backtest V71.1] 6-Frame Turbo Mode (1m Included).")
+        print(f"🧪 [Backtest V71.2] Memory-Safe Turbo Mode (GC Enabled).")
 
     # ==============================================================
     # 🛠️ Helpers
@@ -53,7 +54,7 @@ class HeavyDutyBacktester:
         return df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].values.tolist()
 
     # ==============================================================
-    # PHASE 1: Generate Truth Data (Multi-Frame Vectorization)
+    # PHASE 1: Generate Truth Data (With Memory Cleanup)
     # ==============================================================
     async def generate_truth_data(self):
         print(f"\n🚜 [Phase 1] Processing Logic Tree + Titan ({self.BACKTEST_DAYS} Days)...")
@@ -64,127 +65,142 @@ class HeavyDutyBacktester:
             safe_sym = sym.replace('/', '_')
             scores_file = f"{CACHE_DIR}/{safe_sym}_fullstack_scores.pkl"
             
+            # تنظيف الذاكرة قبل البدء بأي عملة جديدة
+            gc.collect()
+            
             if os.path.exists(scores_file):
                 print(f"   📂 {sym} scores ready. Skipping.")
                 continue
 
             print(f"   ⚙️ Simulating {sym}...", end="", flush=True)
             
-            # 1. جلب بيانات الدقيقة الخام
+            # المتغيرات التي تحتاج تنظيف لاحقاً
             all_candles_1m = []
-            current_since = start_time_ms
-            while current_since < end_time_ms:
-                try:
-                    batch = await self.dm.exchange.fetch_ohlcv(sym, '1m', since=current_since, limit=1000)
-                    if not batch: break
-                    last_ts = batch[-1][0]
-                    if last_ts <= current_since: break
-                    all_candles_1m.extend(batch)
-                    current_since = last_ts + 1
-                    await asyncio.sleep(0.01) 
-                    if current_since >= end_time_ms: break
-                except: await asyncio.sleep(0.5)
-            
-            all_candles_1m = [c for c in all_candles_1m if c[0] <= end_time_ms]
-            if not all_candles_1m: 
-                print(" No Data.")
-                continue
-
-            df_1m = pd.DataFrame(all_candles_1m, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
-            df_1m['datetime'] = pd.to_datetime(df_1m['timestamp'], unit='ms')
-            df_1m.set_index('datetime', inplace=True)
-            df_1m = df_1m.sort_index()
-            
-            # 🔥🔥🔥 التحضير المسبق (Vectorization) 🔥🔥🔥
-            agg_dict = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
-            
+            df_1m = None
             frames = {}
             
-            # ✅ إضافة إطار الدقيقة (1m) مباشرة
-            # لا نحتاج Resample لأنه هو الأساس، فقط نضيف عمود timestamp
-            df_1m_ready = df_1m.copy()
-            df_1m_ready['timestamp'] = df_1m_ready.index.astype(np.int64) // 10**6
-            frames['1m'] = df_1m_ready
-
-            # ✅ إضافة باقي الأطر (Resampling)
-            for tf_str, tf_code in [('5m', '5T'), ('15m', '15T'), ('1h', '1h'), ('4h', '4h'), ('1d', '1D')]:
-                resampled = df_1m.resample(tf_code).agg(agg_dict).dropna()
-                resampled['timestamp'] = resampled.index.astype(np.int64) // 10**6
-                frames[tf_str] = resampled
+            try:
+                # 1. جلب بيانات الدقيقة
+                current_since = start_time_ms
+                while current_since < end_time_ms:
+                    try:
+                        batch = await self.dm.exchange.fetch_ohlcv(sym, '1m', since=current_since, limit=1000)
+                        if not batch: break
+                        last_ts = batch[-1][0]
+                        if last_ts <= current_since: break
+                        all_candles_1m.extend(batch)
+                        current_since = last_ts + 1
+                        # زيادة وقت الراحة قليلاً لإعطاء فرصة للنظام الحي
+                        await asyncio.sleep(0.02) 
+                        if current_since >= end_time_ms: break
+                    except: await asyncio.sleep(0.5)
+                
+                all_candles_1m = [c for c in all_candles_1m if c[0] <= end_time_ms]
+                if not all_candles_1m: 
+                    print(" No Data.")
+                    continue
 
-            ai_results = []
-            
-            # نتحرك بناءً على إطار الـ 5 دقائق كقاعدة زمنية
-            valid_indices = frames['5m'].index[500:] 
-            
-            for t_idx in valid_indices:
-                current_timestamp = int(t_idx.timestamp() * 1000)
+                df_1m = pd.DataFrame(all_candles_1m, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
+                df_1m['datetime'] = pd.to_datetime(df_1m['timestamp'], unit='ms')
+                df_1m.set_index('datetime', inplace=True)
+                df_1m = df_1m.sort_index()
                 
-                # 🔥 قص البيانات لجميع الأطر الستة (1m included)
-                ohlcv_data = {}
+                # 🔥🔥🔥 Vectorization 🔥🔥🔥
+                agg_dict = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
                 
-                try:
-                    # ✅ [FIXED] إضافة 1m للحزمة (نأخذ 500 شمعة للدقة العالية)
-                    ohlcv_data['1m']  = self.df_to_list(frames['1m'].loc[:t_idx].tail(500))
-                    
-                    ohlcv_data['5m']  = self.df_to_list(frames['5m'].loc[:t_idx].tail(200))
-                    ohlcv_data['15m'] = self.df_to_list(frames['15m'].loc[:t_idx].tail(200))
-                    ohlcv_data['1h']  = self.df_to_list(frames['1h'].loc[:t_idx].tail(200))
-                    ohlcv_data['4h']  = self.df_to_list(frames['4h'].loc[:t_idx].tail(100))
-                    ohlcv_data['1d']  = self.df_to_list(frames['1d'].loc[:t_idx].tail(50))
-                except: continue 
+                # 1m Direct
+                df_1m_ready = df_1m.copy()
+                df_1m_ready['timestamp'] = df_1m_ready.index.astype(np.int64) // 10**6
+                frames['1m'] = df_1m_ready
 
-                if len(ohlcv_data['1h']) < 60: continue
-                
-                current_price = frames['5m'].loc[t_idx]['close']
+                # Resampling
+                for tf_str, tf_code in [('5m', '5T'), ('15m', '15T'), ('1h', '1h'), ('4h', '4h'), ('1d', '1D')]:
+                    resampled = df_1m.resample(tf_code).agg(agg_dict).dropna()
+                    resampled['timestamp'] = resampled.index.astype(np.int64) // 10**6
+                    frames[tf_str] = resampled
 
-                # 1. استدعاء شجرة المنطق (Logic Tree)
-                logic_packet = {
-                    'symbol': sym,
-                    'ohlcv_1h': ohlcv_data['1h'][-60:],   
-                    'ohlcv_15m': ohlcv_data['15m'][-60:], 
-                    'change_24h': 0.0 
-                }
+                ai_results = []
+                valid_indices = frames['5m'].index[500:] 
                 
-                try:
-                    if len(ohlcv_data['1h']) >= 24:
-                        p_now = ohlcv_data['1h'][-1][4]
-                        p_old = ohlcv_data['1h'][-24][4]
-                        logic_packet['change_24h'] = ((p_now - p_old) / p_old) * 100
-                except: pass
+                for t_idx in valid_indices:
+                    # تحويل الوقت
+                    current_timestamp = int(t_idx.timestamp() * 1000)
+                    
+                    # 🔥 Fast Slicing
+                    ohlcv_data = {}
+                    try:
+                        ohlcv_data['1m']  = self.df_to_list(frames['1m'].loc[:t_idx].tail(500))
+                        ohlcv_data['5m']  = self.df_to_list(frames['5m'].loc[:t_idx].tail(200))
+                        ohlcv_data['15m'] = self.df_to_list(frames['15m'].loc[:t_idx].tail(200))
+                        ohlcv_data['1h']  = self.df_to_list(frames['1h'].loc[:t_idx].tail(200))
+                        ohlcv_data['4h']  = self.df_to_list(frames['4h'].loc[:t_idx].tail(100))
+                        ohlcv_data['1d']  = self.df_to_list(frames['1d'].loc[:t_idx].tail(50))
+                    except: continue 
 
-                logic_result = self.dm._apply_logic_tree(logic_packet)
-                signal_type = logic_result.get('type', 'NONE')
-                l1_score = logic_result.get('score', 0.0)
-                
-                # 2. استدعاء Titan (Real AI)
-                real_titan = 0.5
-                if signal_type in ['BREAKOUT', 'REVERSAL']:
-                    raw_data_for_proc = {
+                    if len(ohlcv_data['1h']) < 60: continue
+                    
+                    current_price = frames['5m'].loc[t_idx]['close']
+
+                    # 1. Logic Tree Check
+                    logic_packet = {
                         'symbol': sym,
-                        'ohlcv': ohlcv_data, # ✅ الآن تحتوي على 1m
-                        'current_price': current_price
+                        'ohlcv_1h': ohlcv_data['1h'][-60:],   
+                        'ohlcv_15m': ohlcv_data['15m'][-60:], 
+                        'change_24h': 0.0 
                     }
+                    
                     try:
-                        proc_res = await self.proc.process_compound_signal(raw_data_for_proc)
-                        if proc_res:
-                            real_titan = proc_res.get('titan_score', 0.5)
+                        if len(ohlcv_data['1h']) >= 24:
+                            p_now = ohlcv_data['1h'][-1][4]
+                            p_old = ohlcv_data['1h'][-24][4]
+                            logic_packet['change_24h'] = ((p_now - p_old) / p_old) * 100
                     except: pass
 
-                    ai_results.append({
-                        'timestamp': current_timestamp,
-                        'symbol': sym,
-                        'close': current_price,
-                        'real_titan': real_titan, 
-                        'signal_type': signal_type,
-                        'l1_score': l1_score
-                    })
+                    logic_result = self.dm._apply_logic_tree(logic_packet)
+                    signal_type = logic_result.get('type', 'NONE')
+                    l1_score = logic_result.get('score', 0.0)
+                    
+                    # 2. Titan AI Check
+                    real_titan = 0.5
+                    if signal_type in ['BREAKOUT', 'REVERSAL']:
+                        raw_data_for_proc = {
+                            'symbol': sym,
+                            'ohlcv': ohlcv_data, 
+                            'current_price': current_price
+                        }
+                        try:
+                            # نستدعي Titan فقط عند الحاجة القصوى
+                            proc_res = await self.proc.process_compound_signal(raw_data_for_proc)
+                            if proc_res:
+                                real_titan = proc_res.get('titan_score', 0.5)
+                        except: pass
+
+                        ai_results.append({
+                            'timestamp': current_timestamp,
+                            'symbol': sym,
+                            'close': current_price,
+                            'real_titan': real_titan, 
+                            'signal_type': signal_type,
+                            'l1_score': l1_score
+                        })
+                
+                if ai_results:
+                    pd.DataFrame(ai_results).to_pickle(scores_file)
+                    print(f" ✅ Saved ({len(ai_results)} signals).")
+                else:
+                    print(" ⚠️ No signals.")
+
+            except Exception as e:
+                print(f" ❌ Error: {e}")
             
-            if ai_results:
-                pd.DataFrame(ai_results).to_pickle(scores_file)
-                print(f" ✅ Saved ({len(ai_results)} signals).")
-            else:
-                print(" ⚠️ No signals.")
+            finally:
+                # 🧹 CLEANUP MEMORY FORCEFULLY 🧹
+                # هذا الجزء يضمن عدم تراكم البيانات في الرام
+                del all_candles_1m
+                del df_1m
+                del frames
+                # استدعاء جامع القمامة يدوياً
+                gc.collect()
 
     # ==============================================================
     # PHASE 2: Portfolio Digital Twin Engine