Update backtest_engine.py

backtest_engine.py

@@ -1,5 +1,9 @@
 # ============================================================
-# 🧪 backtest_engine.py (V86.3 - GEM-Architect: Stable Parallel)
+# 🧪 backtest_engine.py (V88.0 - GEM-Architect: RAM-Burst Edition)
+# ============================================================
+# استراتيجية المعماري للمواصفات المحدودة (2 vCPU / 16GB RAM):
+# 1. Async I/O Burst: سحب البيانات بالتوازي لأن الشبكة لا تضغط المعالج.
+# 2. In-Memory Analysis: المعالجة تتم بعد اكتمال البيانات بالكامل.
 # ============================================================
 
 import asyncio
@@ -10,7 +14,8 @@ import logging
 import itertools 
 import os
 import gc 
-import concurrent.futures
+import sys
+import traceback
 from datetime import datetime, timezone
 from typing import Dict, Any, List
 
@@ -19,74 +24,13 @@ 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 حصراً
 except ImportError:
     pass
 
 logging.getLogger('ml_engine').setLevel(logging.WARNING)
 CACHE_DIR = "backtest_real_scores"
 
-# ==============================================================================
-# 🚜 ISOLATED WORKER (Stable & Clean)
-# ==============================================================================
-def run_parallel_chunk(chunk_payload):
-    """
-    عامل مستقل بمعايير ثبات عالية.
-    """
-    symbol, start_ms, end_ms, chunk_id = chunk_payload
-    
-    # تأخير بسيط جداً عند الإقلاع لتخفيف صدمة المعالج
-    time.sleep(chunk_id * 1.0)
-    
-    print(f"   ⚡ [Core {chunk_id}] Initializing ML Engine...", flush=True)
-
-    try:
-        # تهيئة بيئة نظيفة
-        local_dm = DataManager(None, None, None)
-        local_proc = MLProcessor(local_dm)
-        
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-        
-        # تحميل النماذج (هنا يكمن الثقل)
-        loop.run_until_complete(local_proc.initialize()) 
-        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}] Fetching Data from {dt_start}...", flush=True)
-        
-        # إضافة فترة تحمية للمؤشرات (2000 دقيقة)
-        warmup_ms = 2000 * 60 * 1000
-        actual_fetch_start = start_ms - warmup_ms
-        
-        success = loop.run_until_complete(
-            local_tester._process_single_coin_task(
-                symbol, 
-                actual_fetch_start,  
-                end_ms, 
-                chunk_suffix=f"_part{chunk_id}",
-                analysis_start_ms=start_ms,
-                worker_id=chunk_id 
-            )
-        )
-        
-        # تنظيف الذاكرة فوراً
-        loop.run_until_complete(local_dm.close())
-        loop.close()
-        del local_dm, local_proc, local_tester
-        gc.collect()
-        
-        print(f"   ✅ [Core {chunk_id}] Completed.", flush=True)
-        return (chunk_id, success)
-        
-    except Exception as e:
-        print(f"   ❌ [Core {chunk_id}] CRASH: {e}", flush=True)
-        return (chunk_id, False)
-
-# ==============================================================================
-# 🧠 Main Class
-# ==============================================================================
 class HeavyDutyBacktester:
     def __init__(self, data_manager, processor):
         self.dm = data_manager
@@ -98,7 +42,9 @@ class HeavyDutyBacktester:
         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)
+        print(f"🧪 [Backtest V88.0] RAM-Burst Edition (High Speed I/O).")
 
     def set_date_range(self, start_str, end_str):
         self.force_start_date = start_str
@@ -109,152 +55,187 @@ class HeavyDutyBacktester:
         return df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].values.tolist()
 
     # ==============================================================
-    # 🧱 Core Logic: Single Coin Processor (With % Progress)
+    # ⚡ FAST DATA DOWNLOADER (Async Burst)
     # ==============================================================
-    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('/', '_')
-        if analysis_start_ms is None: analysis_start_ms = start_time_ms
+    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)
         
