Update backtest_engine.py

backtest_engine.py

@@ -1,5 +1,5 @@
 # ============================================================
-# 🧪 backtest_engine.py (V95.0 - GEM-Architect: Vectorized Sonar)
+# 🧪 backtest_engine.py (V96.0 - GEM-Architect: Vectorized Logic Mirror)
 # ============================================================
 
 import asyncio
@@ -36,7 +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', 'ARB/USDT', 'SUI/USDT', 
@@ -53,7 +53,7 @@ class HeavyDutyBacktester:
         self.force_end_date = None   
         
         if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
-        print(f"🧪 [Backtest V95.0] Vectorized Sonar (Skipping Noise).")
+        print(f"🧪 [Backtest V96.0] Vectorized Logic Mirror (Exact L1 Simulation).")
 
     def set_date_range(self, start_str, end_str):
         self.force_start_date = start_str
@@ -63,32 +63,6 @@ class HeavyDutyBacktester:
         if df.empty: return []
         return df[['timestamp', 'open', 'high', 'low', 'close', 'volume']].values.tolist()
 
-    # ==============================================================
-    # 🚀 VECTORIZED INDICATORS (The Sonar)
-    # ==============================================================
-    def _calculate_sonar_indicators(self, df):
-        """
-        حساب مؤشرات سريعة جداً (Vectorized) لكامل البيانات دفعة واحدة.
-        الهدف: تحديد المناطق "الميتة" لتجاهلها.
-        """
-        # 1. RSI Calculation (Manual Numpy for Speed)
-        delta = df['close'].diff()
-        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
-        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
-        rs = gain / loss
-        df['sonar_rsi'] = 100 - (100 / (1 + rs))
-        
-        # 2. Volume MA
-        df['sonar_vol_ma'] = df['volume'].rolling(window=20).mean()
-        
-        # 3. Bollinger Band Width (Volatility check)
-        # sma20 = df['close'].rolling(window=20).mean()
-        # std = df['close'].rolling(window=20).std()
-        # df['sonar_bb_width'] = (std * 2) / sma20 # عرض القناة كنسبة
-        
-        df.fillna(0, inplace=True)
-        return df
-
     # ==============================================================
     # ⚡ FAST DATA DOWNLOADER
     # ==============================================================
@@ -130,11 +104,44 @@ class HeavyDutyBacktester:
                 unique_candles.append(c)
                 seen.add(c[0])
         unique_candles.sort(key=lambda x: x[0])
-        print(f"     ✅ Downloaded {len(unique_candles)} candles for {sym}.", flush=True)
+        print(f"     ✅ Downloaded {len(unique_candles)} candles.", flush=True)
         return unique_candles
 
     # ==============================================================
-    # 🧠 CPU PROCESSING (SONAR + NUMPY)
+    # 🏎️ VECTORIZED INDICATOR CALCULATION
+    # ==============================================================
+    def _calculate_indicators_vectorized(self, df):
+        """
+        حساب المؤشرات الفنية لكامل البيانات دفعة واحدة باستخدام Pandas Vectorization.
+        هذا يطابق منطق DataManager._calc_indicators بالضبط ولكن أسرع بـ 1000 مرة.
+        """
+        # RSI
+        delta = df['close'].diff()
+        gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
+        loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
+        rs = gain / loss
+        df['rsi'] = 100 - (100 / (1 + rs))
+        
+        # EMA
+        df['ema20'] = df['close'].ewm(span=20, adjust=False).mean()
+        df['ema50'] = df['close'].ewm(span=50, adjust=False).mean()
+        
+        # ATR (Simplified Vectorized)
+        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()
+        
+        # Volume MA
+        df['vol_ma20'] = df['volume'].rolling(window=20).mean()
+        
+        df.fillna(0, inplace=True)
+        return df
+
+    # ==============================================================
+    # 🧠 CPU PROCESSING (VECTORIZED LOGIC MIRROR)
     # ==============================================================
     async def _process_data_in_memory(self, sym, candles, start_ms, end_ms):
         safe_sym = sym.replace('/', '_')
@@ -145,7 +152,7 @@ class HeavyDutyBacktester:
              print(f"   📂 [{sym}] Data Exists -> Skipping.")
              return
 
-        print(f"   ⚙️ [CPU] Processing {sym}...", flush=True)
+        print(f"   ⚙️ [CPU] Analyzing {sym}...", flush=True)
         t0 = time.time()
 
         # 1. Prepare Pandas
