Update backtest_engine.py

backtest_engine.py

@@ -1,5 +1,5 @@
 # ============================================================
-# 🧪 backtest_engine.py (V82.0 - GEM-Architect: Verbose Logger)
+# 🧪 backtest_engine.py (V85.0 - GEM-Architect: Full Audit)
 # ============================================================
 
 import asyncio
@@ -17,6 +17,7 @@ from typing import Dict, Any, List
 try:
     from ml_engine.processor import MLProcessor, SystemLimits 
     from ml_engine.data_manager import DataManager
+    # ✅ استيراد كامل للهب والاستراتيجية
     from learning_hub.adaptive_hub import StrategyDNA, AdaptiveHub
     from r2 import R2Service
 except ImportError:
@@ -35,6 +36,7 @@ class HeavyDutyBacktester:
         self.TRADING_FEES = 0.001 
         self.MAX_SLOTS = 4         
         
+        # القائمة الكاملة (50 عملة)
         self.TARGET_COINS = [ 
             'SOL/USDT', 'XRP/USDT', 'DOGE/USDT', 'ADA/USDT', 'AVAX/USDT', 
             'LINK/USDT', 'TON/USDT', 'INJ/USDT', 'APT/USDT', 'OP/USDT', 
@@ -52,7 +54,7 @@ class HeavyDutyBacktester:
         self.force_end_date = None   
         
         if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
-        print(f"🧪 [Backtest V82.0] Verbose Logger Mode Ready.")
+        print(f"🧪 [Backtest V85.0] Full Audit Mode (Detailed Stats).")
 
     def set_date_range(self, start_str, end_str):
         self.force_start_date = start_str
@@ -63,19 +65,18 @@ class HeavyDutyBacktester:
         return df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].values.tolist()
 
     # ==============================================================
-    # 🧱 Core Logic: Single Coin Processor (VERBOSE)
+    # 🧱 Core Logic: Single Coin Processor (Safe Scope)
     # ==============================================================
     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.")
+            print(f"   📂 [{sym}] Data Exists -> Skipping.")
             return True
 
-        print(f"   ⏳ [{sym}] Starting FETCH sequence...", flush=True)
+        print(f"   ⚙️ Simulating {sym}...", end="", flush=True)
         t0 = time.time()
         
         all_candles_1m = []
@@ -93,38 +94,33 @@ class HeavyDutyBacktester:
                         timeout=10.0
                     )
                 except:
-                    print(f"      ⚠️ [{sym}] Net Retry...", flush=True)
+                    print(f"      ⚠️ [{sym}] Retry...", flush=True)
                     await asyncio.sleep(1)
                     continue
 
-                if not batch: 
-                    print(f"      ⚠️ [{sym}] No more data from exchange.", flush=True)
-                    break
+                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
-                
                 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(f"   ❌ [{sym}] FAILED: No data retrieved.")
+                print(f"   ❌ [{sym}] No data.")
                 return False
 
