Update backtest_engine.py

backtest_engine.py

@@ -1,9 +1,5 @@
 # ============================================================
-# 🧪 backtest_engine.py (V88.0 - GEM-Architect: RAM-Burst Edition)
-# ============================================================
-# استراتيجية المعماري للمواصفات المحدودة (2 vCPU / 16GB RAM):
-# 1. Async I/O Burst: سحب البيانات بالتوازي لأن الشبكة لا تضغط المعالج.
-# 2. In-Memory Analysis: المعالجة تتم بعد اكتمال البيانات بالكامل.
+# 🧪 backtest_engine.py (V88.1 - GEM-Architect: RAM-Burst Fix)
 # ============================================================
 
 import asyncio
@@ -24,7 +20,7 @@ try:
     from ml_engine.data_manager import DataManager
     from learning_hub.adaptive_hub import StrategyDNA, AdaptiveHub
     from r2 import R2Service
-    import ccxt.async_support as ccxt # نستخدم النسخة الـ Async حصراً
+    import ccxt.async_support as ccxt 
 except ImportError:
     pass
 
@@ -44,7 +40,7 @@ class HeavyDutyBacktester:
         self.force_end_date = None   
         
         if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
-        print(f"🧪 [Backtest V88.0] RAM-Burst Edition (High Speed I/O).")
+        print(f"🧪 [Backtest V88.1] RAM-Burst Edition (Fix Applied).")
 
     def set_date_range(self, start_str, end_str):
         self.force_start_date = start_str
@@ -58,44 +54,30 @@ class HeavyDutyBacktester:
     # ⚡ FAST DATA DOWNLOADER (Async Burst)
     # ==============================================================
     async def _fetch_all_data_fast(self, sym, start_ms, end_ms):
-        """
-        يقوم بتحميل كل البيانات دفعة واحدة باستخدام اتصالات متزامنة.
-        يستغل الرام (16GB) لتخزين كل شيء قبل المعالجة.
-        """
         print(f"   ⚡ [Network] Burst-Downloading {sym} ({start_ms} -> {end_ms})...", flush=True)
         
-        # تقسيم الفترة إلى دفعات (كل دفعة 1000 شمعة = 60000000 ميلي ثانية)
         limit = 1000
         duration_per_batch = limit * 60 * 1000
         
         tasks = []
         current = start_ms
-        
-        # إنشاء قائمة بالمهمات الزمنية
         while current < end_ms:
             tasks.append(current)
             current += duration_per_batch
             
         all_candles = []
         total_batches = len(tasks)
-        
-        # نستخدم Semaphore لمنع حظر الـ IP (مثلاً 10 اتصالات في نفس اللحظة)
         sem = asyncio.Semaphore(10) 
 
         async def _fetch_batch(timestamp):
             async with sem:
-                try:
-                    # محاولة 3 مرات في حال الفشل
-                    for _ in range(3):
-                        try:
-                            return await self.dm.exchange.fetch_ohlcv(sym, '1m', since=timestamp, limit=limit)
-                        except Exception:
-                            await asyncio.sleep(1)
-                    return []
-                except: return []
-
-        # تشغيل التنزيل المتوازي
-        # نقسم المهام إلى مجموعات (Chunks) لنظهر التقدم
+                for _ in range(3):
+                    try:
+                        return await self.dm.exchange.fetch_ohlcv(sym, '1m', since=timestamp, limit=limit)
+                    except Exception:
+                        await asyncio.sleep(1)
+                return []
+
         chunk_size = 20
         for i in range(0, len(tasks), chunk_size):
             chunk_tasks = tasks[i:i + chunk_size]
@@ -105,16 +87,12 @@ class HeavyDutyBacktester:
             for res in results:
                 if res: all_candles.extend(res)
             
-            # طباعة التقدم
             progress = min(100, int((i + chunk_size) / total_batches * 100))
             print(f"     📥 Downloaded {progress}%... (Total: {len(all_candles)} candles)", flush=True)
 
-        # ترتيب وإزالة التكرار
         if not all_candles: return None
         
-        # تصفية ما هو خارج النطاق بدقة
         filtered = [c for c in all_candles if c[0] >= start_ms and c[0] <= end_ms]
-        # إزالة التكرارات بناءً على الوقت (المفتاح 0)
         seen = set()
         unique_candles = []
         for c in filtered:
@@ -122,7 +100,6 @@ class HeavyDutyBacktester:
                 unique_candles.append(c)
                 seen.add(c[0])
         