@@ -156,13 +163,13 @@ class HeavyDutyBacktester:
         df_1m.set_index('datetime', inplace=True)
         df_1m = df_1m.sort_index()
 
-        # 2. Resample (Base Frames)
+        # 2. Resample & Calculate Indicators (ONCE)
         frames = {}
         numpy_frames = {}
         time_indices = {}
         agg_dict = {'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
         
-        # 1m
+        # 1m Setup
         frames['1m'] = df_1m.copy()
         frames['1m']['timestamp'] = frames['1m'].index.floor('1min').astype(np.int64) // 10**6
         col_order = ['timestamp', 'open', 'high', 'low', 'close', 'volume']
@@ -173,89 +180,119 @@ class HeavyDutyBacktester:
         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()
             
-            # --- 🚀 SONAR INJECTION ---
-            # نحسب المؤشرات على فريم الـ 15m و 1h (الفريمات المستخدمة في L1)
+            # 🔥🔥 Calculate Indicators HERE (Vectorized) 🔥🔥
             if tf_str in ['15m', '1h']:
-                resampled = self._calculate_sonar_indicators(resampled)
-            # --------------------------
-
+                resampled = self._calculate_indicators_vectorized(resampled)
+            
             resampled['timestamp'] = resampled.index.astype(np.int64) // 10**6
-            frames[tf_str] = resampled # Keep for sonar check
-            numpy_frames[tf_str] = resampled[col_order].values
+            frames[tf_str] = resampled
+            numpy_frames[tf_str] = resampled[col_order].values # Raw data for passing
             time_indices[tf_str] = resampled.index
 
         ai_results = []
         valid_idx_5m = time_indices['5m']
-        start_dt = valid_idx_5m[0] + pd.Timedelta(minutes=500)
-        start_pos = valid_idx_5m.searchsorted(start_dt)
-        total_steps = len(valid_idx_5m)
         
-        # مصفوفات السونار للوصول السريع
-        # نحتاج الوصول لقيم RSI و Volume بسرعة بدون Pandas Loop
-        # سنستخدم مؤشرات الوقت للربط
+        # 3. 🔥 VECTORIZED LOGIC MATCHING (The Speed Force) 🔥
+        # بدلاً من حلقة تكرار عمياء، نجد "أماكن الاهتمام" فوراً باستخدام المنطق البولياني
+        # هذا يطابق شروط DataManager._apply_logic_tree حرفياً
         
-        # --- THE TURBO LOOP ---
-        for i in range(start_pos, len(valid_idx_5m)):
-            current_time = valid_idx_5m[i]
-            
-            if i % (total_steps // 4) == 0:
-                 print(f"     🚀 Processing: {int((i/total_steps)*100)}%...", flush=True)
-
-            # --- 📡 SONAR CHECK (The Accelerator) ---
-            # نتحقق من فريم 1h أولاً (لأنه الأبطأ والأهم)
-            idx_1h = time_indices['1h'].searchsorted(current_time, side='right') - 1
-            if idx_1h < 60: continue
-            
-            # قراءة قيم السونار من الذاكرة (Pandas Access is okay here as it's single row via iloc/fast access or numpy)
-            # لكن الأسرع هو قراءتها من الـ Numpy إذا جهزناها، أو من الـ DataFrame مباشرة.
-            # للسرعة القصوى، سنفترض أننا نصل للصف في الـ DataFrame.
-            
-            row_1h = frames['1h'].iloc[idx_1h]
-            sonar_rsi_1h = row_1h['sonar_rsi']
-            
-            # قاعدة السونار:
-            # إذا RSI بين 35 و 65، فهذا غالباً نطاق عرضي ممل لا يستحق التشغيل الكامل
-            # إلا إذا كان هناك انفجار في الفوليوم.
-            # ملاحظة: نحن نريد "Wide Net" لذلك سنكون متساهلين، لكن ليس أغبياء.
-            # نتجاهل فقط المناطق "الميتة تماماً".
+        # نحتاج لمطابقة وقت الـ 5m مع الـ 1h و 15m
+        # سنقوم بعمل reindex للـ 1h و 15m ليتطابق مع الـ 5m (Forward Fill)
+        # هذا يسمح لنا بمقارنة الأعمدة مباشرة
+        
+        df_5m_aligned = frames['5m'].copy()
+        
+        # دمج بيانات الـ 1h مع الـ 5m (Matching times)
+        df_1h_aligned = frames['1h'].reindex(frames['5m'].index, method='ffill')
+        df_15m_aligned = frames['15m'].reindex(frames['5m'].index, method='ffill')
+        
+        # --- تطبيق شروط L1 (Breakout & Reversal) ---
+        
+        # الشروط المشتركة (Common Filters from V15.2)
+        # 1. 4H Change calculation (approx from 1H data)
+        # shift(4) في فريم الساعة يقابل shift(48) في فريم 5 دقائق (تقريباً)
+        # للأمان نست��دم بيانات الساعة المحاذية
+        change_4h = ((df_1h_aligned['close'] - df_1h_aligned['close'].shift(4)) / df_1h_aligned['close'].shift(4)) * 100
+        
+        # فلتر: ممنوع أكثر من 8% صعود في 4 ساعات
+        cond_not_pump = change_4h <= 8.0
+        # فلتر: RSI 1H ممنوع فوق 70
+        cond_rsi_1h_safe = df_1h_aligned['rsi'] <= 70
+        # فلتر: الامتداد (Deviation)
+        deviation = (df_1h_aligned['close'] - df_1h_aligned['ema20']) / df_1h_aligned['atr']
+        cond_deviation_safe = deviation <= 1.8
+        
+        filters_pass = cond_not_pump & cond_rsi_1h_safe & cond_deviation_safe
+        
+        # --- Breakout Logic ---
+        # 1. Bullish Structure (1H)
+        bullish_1h = (df_1h_aligned['ema20'] > df_1h_aligned['ema50']) | (df_1h_aligned['close'] > df_1h_aligned['ema20'])
+        # 2. RSI 1H Room (45-68)
+        rsi_1h_ok = (df_1h_aligned['rsi'] >= 45) & (df_1h_aligned['rsi'] <= 68)
+        # 3. 15M Close > EMA20
+        close_above_ema_15m = df_15m_aligned['close'] >= df_15m_aligned['ema20']
+        # 4. Volume 15M Spike
+        vol_spike_15m = df_15m_aligned['volume'] >= (1.5 * df_15m_aligned['vol_ma20'])
+        
+        is_breakout = filters_pass & bullish_1h & rsi_1h_ok & close_above_ema_15m & vol_spike_15m
+        
+        # --- Reversal Logic ---
+        # 1. RSI 1H Oversold (20-40)
+        rsi_oversold = (df_1h_aligned['rsi'] >= 20) & (df_1h_aligned['rsi'] <= 40)
+        # 2. Drop in price (change_4h <= -2)
+        price_drop = change_4h <= -2.0
+        # 3. Hammer/Rejection on 15M (Vectorized Approximation)
+        # Hammer: Lower wick > 1.5 * Body
+        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)
+        
+        # --- Combined Mask ---
+        # هذه هي اللحظات التي تستحق التحليل فقط!
+        valid_mask = is_breakout | is_reversal
+        valid_indices = df_5m_aligned[valid_mask].index
+        
+        # --------------------------------------------------------
+        
+        # 4. Loop ONLY on Valid Indices (The massive speedup)
+        # بدلاً من 129,000 لفة، سنجد ربما 2,000 - 5,000 لفة فقط.
+        
+        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} potential setups. Running Titan...", flush=True)
+        
+        for i, current_time in enumerate(final_valid_indices):
+            # قص البيانات (Slicing) لتمريرها للنماذج
+            # نستخدم searchsorted للسرعة القصوى
             
