Update backtest_engine.py

backtest_engine.py

@@ -1,5 +1,5 @@
 # ============================================================
-# 🧪 backtest_engine.py (V101.1 - GEM-Architect: Silent Mode)
+# 🧪 backtest_engine.py (V104.0 - GEM-Architect: High Fidelity & Speed Injection)
 # ============================================================
 
 import asyncio
@@ -31,6 +31,7 @@ class HeavyDutyBacktester:
     def __init__(self, data_manager, processor):
         self.dm = data_manager
         self.proc = processor
+        # كثافة الشبكة للدخول
         self.GRID_DENSITY = 6 
         self.INITIAL_CAPITAL = 10.0
         self.TRADING_FEES = 0.001 
@@ -52,17 +53,14 @@ class HeavyDutyBacktester:
         self.force_end_date = None   
         
         if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
-        print(f"🧪 [Backtest V101.1] Smart Adaptive Grid (Silent Hydra).")
+        print(f"🧪 [Backtest V104.0] High-Fidelity 1-Minute Scan (Speed Injected).")
 
     def set_date_range(self, start_str, end_str):
         self.force_start_date = start_str
         self.force_end_date = end_str
 
-    # ... (Rest of fetch and vector logic unchanged) ...
-    # ... Keeping standard methods to save space ...
-
     # ==============================================================
-    # ⚡ FAST DATA DOWNLOADER (Same as before)
+    # ⚡ FAST DATA DOWNLOADER
     # ==============================================================
     async def _fetch_all_data_fast(self, sym, start_ms, end_ms):
         print(f"   ⚡ [Network] Downloading {sym}...", flush=True)
@@ -105,7 +103,17 @@ class HeavyDutyBacktester:
         print(f"     ✅ Downloaded {len(unique_candles)} candles.", flush=True)
         return unique_candles
 
+    # ==============================================================
+    # 🏎️ VECTORIZED INDICATORS (Optimized for Hydra Inputs)
+    # ==============================================================
     def _calculate_indicators_vectorized(self, df):
+        # 1. Basic Setup
+        df['close'] = df['close'].astype(float)
+        df['high'] = df['high'].astype(float)
+        df['low'] = df['low'].astype(float)
+        df['volume'] = df['volume'].astype(float)
+        
+        # 2. RSI & EMA
         delta = df['close'].diff()
         gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
         loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
@@ -114,16 +122,29 @@ class HeavyDutyBacktester:
         df['ema20'] = df['close'].ewm(span=20, adjust=False).mean()
         df['ema50'] = df['close'].ewm(span=50, adjust=False).mean()
         
+        # 3. ATR
         high_low = df['high'] - df['low']
         high_close = (df['high'] - df['close'].shift()).abs()
         low_close = (df['low'] - df['close'].shift()).abs()
         ranges = pd.concat([high_low, high_close, low_close], axis=1)
         true_range = ranges.max(axis=1)
         df['atr'] = true_range.rolling(14).mean()
+        
+        # 4. 🔥 Hydra Specifics (Pre-calc for Speed)
+        sma20 = df['close'].rolling(20).mean()
+        std20 = df['close'].rolling(20).std()
+        df['bb_width'] = ((sma20 + 2*std20) - (sma20 - 2*std20)) / sma20
+        
         df['vol_ma20'] = df['volume'].rolling(window=20).mean()
+        df['vol_ma50'] = df['volume'].rolling(window=50).mean()
+        df['rel_vol'] = df['volume'] / (df['vol_ma50'] + 1e-9)
+        
         df.fillna(0, inplace=True)
         return df
 
+    # ==============================================================
+    # 🧠 CPU PROCESSING (1-Minute Scan)
+    # ==============================================================
     async def _process_data_in_memory(self, sym, candles, start_ms, end_ms):
         safe_sym = sym.replace('/', '_')
         period_suffix = f"{start_ms}_{end_ms}"
@@ -133,12 +154,10 @@ class HeavyDutyBacktester:
              print(f"   📂 [{sym}] Data Exists -> Skipping.")
              return
 
