Create backtest_engine.py

backtest_engine.py

@@ -0,0 +1,280 @@
+# ============================================================
+# 🧪 backtest_engine.py (V36.1 - GEM-Architect: Full Spectrum)
+# ============================================================
+# التحديثات:
+# 1. إدراج Sniper في المحاكاة (ML Only Mode).
+# 2. تحييد وزن دفتر الطلبات داخل Sniper مؤقتاً أثناء الاختبار.
+# 3. تحسين نظام التوليد (Resampling) ليكون Async بالكامل.
+# ============================================================
+
+import asyncio
+import pandas as pd
+import numpy as np
+import json
+import time
+from datetime import datetime, timedelta
+
+# استيراد المكونات الأساسية
+from ml_engine.processor import MLProcessor, SystemLimits
+from ml_engine.data_manager import DataManager
+from learning_hub.adaptive_hub import StrategyDNA
+
+class BacktestSimulator:
+    def __init__(self, data_manager, processor):
+        self.dm = data_manager
+        self.proc = processor
+        self.history_cache = {} # {symbol: DataFrame_1m_Full}
+        
+        # إعدادات المحاكاة
+        self.DAYS_TO_FETCH = 30  
+        self.CHUNK_LIMIT = 1500  
+        
+        print("🧪 [Backtest Engine V36.1] Full-Spectrum Simulation Initialized.")
+
+    # ==========================================================================
+    # 1. نظام جلب البيانات (كما هو)
+    # ==========================================================================
+    async def fetch_deep_history_1m(self, symbols: list):
+        """جلب شهر كامل بدقة دقيقة واحدة"""
+        print(f"⏳ [Backtest] Fetching 1m candles for {self.DAYS_TO_FETCH} days...")
+        end_time = int(time.time() * 1000)
+        start_time = end_time - (self.DAYS_TO_FETCH * 24 * 60 * 60 * 1000)
+        
+        for sym in symbols:
+            print(f"   📥 Downloading {sym}...", end="", flush=True)
+            all_candles = []
+            current_end = end_time
+            
+            while current_end > start_time:
+                candles = await self.dm.exchange.fetch_ohlcv(
+                    sym, '1m', limit=self.CHUNK_LIMIT, params={'endAt': int(current_end / 1000)}
+                )
+                if not candles: break
+                
+                first_candle_time = candles[0][0]
+                current_end = first_candle_time - 60000 
+                all_candles = candles + all_candles 
+                await asyncio.sleep(0.05) 
+            
+            if all_candles:
+                df = pd.DataFrame(all_candles, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
+                df['datetime'] = pd.to_datetime(df['timestamp'], unit='ms')
+                df = df.set_index('datetime').sort_index()
+                mask = (df.index >= pd.to_datetime(start_time, unit='ms')) & (df.index <= pd.to_datetime(end_time, unit='ms'))
+                self.history_cache[sym] = df.loc[mask]
+                print(f" ✅ ({len(self.history_cache[sym])} candles)")
+            else:
+                print(" ❌ No Data")
+
+    # ==========================================================================
+    # 2. نظام توليد الإطارات (Resampling)
+    # ==========================================================================
+    def resample_data(self, df_1m, end_idx):
+        """توليد الهرم المقلوب من البيانات"""
+        start_pos = max(0, end_idx - 1000) # يكفي 1000 دقيقة للوراء
+        slice_1m = df_1m.iloc[start_pos : end_idx].copy()
+        
+        if slice_1m.empty or len(slice_1m) < 60: return None
+
+        timeframes = {'1m': slice_1m.reset_index(drop=True).values.tolist()}
+        agg_dict = {'timestamp': 'first', 'open': 'first', 'high': 'max', 'low': 'min', 'close': 'last', 'volume': 'sum'}
+
+        try:
+            for tf, rule in [('5m', '5T'), ('15m', '15T'), ('1h', '1H'), ('4h', '4H')]:
+                resampled = slice_1m.resample(rule, on='datetime').agg(agg_dict).dropna()
+                timeframes[tf] = resampled.reset_index(drop=True).values.tolist()
+            return timeframes
+        except: return None
+
+    # ==========================================================================
+    # 3. محرك المحاكاة (Async Simulation Core)
+    # ==========================================================================
+    async def run_regime_simulation(self, regime_name: str, candidate_weights: dict):
+        """
+        تشغيل المحاكاة مع دمج Sniper في المعادلة
+        """
+        # 1. حقن الأوزان في الدستور (SystemLimits)
+        SystemLimits.CURRENT_REGIME = regime_name
+        SystemLimits.L2_WEIGHT_TITAN = candidate_weights['titan']
+        SystemLimits.L2_WEIGHT_PATTERNS = candidate_weights['patterns']
+        SystemLimits.L2_WEIGHT_MC = candidate_weights['mc']
+        
+        # 2. تكوين Sniper لوضع "الباكتست" (تجاهل دفتر الطلبات المفقود)
+        if self.proc.sniper:
+            self.proc.sniper.configure_settings(
+                threshold=0.4, # عتبة افتراضية
+                wall_ratio=0.9, # تساهل تام مع الجدران (لأنها غير موجودة)
+                w_ml=1.0,      # 🔥 الاعتماد الكلي على ML
+                w_ob=0.0       # إلغاء وزن دفتر الطلبات
+            )
+
+        capital = 1000.0
+        trades = 0
+        wins = 0
+        
+        print(f"   🧪 Testing Weights: {candidate_weights} ...")
+        
+        for sym, df_1m_full in self.history_cache.items():
+            if df_1m_full.empty: continue
+            
+            # نمشي بخطوات 30 دقيقة (تسريعاً للعملية)
+            step = 30
+            for i in range(1000, len(df_1m_full) - 240, step):
+                
+                ohlcv_multi = self.resample_data(df_1m_full, i)
+                if not ohlcv_multi: continue
+                
+                current_price = df_1m_full.iloc[i]['close']
+                
+                # --- A. Titan Prediction ---
+                titan_score = 0.5
+                if self.proc.titan:
+                    res = await asyncio.to_thread(self.proc.titan.predict, ohlcv_multi)
+                    titan_score = res.get('score', 0.5)
+                
+                # --- B. Patterns Prediction ---
+                pattern_score = 0.5
+                if self.proc.patterns:
+                    self.proc.patterns.configure_thresholds(
+                        weights=SystemLimits.PATTERN_TF_WEIGHTS,
+                        bull_thresh=SystemLimits.PATTERN_THRESH_BULLISH,
+                        bear_thresh=SystemLimits.PATTERN_THRESH_BEARISH
+                    )
+                    res = await self.proc.patterns.detect_chart_patterns(ohlcv_multi)
+                    pattern_score = res.get('pattern_confidence', 0.5)
+                
+                # --- C. Sniper Prediction (ML Only) ---
+                sniper_score = 0.5
+                if self.proc.sniper:
+                    # Sniper يحتاج قائمة 1m
+                    res = await self.proc.sniper.check_entry_signal_async(ohlcv_multi['1m'], order_book_data=None)
+                    sniper_score = res.get('ml_score', 0.5) # نأخذ ml_score الخام
+                
+                # --- D. Monte Carlo ---
+                mc_score = 0.5
+                if self.proc.mc_analyzer:
+                    closes_1h = [c[4] for c in ohlcv_multi['1h']]
+                    raw_mc = self.proc.mc_analyzer.run_light_check(closes_1h)
+                    mc_score = 0.5 + (raw_mc * 5.0)
+
+                # --- E. حساب النتيجة الموزونة (Weighted Score) ---
+                w_t = candidate_weights['titan']
+                w_p = candidate_weights['patterns']
+                w_s = candidate_weights['sniper'] # ✅ الوزن الجديد
+                w_m = candidate_weights['mc']
+                
+                total = w_t + w_p + w_s + w_m
+                final_score = ((titan_score * w_t) + (pattern_score * w_p) + (sniper_score * w_s) + (mc_score * w_m)) / total
+                
+                # قرار الدخول
+                if final_score > 0.75:
+                    # فحص النتيجة بعد 4 ساعات
+                    future_idx = i + 240 # 4 ساعات * 60 دقيقة
+                    if future_idx >= len(df_1m_full): break
+                    
+                    future_price = df_1m_full.iloc[future_idx]['close']
+                    change = (future_price - current_price) / current_price
+                    
+                    if change > 0.02: # ربح 2%
+                        capital *= 1.02
+                        wins += 1
+                        i += 120 # قفزة لتجنب التكرار
+                    elif change < -0.015: # خسارة 1.5%
+                        capital *= 0.985
+                        i += 120
+                    
+                    trades += 1
+
+        win_rate = (wins / trades * 100) if trades > 0 else 0
+        return {'final_capital': capital, 'win_rate': win_rate, 'trades': trades}
+
+    # ==========================================================================
+    # 4. المايسترو (Genetic Optimizer with Sniper)
+    # ==========================================================================
+    async def optimize_dna(self):
+        best_dna = {}
+        regimes = ['BULL', 'BEAR', 'RANGE'] # DEAD يحتاج استراتيجية خاصة
+        
+        # مساحة البحث (تشمل الآن Sniper)
+        search_space = [
+            # 1. متوازن
+            {'titan': 0.3, 'patterns': 0.3, 'sniper': 0.3, 'mc': 0.1},
+            # 2. تركيز على Titan (Trend)
+            {'titan': 0.6, 'patterns': 0.2, 'sniper': 0.1, 'mc': 0.1},
+            # 3. تركيز على Sniper (Scalping/Momentum)
+            {'titan': 0.2, 'patterns': 0.2, 'sniper': 0.5, 'mc': 0.1},
+            # 4. تركيز على Patterns (Reversal)
+            {'titan': 0.2, 'patterns': 0.6, 'sniper': 0.1, 'mc': 0.1},
+        ]
+        
+        for regime in regimes:
+            print(f"\n🧬 Optimizing DNA for {regime}...")
+            best_score = -9999
+            best_w = None
+            
+            for weights in search_space:
+                # تشغيل المحاكاة (Async مباشرة)
+                res = await self.run_regime_simulation(regime, weights)
+                
+                score = res['final_capital']
+                print(f"    -> {weights} => Cap: ${score:.1f} | WR: {res['win_rate']:.1f}% ({res['trades']} trds)")
+                
+                if score > best_score:
+                    best_score = score
+                    best_w = weights
+            
+            # حفظ أفضل إعدادات (مع إضافة Hydra كوزن ثابت لأنه خروج أكثر منه دخول)
+            if best_w:
+                best_dna[regime] = {
+                    "model_weights": {
+                        "titan": best_w['titan'],
+                        "patterns": best_w['patterns'],
+                        "sniper": best_w['sniper'], 
+                        "hydra": 0.1, # Hydra يضاف كوزن ثابت للدخول (أو يترك للحراس)
+                        "mc": best_w['mc']
+                    },
+                    "ob_settings": {"wall_ratio_limit": 0.4, "imbalance_thresh": 0.5},
+                    "filters": {"l1_min_score": 15.0, "l3_conf_thresh": 0.65}
+                }
+                print(f"🏆 WINNER for {regime}: Ti={best_w['titan']}, Sn={best_w['sniper']}")
+
+        return best_dna
+
+# ============================================================
+# 🚀 نقطة التشغيل
+# ============================================================
+async def run_strategic_optimization_task():
+    print("\n🧪 [STRATEGIC BACKTEST] Starting Sequence...")
+    
+    # 1. تهيئة الخدمات
+    from r2 import R2Service
+    r2 = R2Service()
+    dm = DataManager(None, None, r2)
+    await dm.initialize()
+    proc = MLProcessor(dm)
+    await proc.initialize() 
+    
+    sim = BacktestSimulator(dm, proc)
+    
+    # 2. تحميل البيانات
+    benchmark_coins = ['BTC/USDT', 'ETH/USDT', 'SOL/USDT', 'XRP/USDT'] 
+    await sim.fetch_deep_history_1m(benchmark_coins)
+    
+    # 3. التشغيل والتحسين
+    optimized_strategies = await sim.optimize_dna()
+    
+    # 4. حفظ النتائج وتحديث AdaptiveHub
+    from learning_hub.adaptive_hub import AdaptiveHub
+    hub = AdaptiveHub(r2)
+    await hub.initialize()
+    
+    for reg, data in optimized_strategies.items():
+        if reg in hub.strategies:
+            hub.strategies[reg].model_weights.update(data['model_weights'])
+            
+    await hub._save_state_to_r2()
+    await dm.close()
+    print("✅ [STRATEGIC BACKTEST] Completed Successfully.")
+
+if __name__ == "__main__":
+    asyncio.run(run_strategic_optimization_task())