-        period_suffix = f"{analysis_start_ms}_{end_time_ms}{chunk_suffix}"
-        scores_file = f"{CACHE_DIR}/{safe_sym}_{period_suffix}_scores.pkl"
+        # تقسيم الفترة إلى دفعات (كل دفعة 1000 شمعة = 60000000 ميلي ثانية)
+        limit = 1000
+        duration_per_batch = limit * 60 * 1000
         
-        if os.path.exists(scores_file):
-            print(f"   📂 [Core {worker_id}] File Exists -> Skipping.", flush=True)
-            return True
-
-        t0 = time.time()
-        all_candles_1m = []
-        df_1m = None
-        frames = {}
+        tasks = []
+        current = start_ms
         
-        # 1. تنزيل البيانات
-        try:
-            current_since = start_time_ms
+        # إنشاء قائمة بالمهمات الزمنية
+        while current < end_ms:
+            tasks.append(current)
+            current += duration_per_batch
             
-            while current_since < end_time_ms:
+        all_candles = []
+        total_batches = len(tasks)
+        
+        # نستخدم Semaphore لمنع حظر الـ IP (مثلاً 10 اتصالات في نفس اللحظة)
+        sem = asyncio.Semaphore(10) 
+
+        async def _fetch_batch(timestamp):
+            async with sem:
                 try:
-                    batch = await self.dm.exchange.fetch_ohlcv(sym, '1m', since=current_since, limit=1000)
-                except Exception:
-                    await asyncio.sleep(2)
-                    continue
-
-                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.05)
-                if current_since >= end_time_ms: break
+                    # محاولة 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) لنظهر التقدم
+        chunk_size = 20
+        for i in range(0, len(tasks), chunk_size):
+            chunk_tasks = tasks[i:i + chunk_size]
+            futures = [_fetch_batch(ts) for ts in chunk_tasks]
+            results = await asyncio.gather(*futures)
             
-            all_candles_1m = [c for c in all_candles_1m if c[0] <= end_time_ms]
+            for res in results:
+                if res: all_candles.extend(res)
             
-            if not all_candles_1m: 
-                print(f"   ⚠️ [Core {worker_id}] No data found.", flush=True)
-                return False
-
-            # print(f"   ⚙️ [Core {worker_id}] Parsing {len(all_candles_1m)} candles...", flush=True)
-
-            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')
-            df_1m['datetime'] = pd.to_datetime(df_1m['timestamp'], unit='ms')
-            df_1m.set_index('datetime', inplace=True)
-            df_1m = df_1m.sort_index()
-            
-            agg_dict = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
-            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()
-                resampled[cols] = resampled[cols].astype('float32')
-                resampled['timestamp'] = resampled.index.astype(np.int64) // 10**6
-                frames[tf_str] = resampled
-
-            ai_results = []
-            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
+            # طباعة التقدم
+            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:
+            if c[0] not in seen:
+                unique_candles.append(c)
+                seen.add(c[0])
+        
+        # ترتيب نهائي
+        unique_candles.sort(key=lambda x: x[0])
+        return unique_candles
+
+    # ==============================================================
+    # 🧠 CPU PROCESSING (In-Memory)
+    # ==============================================================
+    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}" # سيتم تعريفه لاحقاً
+        # لكن هنا سنستخدم معرف الفترة الممرر
+        period_suffix = f"{start_ms}_{end_ms}"
+        scores_file = f"{CACHE_DIR}/{safe_sym}_{period_suffix}_scores.pkl"
+
+        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')
+        df_1m['datetime'] = pd.to_datetime(df_1m['timestamp'], unit='ms')
+        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()
+        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')]:
+            frames[tf_str] = df_1m.resample(tf_code).agg(agg_dict).dropna()
+            frames[tf_str]['timestamp'] = frames[tf_str].index.astype(np.int64) // 10**6
+
+        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:
+                 pct = int((step_count / total_steps) * 100)
+                 print(f"     🧠 AI Analysis: {pct}%...", flush=True)
+
+            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))
+                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
+            current_price = frames['5m'].loc[t_idx]['close']
+
+            # L1 Logic
+            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)
             
-            # حلقة المعالجة
-            for t_idx in valid_indices:
-                if t_idx.timestamp() * 1000 > end_time_ms: break
-                step_count += 1
-                
-                # طباعة نسبة التقدم كل 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 = {}
+            # 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}
                 try:
-                    # استخراج البيانات باستخدام 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
-                current_price = frames['5m'].loc[t_idx]['close']
-
-                logic_packet = {
-                    'symbol': sym,
-                    'ohlcv_1h': ohlcv_data['1h'][-60:],   
-                    'ohlcv_15m': ohlcv_data['15m'][-60:], 
-                    'change_24h': 0.0 
-                }
-                
-                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
-                    })
-            
-            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. ({dt:.1f}s)", flush=True)
-            else:
-                print(f"   ⚠️ [Core {worker_id}] No signals found.", flush=True)
-            
-            return True
+                    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
 
-        except Exception as e:
-            print(f"   ❌ [Core {worker_id}] ERR: {e}", flush=True)
-            return False
+                ai_results.append({
+                    'timestamp': int(t_idx.timestamp() * 1000),
+                    'symbol': sym,
+                    'close': current_price,
+                    'real_titan': real_titan, 
+                    'signal_type': signal_type,
+                    'l1_score': l1_score
+                })
         
-        finally:
-            del all_candles_1m
-            if df_1m is not None: del df_1m
-            del frames
-            gc.collect()
+        dt = time.time() - t0
+        if ai_results:
+            pd.DataFrame(ai_results).to_pickle(scores_file)
+            print(f"   💾 [{sym}] Saved {len(ai_results)} signals. (Compute Time: {dt:.1f}s)")
+        else:
+            print(f"   ⚠️ [{sym}] No signals found.")
+            
+        del frames, df_1m, candles
+        gc.collect()
 
     # ==============================================================
-    # PHASE 1: Main Loop (Restricted Concurrency)
+    # PHASE 1: Main Loop
     # ==============================================================
     async def generate_truth_data(self):
         if self.force_start_date and self.force_end_date:
@@ -262,65 +243,32 @@ class HeavyDutyBacktester:
             dt_end = datetime.strptime(self.force_end_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
             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: Safe Parallel Execution (Max 4 Cores)...")
+            print(f"\n🚜 [Phase 1] Era: {self.force_start_date} -> {self.force_end_date}")
         else:
             return 
         
-        # ⚠️ تقييد عدد العمال لتجنب تجميد الجهاز بسبب نماذج الذكاء الاصطناعي
-        # 4 عمال هو حد آمن لمعظم الأجهزة
-        workers_count = 4 
-        
-        total_duration = end_time_ms - start_time_ms
-        chunk_size = total_duration // workers_count
-        
         for sym in self.TARGET_COINS:
             safe_sym = sym.replace('/', '_')
-            final_full_file = f"{CACHE_DIR}/{safe_sym}_{start_time_ms}_{end_time_ms}_scores.pkl"
+            period_suffix = f"{start_time_ms}_{end_time_ms}"
+            scores_file = f"{CACHE_DIR}/{safe_sym}_{period_suffix}_scores.pkl"
             
-            if os.path.exists(final_full_file):
-                print(f"   📂 [{sym}] Full Data Exists -> Skipping.")
+            if os.path.exists(scores_file):
+                print(f"   📂 [{sym}] Data Exists -> Skipping.")
                 continue
 
-            tasks_payload = []
-            for i in range(workers_count):
-                c_start = start_time_ms + (i * chunk_size)
-                c_end = start_time_ms + ((i + 1) * chunk_size)
-                if i == workers_count - 1: c_end = end_time_ms
-                tasks_payload.append((sym, c_start, c_end, i))
-
-            print(f"   ⚡ Splitting {sym} into {workers_count} chunks...")
+            # 1. Download Phase (Async Burst)
+            candles = await self._fetch_all_data_fast(sym, start_time_ms, end_time_ms)
             
-            loop = asyncio.get_running_loop()
-            with concurrent.futures.ProcessPoolExecutor(max_workers=workers_count) as executor:
-                futures = [loop.run_in_executor(executor, run_parallel_chunk, task) for task in tasks_payload]
-                results = await asyncio.gather(*futures)
-
-            print(f"   🧩 Merging results for {sym}...")
-            all_dfs = []
-            for chunk_id, success in results:
-                if not success: continue
-                task = tasks_payload[chunk_id]
-                part_start = task[1]; part_end = task[2]
-                part_file = f"{CACHE_DIR}/{safe_sym}_{part_start}_{part_end}_part{chunk_id}_scores.pkl"
-                
-                if os.path.exists(part_file):
-                    try:
-                        df_part = pd.read_pickle(part_file)
-                        if not df_part.empty: all_dfs.append(df_part)
-                        os.remove(part_file)
-                    except: pass
-
-            if all_dfs:
-                final_df = pd.concat(all_dfs).drop_duplicates(subset=['timestamp']).sort_values('timestamp')
-                final_df.to_pickle(final_full_file)
-                print(f"   💾 [{sym}] FINAL SAVE: {len(final_df)} signals.")
+            if candles:
+                # 2. Processing Phase (Sequential CPU)
+                await self._process_data_in_memory(sym, candles, start_time_ms, end_time_ms)
             else:
-                print(f"   ⚠️ [{sym}] No signals generated.")
+                print(f"   ❌ Failed to download data for {sym}")
+
             gc.collect()
 
     # ==============================================================
-    # PHASE 2: Portfolio Digital Twin Engine (Unchanged)
+    # PHASE 2: Portfolio Digital Twin Engine (Standard)
     # ==============================================================
     @staticmethod
     def _worker_optimize(combinations_batch, scores_files, initial_capital, fees_pct, max_slots):
@@ -343,6 +291,7 @@ 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]
@@ -357,7 +306,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
@@ -431,7 +380,7 @@ class HeavyDutyBacktester:
         
         current_period_files = []
         for f in os.listdir(CACHE_DIR):