-        print(f"   ⚙️ [CPU] Analyzing {sym} (Silent Profiling)...", flush=True)
+        print(f"   ⚙️ [CPU] Analyzing {sym} (1-Min Fidelity)...", flush=True)
         t0 = time.time()
 
         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()
@@ -148,8 +167,21 @@ class HeavyDutyBacktester:
         time_indices = {}
         agg_dict = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
         
+        # --- 1. Pre-Calculate EVERYTHING (Vectorized) ---
         frames['1m'] = df_1m.copy()
         frames['1m']['timestamp'] = frames['1m'].index.floor('1min').astype(np.int64) // 10**6
+        frames['1m'] = self._calculate_indicators_vectorized(frames['1m'])
+
+        # Store critical columns in Numpy for O(1) Access
+        fast_1m = {
+            'close': frames['1m']['close'].values,
+            'rsi': frames['1m']['rsi'].values,
+            'atr': frames['1m']['atr'].values,
+            'bb_width': frames['1m']['bb_width'].values,
+            'rel_vol': frames['1m']['rel_vol'].values,
+            'timestamp': frames['1m']['timestamp'].values
+        }
+        
         col_order = ['timestamp', 'open', 'high', 'low', 'close', 'volume']
         numpy_frames['1m'] = frames['1m'][col_order].values 
         time_indices['1m'] = frames['1m'].index
@@ -164,8 +196,8 @@ class HeavyDutyBacktester:
             time_indices[tf_str] = resampled.index
 
         ai_results = []
-        valid_idx_5m = time_indices['5m']
         
+        # --- L1 Logic (Filter) ---
         df_5m_aligned = frames['5m'].copy()
         df_1h_aligned = frames['1h'].reindex(frames['5m'].index, method='ffill')
         df_15m_aligned = frames['15m'].reindex(frames['5m'].index, method='ffill')
@@ -185,21 +217,23 @@ class HeavyDutyBacktester:
         
         rsi_oversold = (df_1h_aligned['rsi'] >= 20) & (df_1h_aligned['rsi'] <= 40)
         price_drop = change_4h <= -2.0
-        body = (df_15m_aligned['close'] - df_15m_aligned['open']).abs()
-        lower_wick = df_15m_aligned[['open', 'close']].min(axis=1) - df_15m_aligned['low']
-        is_hammer = lower_wick > (body * 1.5)
         is_green = df_15m_aligned['close'] > df_15m_aligned['open']
-        is_reversal = filters_pass & rsi_oversold & price_drop & (is_hammer | is_green)
-        
-        valid_mask = is_breakout | is_reversal
-        valid_indices = df_5m_aligned[valid_mask].index
-        
+        is_reversal = filters_pass & rsi_oversold & price_drop & is_green 
+
+        valid_indices = df_5m_aligned[is_breakout | is_reversal].index
         start_dt = df_1m.index[0] + pd.Timedelta(minutes=500)
         final_valid_indices = [t for t in valid_indices if t >= start_dt]
         
-        total_hits = len(final_valid_indices)
-        print(f"     🎯 Found {total_hits} signals.", flush=True)
-        
+        print(f"     🎯 Found {len(final_valid_indices)} signals. Running 1-Min Simulations...", flush=True)
+
+        # --- Hydra Model Access ---
+        hydra_models = {}
+        hydra_cols = []
+        if self.proc.guardian_hydra and self.proc.guardian_hydra.initialized:
+            hydra_models = self.proc.guardian_hydra.models
+            hydra_cols = self.proc.guardian_hydra.feature_cols
+
+        # --- 3. The Main Loop (Every Minute Check) ---
         for i, current_time in enumerate(final_valid_indices):
             idx_1m = time_indices['1m'].searchsorted(current_time, side='right') - 1
             idx_5m = time_indices['5m'].searchsorted(current_time, side='right') - 1