-            is_interesting = False
+            # نحتاج تحويل timestamp الـ index إلى مكان في الـ numpy arrays
+            # ملاحظة: time_indices['1m'] مرتب، لذا searchsorted يعمل
             
-            # 1. RSI Extreme (Reversal potential)
-            if sonar_rsi_1h < 42 or sonar_rsi_1h > 58: 
-                is_interesting = True
+            idx_1m = time_indices['1m'].searchsorted(current_time, side='right') - 1
+            idx_5m = time_indices['5m'].searchsorted(current_time, side='right') - 1
+            idx_15m = time_indices['15m'].searchsorted(current_time, side='right') - 1
+            idx_1h = time_indices['1h'].searchsorted(current_time, side='right') - 1
+            idx_4h = time_indices['4h'].searchsorted(current_time, side='right') - 1
+            idx_1d = time_indices['1d'].searchsorted(current_time, side='right') - 1
             
-            # 2. Volume Spike (Breakout potential)
-            # if row_1h['volume'] > row_1h['sonar_vol_ma'] * 1.2:
-            #     is_interesting = True
+            if idx_1m < 500 or idx_4h < 100: continue
             
-            # إذا لم يكن مثيراً للاهتمام في الـ 1h، نتحقق من الـ 15m (للأحداث السريعة)
-            if not is_interesting:
-                idx_15m = time_indices['15m'].searchsorted(current_time, side='right') - 1
-                if idx_15m >= 60:
-                    row_15m = frames['15m'].iloc[idx_15m]
-                    if row_15m['sonar_rsi'] < 35 or row_15m['sonar_rsi'] > 65:
-                        is_interesting = True
-                    # if row_15m['volume'] > row_15m['sonar_vol_ma'] * 1.5:
-                    #     is_interesting = True
-
-            # 🛑 SKIP if boring
-            if not is_interesting:
-                continue 
-
-            # --- إذا وصلنا هنا، فالشمعة تستحق التحليل ---
+            # استخراج نوع الإشارة (لأننا دمجناهم في valid_mask)
+            # نعيد التحقق السريع لنعرف النوع
+            # ملاحظة: الوصول هنا سريع جداً لأننا نعرف التوقيت
+            # أو يمكننا الاعتماد على أن DataManager سيعيد النوع الصحيح
             