-        # ترتيب نهائي
         unique_candles.sort(key=lambda x: x[0])
         return unique_candles
 
@@ -131,15 +108,17 @@ class HeavyDutyBacktester:
     # ==============================================================
     async def _process_data_in_memory(self, sym, candles, start_ms, end_ms):
         safe_sym = sym.replace('/', '_')
-        period_suffix = f"{start_time_ms}_{end_time_ms}" # سيتم تعريفه لاحقاً
-        # لكن هن�� سنستخدم معرف الفترة الممرر
+        # ✅ FIX: Use passed arguments directly
         period_suffix = f"{start_ms}_{end_ms}"
         scores_file = f"{CACHE_DIR}/{safe_sym}_{period_suffix}_scores.pkl"
 
+        if os.path.exists(scores_file):
+             print(f"   📂 [{sym}] Data Exists -> Skipping.")
+             return
+
         print(f"   ⚙️ [CPU] Processing {len(candles)} candles from RAM...", flush=True)
         t0 = time.time()
 
-        # تحويل سريع لـ Pandas
         df_1m = pd.DataFrame(candles, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
         cols = ['open', 'high', 'low', 'close', 'volume']
         df_1m[cols] = df_1m[cols].astype('float32')
@@ -147,7 +126,6 @@ class HeavyDutyBacktester:
         df_1m.set_index('datetime', inplace=True)
         df_1m = df_1m.sort_index()
 
-        # Resampling
         frames = {}
         agg_dict = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
         frames['1m'] = df_1m.copy()
@@ -159,14 +137,12 @@ class HeavyDutyBacktester:
 
         ai_results = []
         
-        # نبدأ التحليل بعد فترة كافية للمؤشرات
         start_analysis_dt = df_1m.index[0] + pd.Timedelta(minutes=500)
         valid_indices = frames['5m'].loc[start_analysis_dt:].index 
         
         total_steps = len(valid_indices)
         step_count = 0
         
-        # حلقة المعالجة السريعة (بدون انتظار شبكة)
         for t_idx in valid_indices:
             step_count += 1
             if step_count % 2000 == 0:
@@ -175,7 +151,6 @@ class HeavyDutyBacktester:
 
             ohlcv_data = {}
             try:
-                # Slicing from RAM is fast
                 cutoff = t_idx
                 ohlcv_data['1m']  = self.df_to_list(frames['1m'].loc[:cutoff].tail(500))
                 ohlcv_data['5m']  = self.df_to_list(frames['5m'].loc[:cutoff].tail(200))
@@ -188,7 +163,6 @@ class HeavyDutyBacktester:
             if len(ohlcv_data['1h']) < 60: continue
             current_price = frames['5m'].loc[t_idx]['close']
 
-            # L1 Logic
             logic_packet = {
                 'symbol': sym,
                 'ohlcv_1h': ohlcv_data['1h'][-60:],   
@@ -206,7 +180,6 @@ class HeavyDutyBacktester:
             signal_type = logic_result.get('type', 'NONE')
             l1_score = logic_result.get('score', 0.0)
             
-            # L2 AI Execution (Only on L1 Signals)
             real_titan = 0.5
             if signal_type in ['BREAKOUT', 'REVERSAL']:
                 raw_data_for_proc = {'symbol': sym, 'ohlcv': ohlcv_data, 'current_price': current_price}
@@ -248,14 +221,6 @@ class HeavyDutyBacktester:
             return 
         
         for sym in self.TARGET_COINS:
-            safe_sym = sym.replace('/', '_')
-            period_suffix = f"{start_time_ms}_{end_time_ms}"
-            scores_file = f"{CACHE_DIR}/{safe_sym}_{period_suffix}_scores.pkl"
-            
-            if os.path.exists(scores_file):
-                print(f"   📂 [{sym}] Data Exists -> Skipping.")
-                continue
-
             # 1. Download Phase (Async Burst)
             candles = await self._fetch_all_data_fast(sym, start_time_ms, end_time_ms)
             
@@ -268,7 +233,7 @@ class HeavyDutyBacktester:
             gc.collect()
 
     # ==============================================================
-    # PHASE 2: Portfolio Digital Twin Engine (Standard)
+    # PHASE 2: Portfolio Digital Twin Engine
     # ==============================================================
     @staticmethod
     def _worker_optimize(combinations_batch, scores_files, initial_capital, fees_pct, max_slots):