@@ -213,102 +247,115 @@ class HeavyDutyBacktester:
             ohlcv_1h = numpy_frames['1h'][idx_1h-60+1 : idx_1h+1].tolist()
             ohlcv_15m = numpy_frames['15m'][idx_15m-60+1 : idx_15m+1].tolist()
             
-            logic_packet = {'symbol': sym, 'ohlcv_1h': ohlcv_1h, 'ohlcv_15m': ohlcv_15m, 'change_24h': 0.0}
-            if len(ohlcv_1h) >= 24:
-                p_now = ohlcv_1h[-1][4]; p_old = ohlcv_1h[-24][4]
-                if p_old > 0: logic_packet['change_24h'] = ((p_now - p_old) / p_old) * 100
-
-            logic_result = self.dm._apply_logic_tree(logic_packet)
-            signal_type = logic_result.get('type', 'NONE')
-            l1_score = logic_result.get('score', 0.0)
+            # Logic Classification (Simplified)
+            sig_type = 'BREAKOUT' if is_breakout[current_time] else 'REVERSAL'
+            l1_score = 100.0 if sig_type == 'REVERSAL' else 20.0
+
+            # 🔥 RISK PROFILING (Every 1 Minute)
+            max_hydra_crash = 0.0
+            max_hydra_giveback = 0.0
+            max_legacy_v2 = 0.0
+            max_legacy_v3 = 0.0
             
-            if signal_type in ['BREAKOUT', 'REVERSAL']:
-                ohlcv_data = {
-                    '1m': numpy_frames['1m'][idx_1m-500+1 : idx_1m+1].tolist(),
-                    '5m': numpy_frames['5m'][idx_5m-200+1 : idx_5m+1].tolist(),
-                    '15m': ohlcv_15m,
-                    '1h': ohlcv_1h,
-                    '4h': numpy_frames['4h'][idx_4h-100+1 : idx_4h+1].tolist(),
-                    '1d': numpy_frames['1d'][idx_1d-50+1 : idx_1d+1].tolist()
-                }
-                current_price = ohlcv_data['5m'][-1][4]
-                real_titan = 0.5
-                
-                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
-
-                # 🔥 RISK PROFILING
-                max_hydra_crash = 0.0
-                max_hydra_giveback = 0.0
-                max_legacy_v2 = 0.0
-                max_legacy_v3 = 0.0
-                
-                hydra_crash_time = 0
-                legacy_panic_time = 0
-                
-                trade_ctx = {
-                    'entry_price': current_price, 'entry_time': str(current_time),
-                    'volume_30m_usd': 1000000 
-                }
-                
-                future_limit = 240
-                end_idx_1m = min(idx_1m + future_limit, len(numpy_frames['1m']) - 1)
-                check_step = 10 
-                current_idx_1m = idx_1m
+            hydra_crash_time = 0
+            legacy_panic_time = 0
+            
+            entry_price = fast_1m['close'][idx_1m]
+            highest_price = entry_price 
+            
+            # Scan next 4 hours (240 minutes), STEP = 1 minute (Live System Fidelity)
+            future_limit = 240
+            end_idx_1m = min(idx_1m + future_limit, len(fast_1m['close']) - 1)
+            
+            # Pre-lookup HTF RSIs to avoid heavy lookups every minute
+            rsi_5m_val = frames['5m']['rsi'].asof(current_time)
+            rsi_15m_val = frames['15m']['rsi'].asof(current_time)
+            
+            # Loop minute by minute
+            for current_idx_1m in range(idx_1m + 1, end_idx_1m + 1):
+                curr_price = fast_1m['close'][current_idx_1m]
+                if curr_price > highest_price: highest_price = curr_price
+                current_ts = int(fast_1m['timestamp'][current_idx_1m])
                 
-                while current_idx_1m < end_idx_1m:
-                    current_idx_1m += check_step
+                # 🐉 A. Hydra Injection (Every Minute)
+                if hydra_models:
+                    atr_val = fast_1m['atr'][current_idx_1m]
+                    sl_dist = 1.5 * atr_val if atr_val > 0 else entry_price * 0.015
                     
