Update backtest_engine.py

backtest_engine.py

@@ -1,5 +1,5 @@
 # ============================================================
-# 🧪 backtest_engine.py (V81.1 - GEM-Architect: Import Fixed)
+# 🧪 backtest_engine.py (V82.0 - GEM-Architect: Verbose Logger)
 # ============================================================
 
 import asyncio
@@ -17,7 +17,6 @@ from typing import Dict, Any, List
 try:
     from ml_engine.processor import MLProcessor, SystemLimits 
     from ml_engine.data_manager import DataManager
-    # ✅ التصحيح هنا: إضافة AdaptiveHub للاستيراد
     from learning_hub.adaptive_hub import StrategyDNA, AdaptiveHub
     from r2 import R2Service
 except ImportError:
@@ -53,7 +52,7 @@ class HeavyDutyBacktester:
         self.force_end_date = None   
         
         if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
-        print(f"🧪 [Backtest V81.1] Time Lord Mode Ready (Import Fixed).")
+        print(f"🧪 [Backtest V82.0] Verbose Logger Mode Ready.")
 
     def set_date_range(self, start_str, end_str):
         self.force_start_date = start_str
@@ -63,15 +62,21 @@ class HeavyDutyBacktester:
         if df.empty: return []
         return df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].values.tolist()
 
+    # ==============================================================
+    # 🧱 Core Logic: Single Coin Processor (VERBOSE)
+    # ==============================================================
     async def _process_single_coin_task(self, sym, start_time_ms, end_time_ms):
         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 fetch.")
             return True
 
-        print(f"   ⚙️ Simulating {sym}...", end="", flush=True)
+        print(f"   ⏳ [{sym}] Starting FETCH sequence...", flush=True)
+        t0 = time.time()
         
         all_candles_1m = []
         df_1m = None
@@ -79,6 +84,8 @@ class HeavyDutyBacktester:
         
         try:
             current_since = start_time_ms
+            fetch_count = 0
+            
             while current_since < end_time_ms:
                 try:
                     batch = await asyncio.wait_for(
@@ -86,22 +93,38 @@ class HeavyDutyBacktester:
                         timeout=10.0
                     )
                 except:
+                    print(f"      ⚠️ [{sym}] Net Retry...", flush=True)
                     await asyncio.sleep(1)
                     continue
 
-                if not batch: break
+                if not batch: 
+                    print(f"      ⚠️ [{sym}] No more data from exchange.", flush=True)
+                    break
+                
                 last_ts = batch[-1][0]
                 if last_ts <= current_since: break
+                
                 all_candles_1m.extend(batch)
                 current_since = last_ts + 1
+                
+                fetch_count += 1
+                # ✅ طباعة التقدم كل 5 دفعات (5000 شمعة) لطمأنة المستخدم
+                if fetch_count % 5 == 0:
+                    print(f"      -> [{sym}] Fetched {len(all_candles_1m)} candles...", flush=True)
+                
                 await asyncio.sleep(0.01)
                 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(" No Data.")
+                print(f"   ❌ [{sym}] FAILED: No data retrieved.")
                 return False
 
+            print(f"   ✅ [{sym}] Download Complete ({len(all_candles_1m)} candles). Processing...", flush=True)
+
+            # تحويل البيانات (Vectorization)
             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')
@@ -124,6 +147,7 @@ class HeavyDutyBacktester:
             ai_results = []
             valid_indices = frames['5m'].index[500:] 
             
+            # محاكاة
             for t_idx in valid_indices:
                 current_timestamp = int(t_idx.timestamp() * 1000)
                 
@@ -174,15 +198,17 @@ class HeavyDutyBacktester:
                         'l1_score': l1_score
                     })
             
+            dt = time.time() - t0
             if ai_results:
                 pd.DataFrame(ai_results).to_pickle(scores_file)
-                print(f" ✅ Saved ({len(ai_results)}).")
+                print(f"   💾 [{sym}] Saved {len(ai_results)} signals. (Time: {dt:.1f}s)")
             else:
-                print(" ⚠️ No signals.")
+                print(f"   ⚠️ [{sym}] No signals found. (Time: {dt:.1f}s)")
+            
             return True
 
         except Exception as e:
-            print(f" ❌ Error: {e}")
+            print(f"   ❌ [{sym}] CRASH: {e}")
             return False
         
         finally:
@@ -191,232 +217,10 @@ class HeavyDutyBacktester:
             del frames
             gc.collect()
 
