Update backtest_engine.py

backtest_engine.py

@@ -1,5 +1,5 @@
 # ============================================================
-# 🧪 backtest_engine.py (V71.2 - GEM-Architect: Memory-Safe Turbo)
+# 🧪 backtest_engine.py (V72.0 - GEM-Architect: Isolation Mode)
 # ============================================================
 
 import asyncio
@@ -9,7 +9,7 @@ import time
 import logging
 import itertools 
 import os
-import gc # ✅ مكتبة تنظيف الذاكرة
+import gc 
 import concurrent.futures
 from typing import Dict, Any, List
 
@@ -31,7 +31,6 @@ class HeavyDutyBacktester:
         self.GRID_DENSITY = 10 
         self.BACKTEST_DAYS = 7 
         
-        # 💰 إعدادات التوأم الرقمي
         self.INITIAL_CAPITAL = 10.0
         self.TRADING_FEES = 0.001 
         self.MAX_SLOTS = 4         
@@ -44,7 +43,7 @@ class HeavyDutyBacktester:
         ]
         
         if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
-        print(f"🧪 [Backtest V71.2] Memory-Safe Turbo Mode (GC Enabled).")
+        print(f"🧪 [Backtest V72.0] Isolation Mode (Zero-Retention).")
 
     # ==============================================================
     # 🛠️ Helpers
@@ -54,153 +53,158 @@ class HeavyDutyBacktester:
         return df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].values.tolist()
 
     # ==============================================================
-    # PHASE 1: Generate Truth Data (With Memory Cleanup)
+    # 🧱 Core Logic: Single Coin Processor (Isolated Scope)
     # ==============================================================
-    async def generate_truth_data(self):
-        print(f"\n🚜 [Phase 1] Processing Logic Tree + Titan ({self.BACKTEST_DAYS} Days)...")
-        end_time_ms = int(time.time() * 1000)
-        start_time_ms = end_time_ms - (self.BACKTEST_DAYS * 24 * 60 * 60 * 1000) 
+    async def _process_single_coin_task(self, sym, start_time_ms, end_time_ms):
+        """
+        دالة معزولة لمعالجة عملة واحدة. 
+        عند انتهاء هذه الدالة، يتم تدمير كل المتغيرات داخلها تلقائياً.
+        """
+        safe_sym = sym.replace('/', '_')
+        scores_file = f"{CACHE_DIR}/{safe_sym}_fullstack_scores.pkl"
         
-        for sym in self.TARGET_COINS:
-            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
+        if os.path.exists(scores_file):
+            print(f"   📂 {sym} scores ready. Skipping.")
+            return True
 
-            print(f"   ⚙️ Simulating {sym}...", end="", flush=True)
-            
-            # المتغيرات التي تحتاج تنظيف لاحقاً
-            all_candles_1m = []
-            df_1m = None
-            frames = {}
-            
+        print(f"   ⚙️ Simulating {sym}...", end="", flush=True)
+        
+        # 1. جلب البيانات
+        all_candles_1m = []
+        current_since = start_time_ms
+        
+        while current_since < end_time_ms:
             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
+                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.")
+            return False
 