-                    future_1m_data = numpy_frames['1m'][current_idx_1m-1000+1 : current_idx_1m+1].tolist()
-                    future_5m_data = numpy_frames['5m'][idx_5m-300+1 : idx_5m+1].tolist() 
-                    current_ts = int(numpy_frames['1m'][current_idx_1m][0])
+                    pnl_r = (curr_price - entry_price) / sl_dist if sl_dist > 0 else 0
+                    max_pnl_r = (highest_price - entry_price) / sl_dist if sl_dist > 0 else 0
                     
-                    if self.proc.guardian_hydra:
-                        hydra_res = self.proc.guardian_hydra.analyze_position(sym, future_1m_data, future_5m_data, ohlcv_15m, trade_ctx)
-                        probs = hydra_res.get('probs', {})
-                        p_crash = probs.get('crash', 0.0)
-                        
+                    row_dict = {
+                        'rsi_1m': fast_1m['rsi'][current_idx_1m],
+                        'rsi_5m': rsi_5m_val,  
+                        'rsi_15m': rsi_15m_val,
+                        'bb_width': fast_1m['bb_width'][current_idx_1m],
+                        'rel_vol': fast_1m['rel_vol'][current_idx_1m],
+                        'dist_ema20_1h': 0.0, 
+                        'atr_pct': atr_val / curr_price if curr_price > 0 else 0,
+                        'norm_pnl_r': pnl_r,
+                        'max_pnl_r': max_pnl_r,
+                        'dist_tp_atr': 0.0, 'dist_sl_atr': 0.0,
+                        'time_in_trade': (current_idx_1m - idx_1m),
+                        'entry_type': 0.0, 'oracle_conf': 0.8, 'l2_score': 0.7, 'target_class': 3.0
+                    }
+                    
+                    vec = [row_dict.get(c, 0.0) for c in hydra_cols]
+                    vec_np = np.array(vec).reshape(1, -1)
+                    
+                    try:
+                        p_crash = hydra_models['crash'].predict_proba(vec_np)[0][1]
                         if p_crash > max_hydra_crash:
                             max_hydra_crash = p_crash
                             if p_crash > 0.6 and hydra_crash_time == 0: hydra_crash_time = current_ts
+                    except: pass
+                    
+                    try:
+                        p_give = hydra_models['giveback'].predict_proba(vec_np)[0][1]
+                        if p_give > max_hydra_giveback: max_hydra_giveback = p_give
+                    except: pass
+
+                # 🕸️ B. Legacy Check (Every Minute)
+                if self.proc.guardian_legacy:
+                    # Update 5m index based on time (approximate but fast)
+                    curr_5m_idx = idx_5m + (current_idx_1m - idx_1m) // 5
+                    if curr_5m_idx >= len(numpy_frames['5m']): curr_5m_idx = len(numpy_frames['5m']) - 1
+                    
+                    f_1m = numpy_frames['1m'][current_idx_1m-60+1 : current_idx_1m+1].tolist()
+                    f_5m = numpy_frames['5m'][curr_5m_idx-30+1 : curr_5m_idx+1].tolist() 
+                    
+                    legacy_res = self.proc.guardian_legacy.analyze_position(
+                        f_1m, f_5m, ohlcv_15m, entry_price, volume_30m_usd=1000000
+                    )
+                    scores = legacy_res.get('scores', {})
+                    s_v2 = scores.get('v2', 0.0)
+                    s_v3 = scores.get('v3', 0.0)
+                    
+                    if s_v2 > max_legacy_v2: 
+                        max_legacy_v2 = s_v2
+                        if s_v2 > 0.8 and legacy_panic_time == 0: legacy_panic_time = current_ts
                         