+    # ==============================================================
+    # PHASE 1: Main Loop
+    # ==============================================================
     async def generate_truth_data(self):
         if self.force_start_date and self.force_end_date:
             dt_start = datetime.strptime(self.force_start_date, "%Y-%m-%d").replace(tzinfo=timezone.utc)
-            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}")
-        else:
-            return 
-        
-        chunk_size = 4
-        chunks = [self.TARGET_COINS[i:i + chunk_size] for i in range(0, len(self.TARGET_COINS), chunk_size)]
-        
-        for chunk_idx, chunk in enumerate(chunks):
-            for sym in chunk:
-                try:
-                    await asyncio.wait_for(
-                        self._process_single_coin_task(sym, start_time_ms, end_time_ms),
-                        timeout=300.0 
-                    )
-                except asyncio.TimeoutError:
-                    print(f" 💀 Killed {sym}. Moving on...")
-                    gc.collect()
-            gc.collect()
-            await asyncio.sleep(1.0)
-
-    @staticmethod
-    def _worker_optimize(combinations_batch, scores_files, initial_capital, fees_pct, max_slots):
-        results = []
-        all_data = []
-        
-        for fp in scores_files:
-            try: 
-                df = pd.read_pickle(fp)
-                if not df.empty: all_data.append(df)
-            except: pass
-        
-        if not all_data: return []
-        
-        global_df = pd.concat(all_data)
-        global_df.sort_values('timestamp', inplace=True)
-        grouped_by_time = global_df.groupby('timestamp')
-        
-        for config in combinations_batch:
-            wallet = { "balance": initial_capital, "allocated": 0.0, "positions": {}, "trades_history": [] }
-            
-            w_titan = config['w_titan']
-            w_struct = config['w_struct']
-            entry_thresh = config['thresh']
-            
-            for ts, group in grouped_by_time:
-                active_symbols = list(wallet["positions"].keys())
-                current_prices = {row['symbol']: row['close'] for _, row in group.iterrows()}
-                
-                for sym in active_symbols:
-                    if sym in current_prices:
-                        curr_p = current_prices[sym]
-                        pos = wallet["positions"][sym]
-                        entry_p = pos['entry_price']
-                        pct_change = (curr_p - entry_p) / entry_p
-                        
-                        if pct_change >= 0.03 or pct_change <= -0.02:
-                            gross_pnl = pos['size_usd'] * pct_change
-                            fees = pos['size_usd'] * fees_pct * 2
-                            net_pnl = gross_pnl - fees
-                            wallet["allocated"] -= pos['size_usd']
-                            wallet["balance"] += net_pnl
-                            del wallet["positions"][sym]
-                            wallet["trades_history"].append({'pnl': net_pnl})
-
-                if len(wallet["positions"]) < max_slots:
-                    free_capital = wallet["balance"] - wallet["allocated"]
-                    slots_left = max_slots - len(wallet["positions"])
-                    
-                    if slots_left > 0 and free_capital > 2.0:
-                        position_size = wallet["balance"] / max_slots
-                        if wallet["balance"] < 20.0: position_size = free_capital / slots_left
-                        position_size = min(position_size, free_capital)
-
-                        for _, row in group.iterrows():
-                            sym = row['symbol']
-                            if sym in wallet["positions"]: continue
-                            
-                            sig_type = row['signal_type']
-                            l1_raw_score = row['l1_score']
-                            real_titan = row['real_titan']
-                            
-                            norm_struct = 0.0
-                            if sig_type == 'BREAKOUT': norm_struct = min(1.0, l1_raw_score / 3.0) 
-                            elif sig_type == 'REVERSAL': norm_struct = l1_raw_score / 100.0
-                            
-                            score = 0.0
-                            if (w_titan + w_struct) > 0:
-                                score = ((real_titan * w_titan) + (norm_struct * w_struct)) / (w_titan + w_struct)
-                            
-                            if score >= entry_thresh:
-                                wallet["positions"][sym] = {'entry_price': row['close'], 'size_usd': position_size}
-                                wallet["allocated"] += position_size
-                                if len(wallet["positions"]) >= max_slots: break
-                
-                if wallet["balance"] < 1.0 and len(wallet["positions"]) == 0: break
-
-            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]
-                win_rate = (len(wins) / len(trades)) * 100
-                
-                results.append({
-                    'config': config,
-                    'final_balance': wallet["balance"],
-                    'net_profit': net_profit,
-                    'total_trades': len(trades),
-                    'win_rate': win_rate,
-                })
-            else:
-                 results.append({'config': config, 'final_balance': initial_capital, 'net_profit': 0.0, 'total_trades': 0, 'win_rate': 0.0})
-
-        return results
-
-    async def run_optimization(self, target_regime="RANGE"):
-        await self.generate_truth_data()
-        
-        start_ts = int(datetime.strptime(self.force_start_date, "%Y-%m-%d").replace(tzinfo=timezone.utc).timestamp() * 1000)
-        end_ts = int(datetime.strptime(self.force_end_date, "%Y-%m-%d").replace(tzinfo=timezone.utc).timestamp() * 1000)
-        period_id = f"{start_ts}_{end_ts}"
-        
-        current_period_files = []
-        for f in os.listdir(CACHE_DIR):
-            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:
-            print(f"❌ No signals for {target_regime}.")
-            return None
-
-        print(f"\n🧩 [Phase 2] Optimizing for {target_regime}...")
-        print(f"   💰 Start Capital: ${self.INITIAL_CAPITAL}")
-        
-        w_titan_range = np.linspace(0.4, 0.9, num=self.GRID_DENSITY)
-        w_struct_range = np.linspace(0.1, 0.6, num=self.GRID_DENSITY)
-        thresh_range = np.linspace(0.20, 0.90, num=self.GRID_DENSITY)
-        
-        combinations = []
-        for wt, ws, th in itertools.product(w_titan_range, w_struct_range, thresh_range):
-            if 0.9 <= (wt + ws) <= 1.1:
-                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)]
-        
-        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}")
-
-        if not final_results: return None
-            
-        best = sorted(final_results, key=lambda x: x['final_balance'], reverse=True)[0]
-        
-        print("\n" + "="*60)
-        print(f"🏆 CHAMPION REPORT [{target_regime}]:")
-        print(f"   📅 Period: {self.force_start_date} -> {self.force_end_date}")
-        print(f"   💰 Final Balance: ${best['final_balance']:,.2f}")
-        print(f"   🚀 Net PnL:       ${best['net_profit']:,.2f}")
-        print(f"   📈 Win Rate:      {best['win_rate']:.1f}%")
-        print(f"   ⚙️ Config: Titan={best['config']['w_titan']} | Struct={best['config']['w_struct']} | Thresh={best['config']['thresh']}")
-        print("="*60)
-        
-        return best['config']
-
-async def run_strategic_optimization_task():
-    print("\n🧪 [STRATEGIC BACKTEST] Time Lord Initiated...")
-    r2 = R2Service()
-    dm = DataManager(None, None, r2)
-    proc = MLProcessor(dm)
-    
-    await dm.initialize()
-    await proc.initialize() 
-    
-    try:
-        hub = AdaptiveHub(r2)
-        await hub.initialize()
-        
-        # 🔥🔥🔥 خريطة السيناريوهات الشاملة 🔥🔥🔥
-        scenarios = [
-            # 1. BULL: ETF Rally
-            {"regime": "BULL", "start": "2024-01-01", "end": "2024-03-30"},
-            
-            # 2. BEAR: Crash of '23
-            {"regime": "BEAR", "start": "2023-08-01", "end": "2023-09-15"},
-            
-            # 3. DEAD: The Great Boredom
-            {"regime": "DEAD", "start": "2023-06-01", "end": "2023-08-01"},
-            
-            # 4. RANGE: Recent Chop
-            {"regime": "RANGE", "start": "2024-07-01", "end": "2024-09-30"}
-        ]
-        
-        optimizer = HeavyDutyBacktester(dm, proc)
-        
-        for scen in scenarios:
-            target = scen["regime"]
-            optimizer.set_date_range(scen["start"], scen["end"])
-            
-            best_config = await optimizer.run_optimization(target_regime=target)
-            
-            if best_config and target in hub.strategies:
-                print(f"💉 Injecting Optimized DNA into {target}...")
-                st = hub.strategies[target]
-                st.model_weights['titan'] = best_config['w_titan']
-                st.model_weights['structure'] = best_config['w_struct']
-                st.filters['l1_min_score'] = best_config['thresh']
-        
-        await hub._save_state_to_r2()
-        hub._inject_current_parameters()
-        print(f"✅ [System] ALL DNA Updated & Saved Successfully.")
-                
-    finally:
-        await dm.close()
-
-if __name__ == "__main__":
-    asyncio.run(run_strategic_optimization_task())
+            dt_end = datetime.strptime(self.force_end_date, "%Y-%m-%d").