-            print(f"   ✅ [{sym}] Download Complete ({len(all_candles_1m)} candles). Processing...", flush=True)
+            print(f"   ✅ [{sym}] Downloaded {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')
@@ -147,7 +143,6 @@ class HeavyDutyBacktester:
             ai_results = []
             valid_indices = frames['5m'].index[500:] 
             
-            # محاكاة
             for t_idx in valid_indices:
                 current_timestamp = int(t_idx.timestamp() * 1000)
                 
@@ -203,7 +198,7 @@ class HeavyDutyBacktester:
                 pd.DataFrame(ai_results).to_pickle(scores_file)
                 print(f"   💾 [{sym}] Saved {len(ai_results)} signals. (Time: {dt:.1f}s)")
             else:
-                print(f"   ⚠️ [{sym}] No signals found. (Time: {dt:.1f}s)")
+                print(f"   ⚠️ [{sym}] No signals. (Time: {dt:.1f}s)")
             
             return True
 
@@ -223,4 +218,279 @@ class HeavyDutyBacktester:
     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").
+            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" 💀 [WATCHDOG] Killed {sym}. Moving on...")
+                    gc.collect()
+            gc.collect()
+            await asyncio.sleep(1.0)
+
+    # ==============================================================
+    # PHASE 2: Portfolio Digital Twin Engine (Full Audit Stats)
+    # ==============================================================
+    @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']
+            
+            # تتبع الـ Drawdown
+            peak_balance = initial_capital
+            max_drawdown = 0.0
+            
+            for ts, group in grouped_by_time:
+                active_symbols = list(wallet["positions"].keys())
+                current_prices = {row['symbol']: row['close'] for _, row in group.iterrows()}
+                
+                # 1. Check Exits
+                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})
+
+                # تحديث قمة الرصيد وحساب Drawdown
+                current_total_equity = wallet["balance"] # نتجاهل العائم للسرعة، نحسب المحقق فقط
+                if current_total_equity > peak_balance:
+                    peak_balance = current_total_equity
+                
+                dd = (peak_balance - current_total_equity) / peak_balance
+                if dd > max_drawdown: max_drawdown = dd
+
+                # 2. Check Entries
+                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]
+                
+                total_trades = len(trades)
+                win_count = len(wins)
+                loss_count = len(losses)
+                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
+                
+                # حساب السلاسل المتتالية
+                current_win_streak = 0; max_win_streak = 0
+                current_loss_streak = 0; max_loss_streak = 0
+                
+                for p in pnls:
+                    if p > 0:
+                        current_win_streak += 1
+                        current_loss_streak = 0
+                        if current_win_streak > max_win_streak: max_win_streak = current_win_streak
+                    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"],
+                    'net_profit': net_profit,
+                    'total_trades': total_trades,
+                    'win_count': win_count,
+                    'loss_count': loss_count,
+                    'win_rate': win_rate,
+                    'max_single_win': max_single_win,
+                    'max_single_loss': max_single_loss,
+                    'max_win_streak': max_win_streak,
+                    'max_loss_streak': max_loss_streak,
+                    'max_drawdown': max_drawdown * 100 # كنسبة مئوية
+                })
+            else:
+                 results.append({
+                     'config': config, 'final_balance': initial_capital, 'net_profit': 0.0, 
+                     'total_trades': 0, 'win_count': 0, 'loss_count': 0, 'win_rate': 0.0,
+                     'max_single_win': 0.0, 'max_single_loss': 0.0, 'max_win_streak': 0, 
+                     'max_loss_streak': 0, 'max_drawdown': 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, 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.60, 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, None
+            
+        best = sorted(final_results, key=lambda x: x['final_balance'], reverse=True)[0]
+        
+        # 🔥🔥🔥 The Full Audit Report 🔥🔥🔥
+        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("-" * 60)
+        print(f"   📊 Total Trades:    {best['total_trades']}")
+        print(f"   ✅ Winning Trades:  {best['win_count']}")
+        print(f"   ❌ Losing Trades:   {best['loss_count']}")
+        print(f"   📈 Win Rate:        {best['win_rate']:.1f}%")
+        print("-" * 60)
+        print(f"   🟢 Max Single Win:  ${best['max_single_win']:.2f}")
+        print(f"   🔴 Max Single Loss: ${best['max_single_loss']:.2f}")
+        print(f"   🔥 Max Win Streak:  {best['max_win_streak']} trades")
+        print(f"   🧊 Max Loss Streak: {best['max_loss_streak']} trades")
+        print(f"   📉 Max Drawdown:    {best['max_drawdown']:.1f}%")
+        print("-" * 60)
+        print(f"   ⚙️ Config: Titan={best['config']['w_titan']} | Struct={best['config']['w_struct']} | Thresh={best['config']['thresh']}")
+        print("="*60)
+        
+        return best['config'], best
+
+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 = [
+            {"regime": "BULL", "start": "2024-01-01", "end": "2024-03-30"},
+            {"regime": "BEAR", "start": "2023-08-01", "end": "2023-09-15"},
+            {"regime": "DEAD", "start": "2023-06-01", "end": "2023-08-01"},
+            {"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, best_stats = await optimizer.run_optimization(target_regime=target)
+            
+            if best_config and best_stats:
+                hub.submit_challenger(target, best_config, best_stats)
+        
+        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())