-                        if probs.get('giveback', 0) > max_hydra_giveback: max_hydra_giveback = probs.get('giveback', 0)
-
-                    if self.proc.guardian_legacy:
-                        legacy_res = self.proc.guardian_legacy.analyze_position(
-                            future_1m_data, future_5m_data, ohlcv_15m, current_price, volume_30m_usd=1000000
-                        )
-                        scores = legacy_res.get('scores', {})
-                        s_v2 = scores.get('v2', 0.0)
-                        s_v3 = scores.get('v3', 0.0)
-                        
-                        if s_v2 > max_legacy_v2: 
-                            max_legacy_v2 = s_v2
-                            if s_v2 > 0.8 and legacy_panic_time == 0: legacy_panic_time = current_ts
-                            
-                        if s_v3 > max_legacy_v3: max_legacy_v3 = s_v3
-
-                ts_aligned = int(current_time.timestamp() // 60) * 60 * 1000
-                ai_results.append({
-                    'timestamp': ts_aligned, 'symbol': sym, 'close': current_price,
-                    'real_titan': real_titan, 'signal_type': signal_type, 'l1_score': l1_score,
-                    'risk_hydra_crash': max_hydra_crash,
-                    'time_hydra_crash': hydra_crash_time,
-                    'risk_legacy_v2': max_legacy_v2,
-                    'time_legacy_panic': legacy_panic_time
-                })
+                    if s_v3 > max_legacy_v3: max_legacy_v3 = s_v3
+
+            ts_aligned = int(current_time.timestamp() // 60) * 60 * 1000
+            ai_results.append({
+                'timestamp': ts_aligned, 'symbol': sym, 'close': entry_price,
+                'real_titan': 0.5, 
+                'signal_type': sig_type, 'l1_score': l1_score,
+                'risk_hydra_crash': max_hydra_crash,
+                'time_hydra_crash': hydra_crash_time,
+                'risk_legacy_v2': max_legacy_v2,
+                'time_legacy_panic': legacy_panic_time
+            })
         
         dt = time.time() - t0
         if ai_results:
             pd.DataFrame(ai_results).to_pickle(scores_file)
-            print(f"   💾 [{sym}] Saved {len(ai_results)} profiled signals. ({dt:.1f}s)", flush=True)
+            print(f"   💾 [{sym}] Done in {dt:.2f}s. Saved {len(ai_results)} signals.", flush=True)
         else:
             print(f"   ⚠️ [{sym}] No signals.", flush=True)
             
-        del numpy_frames, time_indices, df_1m, candles, frames
+        del numpy_frames, time_indices, df_1m, candles, frames, fast_1m
         gc.collect()
 
     async def generate_truth_data(self):
@@ -318,19 +365,15 @@ 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}")
-        else:
-            return 
+        else: return 
         
         for sym in self.TARGET_COINS:
             try:
                 candles = await self._fetch_all_data_fast(sym, start_time_ms, end_time_ms)
                 if candles:
                     await self._process_data_in_memory(sym, candles, start_time_ms, end_time_ms)
-                else:
-                    print(f"   ❌ Failed/Empty data for {sym}.", flush=True)
             except Exception as e:
                 print(f"   ❌ SKIP {sym}: {e}", flush=True)
-                continue
             gc.collect()
 
     @staticmethod
@@ -353,6 +396,7 @@ class HeavyDutyBacktester:
         for config in combinations_batch:
             wallet = { "balance": initial_capital, "allocated": 0.0, "positions": {}, "trades_history": [] }
             
+            # Configs
             w_titan = config['w_titan']; w_struct = config['w_struct']; entry_thresh = config['thresh']
             hydra_thresh = config['hydra_thresh']
             legacy_thresh = config['legacy_thresh']
@@ -565,7 +609,7 @@ class HeavyDutyBacktester:
         print(f"   🛡️ Guard: Hydra={best['config']['hydra_thresh']} | Legacy={best['config']['legacy_thresh']}")
         print("="*60)
         return best['config'], best
-        
+
 async def run_strategic_optimization_task():
     print("\n🧪 [STRATEGIC BACKTEST] Smart Adaptive Grid Initiated...")
     r2 = R2Service()