Update backtest_engine.py

backtest_engine.py

@@ -1,5 +1,5 @@
 # ============================================================
-# 🧪 backtest_engine.py (V86.1 - GEM-Architect: Debug Turbo)
+# 🧪 backtest_engine.py (V86.3 - GEM-Architect: Stable Parallel)
 # ============================================================
 
 import asyncio
@@ -10,8 +10,6 @@ import logging
 import itertools 
 import os
 import gc 
-import sys
-import traceback
 import concurrent.futures
 from datetime import datetime, timezone
 from typing import Dict, Any, List
@@ -28,52 +26,37 @@ logging.getLogger('ml_engine').setLevel(logging.WARNING)
 CACHE_DIR = "backtest_real_scores"
 
 # ==============================================================================
-# 🚜 ISOLATED WORKER FUNCTION (With Staggered Start & Debug Logs)
+# 🚜 ISOLATED WORKER (Stable & Clean)
 # ==============================================================================
 def run_parallel_chunk(chunk_payload):
     """
-    عامل مستقل مع إقلاع متدرج لتخفيف الضغط على الشبكة والمعالج.
+    عامل مستقل بمعايير ثبات عالية.
     """
     symbol, start_ms, end_ms, chunk_id = chunk_payload
     
-    # 1. الإقلاع المتدرج (Staggered Start)
-    # كل عامل ينتظر (chunk_id * 2) ثانية. هذا يمنع 15 عملية من ضرب الـ API في نفس اللحظة.
-    wait_time = chunk_id * 1.5 
-    if chunk_id > 0:
-        time.sleep(wait_time)
+    # تأخير بسيط جداً عند الإقلاع لتخفيف صدمة المعالج
+    time.sleep(chunk_id * 1.0)
     
-    print(f"   ⚡ [Core {chunk_id}] Init... (Waited {wait_time}s)", flush=True)
+    print(f"   ⚡ [Core {chunk_id}] Initializing ML Engine...", flush=True)
 
     try:
-        # 2. تهيئة بيئة خفيفة
-        # نمرر None لتجنب الاتصالات الثقيلة غير الضرورية، لكن DataManager يحتاج لتهيئة
+        # تهيئة بيئة نظيفة
         local_dm = DataManager(None, None, None)
-        
-        # حيلة: لا نستدعي initialize الكاملة لـ DM إذا كانت تتصل بالإنترنت لجلب الأسواق
-        # فقط نهيئ http_client إذا لزم الأمر يدوياً داخل المهمة
-        
-        # تهيئة المعالج
         local_proc = MLProcessor(local_dm)
         
-        # تشغيل حلقة الأحداث الخاصة بهذه العملية
         loop = asyncio.new_event_loop()
         asyncio.set_event_loop(loop)
         
-        # تشغيل التهيئة
-        # ملاحظة: MLProcessor قد يستغرق وقتاً لتحميل النماذج
+        # تحميل النماذج (هنا يكمن الثقل)
         loop.run_until_complete(local_proc.initialize()) 
-        # local_dm.initialize() عادة يتصل بالمنصة، سنقوم بتهيئته يدوياً داخل المهمة لتجنب التعليق
-        if not local_dm.http_client:
-            import httpx
-            local_dm.http_client = httpx.AsyncClient(timeout=30.0)
+        loop.run_until_complete(local_dm.initialize())
 
-        # إنشاء نسخة محلية من الباكتستر
         local_tester = HeavyDutyBacktester(local_dm, local_proc)
         
         dt_start = datetime.fromtimestamp(start_ms/1000, tz=timezone.utc).strftime('%Y-%m-%d')
-        print(f"   📥 [Core {chunk_id}] Downloading Data: {dt_start}...", flush=True)
+        print(f"   📥 [Core {chunk_id}] Fetching Data from {dt_start}...", flush=True)
         
-        # فترة التحمية
+        # إضافة فترة تحمية للمؤشرات (2000 دقيقة)
         warmup_ms = 2000 * 60 * 1000
         actual_fetch_start = start_ms - warmup_ms
         
@@ -88,16 +71,17 @@ def run_parallel_chunk(chunk_payload):
             )
         )
         
-        # تنظيف
+        # تنظيف الذاكرة فوراً
         loop.run_until_complete(local_dm.close())
         loop.close()
+        del local_dm, local_proc, local_tester
+        gc.collect()
         