-                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'}
-                
-                # 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
+        # 2. بناء الـ DataFrame
+        df_1m = pd.DataFrame(all_candles_1m, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
+        
+        # 🔥 تقليل استهلاك الذاكرة: تحويل float64 إلى float32
+        cols = ['open', 'high', 'low', 'close', 'volume']
+        df_1m[cols] = df_1m[cols].astype('float32')
+        
+        df_1m['datetime'] = pd.to_datetime(df_1m['timestamp'], unit='ms')
+        df_1m.set_index('datetime', inplace=True)
+        df_1m = df_1m.sort_index()
+        
+        # 3. Vectorization (Resampling)
+        agg_dict = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
+        frames = {}
+        
+        # 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
 
-                # 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
+        # Resampling Loop
+        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()
+            # Ensure float32 persists
+            resampled[cols] = resampled[cols].astype('float32')
+            resampled['timestamp'] = resampled.index.astype(np.int64) // 10**6
+            frames[tf_str] = resampled
 
-                ai_results = []
-                valid_indices = frames['5m'].index[500:] 
-                
-                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 
+        ai_results = []
+        valid_indices = frames['5m'].index[500:] 
+        
+        # 4. Scanning Loop
+        for t_idx in valid_indices:
+            current_timestamp = int(t_idx.timestamp() * 1000)
+            
+            # 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 
 
-                    if len(ohlcv_data['1h']) < 60: continue
-                    
-                    current_price = frames['5m'].loc[t_idx]['close']
+            if len(ohlcv_data['1h']) < 60: continue
+            
+            current_price = frames['5m'].loc[t_idx]['close']
 
-                    # 1. Logic Tree Check
-                    logic_packet = {
-                        'symbol': sym,
-                        'ohlcv_1h': ohlcv_data['1h'][-60:],   
-                        'ohlcv_15m': ohlcv_data['15m'][-60:], 
-                        'change_24h': 0.0 
-                    }
-                    
-                    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
+            # Logic Check
+            logic_packet = {
+                'symbol': sym,
+                'ohlcv_1h': ohlcv_data['1h'][-60:],   
+                'ohlcv_15m': ohlcv_data['15m'][-60:], 
+                'change_24h': 0.0 
+            }
+            
+            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
 
-                    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
+            logic_result = self.dm._apply_logic_tree(logic_packet)
+            signal_type = logic_result.get('type', 'NONE')
+            l1_score = logic_result.get('score', 0.0)
+            
+            real_titan = 0.5
+            if signal_type in ['BREAKOUT', 'REVERSAL']:
+                raw_data_for_proc = {
+                    'symbol': sym,
+                    'ohlcv': ohlcv_data, 
+                    'current_price': current_price
+                }
+                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)
+                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.")
+                ai_results.append({
+                    'timestamp': current_timestamp,
+                    'symbol': sym,
+                    'close': current_price,
+                    'real_titan': real_titan, 
+                    'signal_type': signal_type,
+                    'l1_score': l1_score
+                })
+        
+        # Save & Clear
+        if ai_results:
+            pd.DataFrame(ai_results).to_pickle(scores_file)
+            print(f" ✅ Saved ({len(ai_results)}).")
+        else:
+            print(" ⚠️ No signals.")
+            
+        return True
 
-            except Exception as e:
-                print(f" ❌ Error: {e}")
+    # ==============================================================
+    # PHASE 1: Main Loop
+    # ==============================================================
+    async def generate_truth_data(self):
+        print(f"\n🚜 [Phase 1] Processing Logic Tree + Titan ({self.BACKTEST_DAYS} Days)...")
+        end_time_ms = int(time.time() * 1000)
+        start_time_ms = end_time_ms - (self.BACKTEST_DAYS * 24 * 60 * 60 * 1000) 
+        
+        for sym in self.TARGET_COINS:
+            # 🔥 استدعاء الدالة المعزولة
+            await self._process_single_coin_task(sym, start_time_ms, end_time_ms)
             
-            finally:
-                # 🧹 CLEANUP MEMORY FORCEFULLY 🧹
-                # هذا الجزء يضمن عدم تراكم البيانات في الرام
-                del all_candles_1m
-                del df_1m
-                del frames
-                # استدعاء جامع القمامة يدوياً
-                gc.collect()
+            # 🔥 تنظيف الذاكرة الإجباري بعد كل عملة
+            gc.collect()
+            await asyncio.sleep(0.1) # استراحة قصيرة للمعالج
 
     # ==============================================================
     # PHASE 2: Portfolio Digital Twin Engine
@@ -208,8 +212,9 @@ class HeavyDutyBacktester:
     @staticmethod
     def _worker_optimize(combinations_batch, scores_files, initial_capital, fees_pct, max_slots):
         results = []
-        
         all_data = []
+        
+        # Load data safely
         for fp in scores_files:
             try: 
                 df = pd.read_pickle(fp)
@@ -288,11 +293,10 @@ class HeavyDutyBacktester:
                 
                 if wallet["balance"] < 1.0 and len(wallet["positions"]) == 0: break
 
-            # Detailed Analytics
+            # Analytics
             trades = wallet["trades_history"]
             if trades:
                 net_profit = wallet["balance"] - initial_capital
-                
                 pnls = [t['pnl'] for t in trades]
                 wins = [p for p in pnls if p > 0]
                 losses = [p for p in pnls if p <= 0]
@@ -300,7 +304,7 @@ class HeavyDutyBacktester:
                 win_count = len(wins)
                 loss_count = len(losses)
                 total_trades = len(trades)
-                win_rate = (win_count / total_trades) * 100
+                win_rate = (win_count / total_trades) * 100 if total_trades > 0 else 0
                 
                 max_single_win = max(pnls) if pnls else 0.0
                 max_single_loss = min(pnls) if pnls else 0.0