-            if f.endswith('_scores.pkl') and period_id in f and "_part" not in f:
+            if f.endswith('_scores.pkl') and period_id in f:
                 current_period_files.append(os.path.join(CACHE_DIR, f))
 
         if not current_period_files:
@@ -449,16 +398,13 @@ class HeavyDutyBacktester:
                 combinations.append({'w_titan': round(wt, 2), 'w_struct': round(ws, 2), 'thresh': round(th, 2)})
             
         final_results = []
-        batch_size = max(20, len(combinations) // (os.cpu_count() * 2))
-        batches = [combinations[i:i+batch_size] for i in range(0, len(combinations), batch_size)]
+        batch_size = 100 
         
-        with concurrent.futures.ProcessPoolExecutor() as executor:
-            futures = [executor.submit(self._worker_optimize, batch, current_period_files, 
-                                       self.INITIAL_CAPITAL, self.TRADING_FEES, self.MAX_SLOTS) 
-                       for batch in batches]
-            for future in concurrent.futures.as_completed(futures):
-                try: final_results.extend(future.result())
-                except Exception as e: print(f"Grid Error: {e}")
+        for i in range(0, len(combinations), batch_size):
+            batch = combinations[i:i+batch_size]
+            res = self._worker_optimize(batch, current_period_files, self.INITIAL_CAPITAL, self.TRADING_FEES, self.MAX_SLOTS)
+            final_results.extend(res)
+            if i % 1000 == 0: print(f"     ...Analyzed {i}/{len(combinations)} configs", flush=True)
 
         if not final_results: return None, None
         best = sorted(final_results, key=lambda x: x['final_balance'], reverse=True)[0]
@@ -472,10 +418,14 @@ class HeavyDutyBacktester:
         return best['config'], best
 
 async def run_strategic_optimization_task():
-    print("\n🧪 [STRATEGIC BACKTEST] Time Lord Initiated (Stable Parallel)...")
+    print("\n🧪 [STRATEGIC BACKTEST] RAM-Burst Mode Initiated...")
     r2 = R2Service()
     dm = DataManager(None, None, r2)
     proc = MLProcessor(dm)
+    
+    await dm.initialize()
+    await proc.initialize()
+    
     try:
         hub = AdaptiveHub(r2)
         await hub.initialize()
@@ -499,6 +449,4 @@ async def run_strategic_optimization_task():
         await dm.close()
 
 if __name__ == "__main__":
-    import multiprocessing
-    multiprocessing.freeze_support()
     asyncio.run(run_strategic_optimization_task())