-        print(f"   ✅ [Core {chunk_id}] Finished Chunk.", flush=True)
+        print(f"   ✅ [Core {chunk_id}] Completed.", flush=True)
         return (chunk_id, success)
         
     except Exception as e:
-        print(f"   ❌ [Core {chunk_id}] CRASHED: {e}", flush=True)
-        traceback.print_exc() # طباعة تفاصيل الخطأ
+        print(f"   ❌ [Core {chunk_id}] CRASH: {e}", flush=True)
         return (chunk_id, False)
 
 # ==============================================================================
@@ -108,16 +92,12 @@ class HeavyDutyBacktester:
         self.dm = data_manager
         self.proc = processor
         self.GRID_DENSITY = 10 
-        
         self.INITIAL_CAPITAL = 10.0
         self.TRADING_FEES = 0.001 
         self.MAX_SLOTS = 4         
-        
         self.TARGET_COINS = ['SOL/USDT']
-        
         self.force_start_date = None 
         self.force_end_date = None   
-        
         if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
 
     def set_date_range(self, start_str, end_str):
@@ -129,7 +109,7 @@ class HeavyDutyBacktester:
         return df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].values.tolist()
 
     # ==============================================================
-    # 🧱 Core Logic: Single Coin Processor (With Progress Logs)
+    # 🧱 Core Logic: Single Coin Processor (With % Progress)
     # ==============================================================
     async def _process_single_coin_task(self, sym, start_time_ms, end_time_ms, chunk_suffix="", analysis_start_ms=None, worker_id=0):
         safe_sym = sym.replace('/', '_')
@@ -147,24 +127,15 @@ class HeavyDutyBacktester:
         df_1m = None
         frames = {}
         
-        # 1. تنزيل البيانات (الحلقة التي كانت تتوقف)
+        # 1. تنزيل البيانات
         try:
             current_since = start_time_ms
-            req_count = 0
             
             while current_since < end_time_ms:
                 try:
-                    # إضافة مهلة وتكرار في حال الفشل
                     batch = await self.dm.exchange.fetch_ohlcv(sym, '1m', since=current_since, limit=1000)
-                    req_count += 1
-                    
-                    # طباعة تقدم كل 5 طلبات لنعرف أن العملية حية
-                    if req_count % 5 == 0:
-                        print(f"   ⏳ [Core {worker_id}] Fetched batch {req_count}...", flush=True)
-
-                except Exception as net_err:
-                    print(f"   ⚠️ [Core {worker_id}] Network hiccup: {net_err}. Retrying in 5s...", flush=True)
-                    await asyncio.sleep(5)
+                except Exception:
+                    await asyncio.sleep(2)
                     continue
 
                 if not batch: break
@@ -174,21 +145,17 @@ class HeavyDutyBacktester:
                 
                 all_candles_1m.extend(batch)
                 current_since = last_ts + 1
-                
-                # تخفيف الضغط قليلاً على الـ API
-                await asyncio.sleep(0.1) 
-                
+                await asyncio.sleep(0.05)
                 if current_since >= end_time_ms: break
             
             all_candles_1m = [c for c in all_candles_1m if c[0] <= end_time_ms]
             
             if not all_candles_1m: 
-                print(f"   ⚠️ [Core {worker_id}] No candles found.", flush=True)
+                print(f"   ⚠️ [Core {worker_id}] No data found.", flush=True)
                 return False
 
-            print(f"   ⚙️ [Core {worker_id}] Processing {len(all_candles_1m)} candles...", flush=True)
+            # print(f"   ⚙️ [Core {worker_id}] Parsing {len(all_candles_1m)} candles...", flush=True)
 