-            # قص البيانات (Slicing) - كما في V94
-            idx_15m = time_indices['15m'].searchsorted(current_time, side='right') - 1
+            # نجهز الـ Packet ونرسلها لـ DataManager للتأكيد النهائي واستخراج السكور
+            # هذا يضمن التطابق 100%
             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 
-            }
+            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
@@ -265,13 +302,6 @@ class HeavyDutyBacktester:
             l1_score = logic_result.get('score', 0.0)
             
             if signal_type in ['BREAKOUT', 'REVERSAL']:
-                idx_1m = time_indices['1m'].searchsorted(current_time, side='right') - 1
-                idx_5m = i 
-                idx_4h = time_indices['4h'].searchsorted(current_time, side='right') - 1
-                idx_1d = time_indices['1d'].searchsorted(current_time, side='right') - 1
-                
-                if idx_1m < 500 or idx_4h < 100: continue 
-                
                 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(),
@@ -280,7 +310,6 @@ class HeavyDutyBacktester:
                     '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
                 
@@ -291,22 +320,17 @@ class HeavyDutyBacktester:
                 except: pass
 
                 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
+                    'timestamp': ts_aligned, '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"   💾 [{sym}] Saved {len(ai_results)} candidates. (Processed in {dt:.1f}s)", flush=True)
+            print(f"   💾 [{sym}] Saved {len(ai_results)} verified signals. (Compute: {dt:.1f}s)", flush=True)
         else:
-            print(f"   ⚠️ [{sym}] No interesting candidates.", flush=True)
+            print(f"   ⚠️ [{sym}] No signals.", flush=True)
             
         del numpy_frames, time_indices, df_1m, candles, frames
         gc.collect()
@@ -320,7 +344,7 @@ 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"\n🚜 [Phase 1] Era: {self.force_start_date} -> {self.force_end_date}")
         else:
             return 
         
@@ -330,9 +354,9 @@ class HeavyDutyBacktester:
                 if candles:
                     await self._process_data_in_memory(sym, candles, start_time_ms, end_time_ms)
                 else:
-                    print(f"   ❌ Failed/Empty data for {sym}. Continuing...", flush=True)
+                    print(f"   ❌ Failed/Empty data for {sym}.", flush=True)
             except Exception as e:
-                print(f"   ❌ SKIP: Error processing {sym}: {e}", flush=True)
+                print(f"   ❌ SKIP {sym}: {e}", flush=True)
                 continue
             gc.collect()
 
@@ -497,27 +521,14 @@ class HeavyDutyBacktester:
         
         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] Vectorized Sonar Mode Initiated...")
+    print("\n🧪 [STRATEGIC BACKTEST] Vectorized Logic Mirror Initiated...")
     r2 = R2Service()
     dm = DataManager(None, None, r2)
     proc = MLProcessor(dm)