-            # معالجة البيانات (Pandas)
             df_1m = pd.DataFrame(all_candles_1m, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
             cols = ['open', 'high', 'low', 'close', 'volume']
             df_1m[cols] = df_1m[cols].astype('float32')
@@ -197,9 +164,8 @@ class HeavyDutyBacktester:
             df_1m = df_1m.sort_index()
             
             agg_dict = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
-            df_1m_ready = df_1m.copy()
-            df_1m_ready['timestamp'] = df_1m_ready.index.astype(np.int64) // 10**6
-            frames['1m'] = df_1m_ready
+            frames['1m'] = df_1m.copy()
+            frames['1m']['timestamp'] = frames['1m'].index.astype(np.int64) // 10**6
 
             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()
@@ -211,34 +177,30 @@ class HeavyDutyBacktester:
             analysis_start_dt = pd.to_datetime(analysis_start_ms, unit='ms')
             valid_indices = frames['5m'].loc[analysis_start_dt:].index 
             
-            # حلقة الذكاء الاصطناعي
             total_steps = len(valid_indices)
             step_count = 0
             
+            # حلقة المعالجة
             for t_idx in valid_indices:
                 if t_idx.timestamp() * 1000 > end_time_ms: break
                 step_count += 1
                 
-                # طباعة تقدم المعالجة كل 20%
-                if step_count % max(1, (total_steps // 5)) == 0:
-                     print(f"   🧠 [Core {worker_id}] AI Analyzing: {step_count}/{total_steps}...", flush=True)
+                # طباعة نسبة التقدم كل 10%
+                if total_steps > 0 and step_count % max(1, int(total_steps * 0.1)) == 0:
+                     pct = int((step_count / total_steps) * 100)
+                     print(f"   🧠 [Core {worker_id}] Progress: {pct}%", flush=True)
 
                 current_timestamp = int(t_idx.timestamp() * 1000)
                 ohlcv_data = {}
                 try:
-                    current_slice_1m = frames['1m'].loc[:t_idx]
-                    current_slice_5m = frames['5m'].loc[:t_idx]
-                    current_slice_15m = frames['15m'].loc[:t_idx]
-                    current_slice_1h = frames['1h'].loc[:t_idx]
-                    current_slice_4h = frames['4h'].loc[:t_idx]
-                    current_slice_1d = frames['1d'].loc[:t_idx]
-
-                    ohlcv_data['1m']  = self.df_to_list(current_slice_1m.tail(500))
-                    ohlcv_data['5m']  = self.df_to_list(current_slice_5m.tail(200))
-                    ohlcv_data['15m'] = self.df_to_list(current_slice_15m.tail(200))
-                    ohlcv_data['1h']  = self.df_to_list(current_slice_1h.tail(200))
-                    ohlcv_data['4h']  = self.df_to_list(current_slice_4h.tail(100))
-                    ohlcv_data['1d']  = self.df_to_list(current_slice_1d.tail(50))
+                    # استخراج البيانات باستخدام loc (أسرع وأدق)
+                    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))
+                    ohlcv_data['15m'] = self.df_to_list(frames['15m'].loc[:cutoff].tail(200))
+                    ohlcv_data['1h']  = self.df_to_list(frames['1h'].loc[:cutoff].tail(200))
+                    ohlcv_data['4h']  = self.df_to_list(frames['4h'].loc[:cutoff].tail(100))
+                    ohlcv_data['1d']  = self.df_to_list(frames['1d'].loc[:cutoff].tail(50))
                 except: continue 
 
                 if len(ohlcv_data['1h']) < 60: continue
@@ -250,13 +212,7 @@ class HeavyDutyBacktester:
                     '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)
@@ -281,13 +237,14 @@ class HeavyDutyBacktester:
             dt = time.time() - t0
             if ai_results:
                 pd.DataFrame(ai_results).to_pickle(scores_file)
-                print(f"   💾 [Core {worker_id}] Saved {len(ai_results)} signals. (Time: {dt:.1f}s)", flush=True)
+                print(f"   💾 [Core {worker_id}] Saved {len(ai_results)} signals. ({dt:.1f}s)", flush=True)
+            else:
+                print(f"   ⚠️ [Core {worker_id}] No signals found.", flush=True)
             
             return True
 
         except Exception as e:
-            print(f"   ❌ [Core {worker_id}] CRASH: {e}", flush=True)
-            traceback.print_exc()
+            print(f"   ❌ [Core {worker_id}] ERR: {e}", flush=True)
             return False
         
         finally:
@@ -297,7 +254,7 @@ class HeavyDutyBacktester:
             gc.collect()
 
     # ==============================================================
-    # PHASE 1: Main Loop
+    # PHASE 1: Main Loop (Restricted Concurrency)
     # ==============================================================
     async def generate_truth_data(self):
         if self.force_start_date and self.force_end_date:
@@ -306,13 +263,13 @@ class HeavyDutyBacktester:
             start_time_ms = int(dt_start.timestamp() * 1000)
             end_time_ms = int(dt_end.timestamp() * 1000)
             print(f"\n🚜 [Phase 1] Processing Era: {self.force_start_date} -> {self.force_end_date}")
-            print(f"   🚀 Turbo Mode: Engaging all CPU cores (Staggered Start)...")
+            print(f"   🚀 Turbo Mode: Safe Parallel Execution (Max 4 Cores)...")
         else:
             return 
         
-        cpu_count = os.cpu_count() or 4
-        # نستخدم عدد عمال معقول لتجنب حظر الـ API (مثلاً 10 عمال كحد أقصى)
-        workers_count = min(10, max(1, cpu_count - 1))
+        # ⚠️ تقييد عدد العمال لتجنب تجميد الجهاز بسبب نماذج الذكاء الاصطناعي
+        # 4 عمال هو حد آمن لمعظم الأجهزة
+        workers_count = 4 
         
         total_duration = end_time_ms - start_time_ms
         chunk_size = total_duration // workers_count
@@ -352,8 +309,7 @@ class HeavyDutyBacktester:
                         df_part = pd.read_pickle(part_file)
                         if not df_part.empty: all_dfs.append(df_part)
                         os.remove(part_file)
-                    except Exception as e:
-                        print(f"   ⚠️ Merge Error (Part {chunk_id}): {e}")
+                    except: pass
 
             if all_dfs:
                 final_df = pd.concat(all_dfs).drop_duplicates(subset=['timestamp']).sort_values('timestamp')
@@ -361,11 +317,10 @@ class HeavyDutyBacktester:
                 print(f"   💾 [{sym}] FINAL SAVE: {len(final_df)} signals.")
             else:
                 print(f"   ⚠️ [{sym}] No signals generated.")
-                
             gc.collect()
 
     # ==============================================================
-    # PHASE 2: Portfolio Digital Twin Engine (Unchanged Logic)
+    # PHASE 2: Portfolio Digital Twin Engine (Unchanged)
     # ==============================================================
     @staticmethod
     def _worker_optimize(combinations_batch, scores_files, initial_capital, fees_pct, max_slots):
@@ -388,7 +343,6 @@ class HeavyDutyBacktester:
             for ts, group in grouped_by_time:
                 active_symbols = list(wallet["positions"].keys())
                 current_prices = {row['symbol']: row['close'] for _, row in group.iterrows()}
-                # Exits
                 for sym in active_symbols:
                     if sym in current_prices:
                         curr_p = current_prices[sym]
@@ -403,7 +357,7 @@ class HeavyDutyBacktester:
                             wallet["balance"] += net_pnl
                             del wallet["positions"][sym]
                             wallet["trades_history"].append({'pnl': net_pnl})
-                # Entries
+                
                 current_total_equity = wallet["balance"] + wallet["allocated"] 
                 if current_total_equity > peak_balance: peak_balance = current_total_equity
                 dd = (peak_balance - current_total_equity) / peak_balance
@@ -442,7 +396,6 @@ class HeavyDutyBacktester:
                 win_count = len([p for p in pnls if p > 0]); loss_count = len([p for p in pnls if p <= 0])
                 win_rate = (win_count / len(trades)) * 100
                 max_single_win = max(pnls) if pnls else 0.0; max_single_loss = min(pnls) if pnls else 0.0
-                
                 current_win_streak = 0; max_win_streak = 0
                 current_loss_streak = 0; max_loss_streak = 0
                 for p in pnls:
@@ -452,7 +405,6 @@ class HeavyDutyBacktester:
                     else:
                         current_loss_streak += 1; current_win_streak = 0
                         if current_loss_streak > max_loss_streak: max_loss_streak = current_loss_streak
-
                 results.append({
                     'config': config, 'final_balance': wallet["balance"] + wallet["allocated"],
                     'net_profit': net_profit, 'total_trades': len(trades),
@@ -520,7 +472,7 @@ class HeavyDutyBacktester:
         return best['config'], best
 
 async def run_strategic_optimization_task():
-    print("\n🧪 [STRATEGIC BACKTEST] Time Lord Initiated (Parallel Turbo)...")
+    print("\n🧪 [STRATEGIC BACKTEST] Time Lord Initiated (Stable Parallel)...")
     r2 = R2Service()
     dm = DataManager(None, None, r2)
     proc = MLProcessor(dm)