Update backtest_engine.py

backtest_engine.py

@@ -1,5 +1,5 @@
 # ============================================================
-# 🧪 backtest_engine.py (V140.0 - GEM-Architect: Bulletproof Logic)
+# 🧪 backtest_engine.py (V141.0 - GEM-Architect: Scalar Enforcement)
 # ============================================================
 
 import asyncio
@@ -47,7 +47,7 @@ def _z_roll(x, w=500):
     return ((x - r) / s).fillna(0)
 
 def _revive_score_distribution(scores):
-    scores = np.array(scores, dtype=np.float32)
+    scores = np.array(scores, dtype=np.float32).flatten() # ✅ Force Flatten
     if len(scores) < 10: return scores
     std = np.std(scores)
     if std < 0.05: 
@@ -73,6 +73,9 @@ def _zv(x):
         s = np.nanstd(x, axis=0) + 1e-9
         return np.nan_to_num((x - m) / s, nan=0.0)
 
+# ============================================================
+# 🧩 PATTERN RECOGNITION HELPER
+# ============================================================
 def _transform_window_for_pattern(df_window):
     try:
         c = df_window['close'].values.astype('float32')
@@ -80,12 +83,14 @@ def _transform_window_for_pattern(df_window):
         h = df_window['high'].values.astype('float32')
         l = df_window['low'].values.astype('float32')
         v = df_window['volume'].values.astype('float32')
+
         base = np.stack([o, h, l, c, v], axis=1)
         base_z = _zv(base)
         lr = np.zeros_like(c); lr[1:] = np.diff(np.log1p(c))
         rng = (h - l) / (c + 1e-9)
         extra = np.stack([lr, rng], axis=1)
         extra_z = _zv(extra)
+
         def _ema(arr, n): return pd.Series(arr).ewm(span=n, adjust=False).mean().values
         ema9 = _ema(c, 9); ema21 = _ema(c, 21); ema50 = _ema(c, 50); ema200 = _ema(c, 200)
         slope21 = np.gradient(ema21); slope50 = np.gradient(ema50)
@@ -96,6 +101,7 @@ def _transform_window_for_pattern(df_window):
         roll_down = pd.Series(down).abs().ewm(alpha=1/14, adjust=False).mean().values
         rs = roll_up / (roll_down + 1e-9)
         rsi = 100.0 - (100.0 / (1.0 + rs))
+
         indicators = np.stack([ema9, ema21, ema50, ema200, slope21, slope50, rsi], axis=1)
         X_seq = np.concatenate([base_z, extra_z, _zv(indicators)], axis=1)
         X_flat = X_seq.flatten()
@@ -114,13 +120,15 @@ class HeavyDutyBacktester:
         self.INITIAL_CAPITAL = 10.0
         self.TRADING_FEES = 0.001 
         self.MAX_SLOTS = 4         
+        
         self.TARGET_COINS = [
-            'SOL/USDT'
-        ]
+            'SOL/USDT'  ]
+        
         self.force_start_date = None 
         self.force_end_date = None   
+        
         if not os.path.exists(CACHE_DIR): os.makedirs(CACHE_DIR)
-        print(f"🧪 [Backtest V140.0] Bulletproof Scalar Logic.")
+        print(f"🧪 [Backtest V141.0] Scalar Enforcement Mode.")
 
     def set_date_range(self, start_str, end_str):
         self.force_start_date = start_str
@@ -137,6 +145,7 @@ class HeavyDutyBacktester:
             current += duration_per_batch
         all_candles = []
         sem = asyncio.Semaphore(20) 
+
         async def _fetch_batch(timestamp):
             async with sem:
                 for _ in range(3): 
@@ -144,6 +153,7 @@ class HeavyDutyBacktester:
                         return await self.dm.exchange.fetch_ohlcv(sym, '1m', since=timestamp, limit=limit)
                     except: await asyncio.sleep(0.5)
                 return []
+
         chunk_size = 50
         for i in range(0, len(tasks), chunk_size):
             chunk_tasks = tasks[i:i + chunk_size]
@@ -151,6 +161,7 @@ class HeavyDutyBacktester:
             results = await asyncio.gather(*futures)
             for res in results:
                 if res: all_candles.extend(res)
+
         if not all_candles: return None
         df = pd.DataFrame(all_candles, columns=['timestamp', 'o', 'h', 'l', 'c', 'v'])
         df.drop_duplicates('timestamp', inplace=True)
@@ -159,10 +170,15 @@ class HeavyDutyBacktester:
         print(f"     ✅ Downloaded {len(df)} candles.", flush=True)
         return df.values.tolist()
 
+    # ==============================================================
+    # 🏎️ VECTORIZED INDICATORS (EXACT MATCH TO LIVE SYSTEM)
+    # ==============================================================
     def _calculate_indicators_vectorized(self, df, timeframe='1m'):
         cols = ['close', 'high', 'low', 'volume', 'open']
         for c in cols: df[c] = df[c].astype(np.float64)
         idx = df.index
+        
+        # --- TITAN FEATURES ---
         df['RSI'] = safe_ta(ta.rsi(df['close'], length=14), idx, 50)
         macd = ta.macd(df['close'])
         if macd is not None:
@@ -173,28 +189,35 @@ class HeavyDutyBacktester:
         adx = ta.adx(df['high'], df['low'], df['close'], length=14)
         if adx is not None: df['ADX'] = safe_ta(adx.iloc[:, 0], idx, 0)
         else: df['ADX'] = 0.0
-        if timeframe == '1d': df['Trend_Strong'] = np.where(df['ADX'] > 25, 1.0, 0.0)
+        if timeframe == '1d':
+            df['Trend_Strong'] = np.where(df['ADX'] > 25, 1.0, 0.0)
+
         for p in [9, 21, 50, 200]:
             ema = safe_ta(ta.ema(df['close'], length=p), idx, 0)
             df[f'EMA_{p}_dist'] = ((df['close'] / ema.replace(0, np.nan)) - 1).fillna(0)
             df[f'ema{p}'] = ema 
+
         df['ema20'] = safe_ta(ta.ema(df['close'], length=20), idx, df['close'])
+
         bb = ta.bbands(df['close'], length=20, std=2.0)
         if bb is not None:
             w = ((bb.iloc[:, 2] - bb.iloc[:, 0]) / bb.iloc[:, 1].replace(0, np.nan)).fillna(0)
             p = ((df['close'] - bb.iloc[:, 0]) / (bb.iloc[:, 2] - bb.iloc[:, 0]).replace(0, np.nan)).fillna(0)
             df['BB_w'] = w; df['BB_p'] = p; df['bb_width'] = w
         else: df['BB_w'] = 0; df['BB_p'] = 0; df['bb_width'] = 0
+        
         df['MFI'] = safe_ta(ta.mfi(df['high'], df['low'], df['close'], df['volume'], length=14), idx, 50)
         vwap = ta.vwap(df['high'], df['low'], df['close'], df['volume'])
         if vwap is not None:
             df['VWAP_dist'] = ((df['close'] / vwap.replace(0, np.nan)) - 1).fillna(0)
             df['vwap'] = vwap
         else: df['VWAP_dist'] = 0.0; df['vwap'] = df['close']
+
         df['atr'] = safe_ta(ta.atr(df['high'], df['low'], df['close'], length=14), idx, 0)
         df['atr_pct'] = (df['atr'] / df['close'].replace(0, np.nan)).fillna(0)
         df['ATR_pct'] = df['atr_pct'] 
 
+        # --- SNIPER FEATURES ---
         if timeframe == '1m':
             df['return_1m'] = df['close'].pct_change().fillna(0)
             df['return_3m'] = df['close'].pct_change(3).fillna(0)
@@ -233,14 +256,14 @@ class HeavyDutyBacktester:
             df['rv_gk'] = _z_roll(rv_gk)
             df['L_score'] = (df['vol_zscore_50'] - df['amihud'] - df['roll_spread'] - df['rv_gk'].abs() - df['vwap_dev'].abs() + df['ofi']).fillna(0)
 
+        # --- EXTRAS ---
         df['slope'] = safe_ta(ta.slope(df['close'], length=7), idx, 0)
         vol_mean = df['volume'].rolling(20).mean()
         vol_std = df['volume'].rolling(20).std().replace(0, np.nan)
         df['vol_z'] = ((df['volume'] - vol_mean) / vol_std).fillna(0)
         df['rel_vol'] = df['volume'] / (df['volume'].rolling(50).mean() + 1e-9)
         df['log_ret'] = np.log(df['close'] / df['close'].shift(1).replace(0, np.nan)).fillna(0)
-        roll_max = df['high'].rolling(50).max()
-        roll_min = df['low'].rolling(50).min()
+        roll_max = df['high'].rolling(50).max(); roll_min = df['low'].rolling(50).min()
         diff = (roll_max - roll_min).replace(0, 1e-9)
         df['fib_pos'] = ((df['close'] - roll_min) / diff).fillna(0.5)
         e20_s = df['ema20'].shift(5).replace(0, np.nan)
@@ -252,19 +275,25 @@ class HeavyDutyBacktester:
         e200 = safe_ta(ta.ema(df['close'], length=200), idx, df['close'])
         df['ema200'] = e200
         df['dist_ema200'] = ((df['close'] - e200) / e200.replace(0, np.nan)).fillna(0)
+        
         if timeframe == '1m':
             for lag in [1, 2, 3, 5, 10, 20]:
                 df[f'log_ret_lag_{lag}'] = df['log_ret'].shift(lag).fillna(0)
                 df[f'rsi_lag_{lag}'] = (df['RSI'].shift(lag) / 100.0).fillna(0.5)
                 df[f'fib_pos_lag_{lag}'] = df['fib_pos'].shift(lag).fillna(0.5)
                 df[f'volatility_lag_{lag}'] = df['volatility'].shift(lag).fillna(0)
+
         df.fillna(0, inplace=True)
         return df
 
+    # ==============================================================
+    # 🧠 CPU PROCESSING (GLOBAL INFERENCE)
+    # ==============================================================
     async def _process_data_in_memory(self, sym, candles, start_ms, end_ms):
         safe_sym = sym.replace('/', '_')
         period_suffix = f"{start_ms}_{end_ms}"
         scores_file = f"{CACHE_DIR}/{safe_sym}_{period_suffix}_scores.pkl"
+
         if os.path.exists(scores_file):
              print(f"   📂 [{sym}] Data Exists -> Skipping.")
              return
@@ -332,7 +361,8 @@ class HeavyDutyBacktester:
                     elif target_arr and feat in ['open','high','low','close','volume']: t_vecs.append(target_arr[feat][target_map])
                     else: t_vecs.append(np.zeros(len(arr_ts_1m)))
                 X_TITAN = np.column_stack(t_vecs)
-                global_titan_scores = _revive_score_distribution(titan_model.predict(xgb.DMatrix(X_TITAN, feature_names=titan_cols)))
+                preds_t = titan_model.predict(xgb.DMatrix(X_TITAN, feature_names=titan_cols))
+                global_titan_scores = _revive_score_distribution(preds_t) # ✅ Flattened inside
             except: pass
 
         # B. SNIPER (Global)
@@ -383,6 +413,7 @@ class HeavyDutyBacktester:
                 X_V2 = np.column_stack([l_log, l_rsi, l_fib, l_vol, l5_log, l5_rsi, l5_fib, l5_trd, l15_log, l15_rsi, l15_fib618, l15_trd, *lags])
                 preds = legacy_v2.predict(xgb.DMatrix(X_V2))
                 global_v2_scores = preds[:, 2] if len(preds.shape) > 1 else preds
+                global_v2_scores = global_v2_scores.flatten() # ✅ Safety Flatten
             except: pass
 
         # Filter
@@ -407,13 +438,15 @@ class HeavyDutyBacktester:
                 for idx in chunk_idxs:
                     sl_st = h_static[idx:idx+240] 
                     sl_close = sl_st[:, 6]; sl_atr = sl_st[:, 5]
-                    entry = fast_1m['close'][idx]
+                    entry = float(fast_1m['close'][idx]) # ✅ Scalar
                     dist = np.maximum(1.5 * sl_atr, entry * 0.015)
                     pnl = sl_close - entry
                     norm_pnl = pnl / dist
                     max_pnl_r = (np.maximum.accumulate(sl_close) - entry) / dist
                     atr_pct = sl_atr / sl_close
-                    zeros = np.zeros(240); time_vec = np.arange(1, 241); s_oracle = global_oracle_scores[idx]
+                    zeros = np.zeros(240); time_vec = np.arange(1, 241)
+                    s_oracle = float(global_oracle_scores[idx]) # ✅ Scalar
+                    
                     X_H = np.column_stack([
                         sl_st[:,0], sl_st[:,1], sl_st[:,2], sl_st[:,3], sl_st[:,4],
                         zeros, atr_pct, norm_pnl, max_pnl_r, zeros, zeros, time_vec, zeros,
@@ -422,22 +455,24 @@ class HeavyDutyBacktester:
                     max_hydra = 0.0; hydra_time = 0
                     try:
                         probs = hydra_models['crash'].predict_proba(X_H)[:, 1]
-                        max_hydra = np.max(probs)
+                        max_hydra = float(np.max(probs)) # ✅ Scalar
                         if max_hydra > 0.6:
                             t = np.argmax(probs)
                             hydra_time = int(fast_1m['timestamp'][idx + t])
                     except: pass
-                    max_v2 = np.max(global_v2_scores[idx:idx+240])
+                    
+                    max_v2 = float(np.max(global_v2_scores[idx:idx+240])) # ✅ Scalar
                     v2_time = 0
                     if max_v2 > 0.8:
                         t2 = np.argmax(global_v2_scores[idx:idx+240])
                         v2_time = int(fast_1m['timestamp'][idx + t2])
+                        
                     ai_results.append({
                         'timestamp': int(fast_1m['timestamp'][idx]), 
                         'symbol': sym, 'close': entry,
-                        'real_titan': global_titan_scores[idx], 
+                        'real_titan': float(global_titan_scores[idx]), # ✅ Scalar
                         'oracle_conf': s_oracle, 
-                        'sniper_score': global_sniper_scores[idx],
+                        'sniper_score': float(global_sniper_scores[idx]), # ✅ Scalar
                         'risk_hydra_crash': max_hydra, 'time_hydra_crash': hydra_time,
                         'risk_legacy_v2': max_v2, 'time_legacy_panic': v2_time,
                         'signal_type': 'BREAKOUT', 'l1_score': 50.0
@@ -468,114 +503,70 @@ class HeavyDutyBacktester:
             except: pass
         if not data: return []
         
-        # ✅ [GEM-FIX] Reset Index to avoid 'Truth value' error
+        # ✅ Reset Index to ensure aligned access
         df = pd.concat(data).sort_values('timestamp').reset_index(drop=True)
         
         ts = df['timestamp'].values
         close = df['close'].values.astype(float)
         sym = df['symbol'].values
+        u_syms = np.unique(sym); sym_map = {s: i for i, s in enumerate(u_syms)}; sym_id = np.array([sym_map[s] for s in sym])
         
-        # Map symbols to integers
-        u_syms = np.unique(sym)
-        sym_map = {s: i for i, s in enumerate(u_syms)}
-        sym_id = np.array([sym_map[s] for s in sym])
-        
-        # Extract features as pure numpy arrays (scalar safety)
+        # ✅ Explicit Scalar Conversion for Features
         oracle = df['oracle_conf'].values.astype(float)
         sniper = df['sniper_score'].values.astype(float)
         hydra = df['risk_hydra_crash'].values.astype(float)
         titan = df['real_titan'].values.astype(float)
         l1 = df['l1_score'].values.astype(float)
-        
-        # Handle Legacy (fill 0 if missing)
         legacy_v2 = df['risk_legacy_v2'].values.astype(float) if 'risk_legacy_v2' in df else np.zeros(len(df))
-        
-        # Extra: Hydra Time (for expiry check)
-        h_times = df['time_hydra_crash'].values.astype(int)
+        h_times = df['time_hydra_crash'].values.astype(np.int64)
         
         N = len(ts)
         print(f"     🚀 [System] Testing {len(combinations_batch)} configs on {N} candles...", flush=True)
         
         res = []
         for cfg in combinations_batch:
-            pos = {} 
-            log = [] 
-            bal = float(initial_capital)
-            alloc = 0.0
+            pos = {}; log = [] 
+            bal = float(initial_capital); alloc = 0.0
             
-            # Pre-calc mask (Boolean Array)
-            mask = (l1 >= cfg['l1_thresh']) & \
-                   (oracle >= cfg['oracle_thresh']) & \
-                   (sniper >= cfg['sniper_thresh']) & \
-                   (titan >= 0.55)
+            # Mask is now purely boolean (True/False)
+            mask = (l1 >= cfg['l1_thresh']) & (oracle >= cfg['oracle_thresh']) & (sniper >= cfg['sniper_thresh']) & (titan >= 0.55)
             
-            # Loop
             for i in range(N):
-                s = sym_id[i]
-                p = float(close[i])
-                curr_t = ts[i]
+                s = sym_id[i]; p = float(close[i]); curr_t = ts[i]
 
-                # 1. Exit Logic
                 if s in pos:
                     entry_p, h_risk_val, size_val, h_time_val = pos[s]
                     
-                    # Explicit Scalar bools
                     crash_hydra = bool(h_risk_val > cfg['hydra_thresh'])
-                    
-                    # Logic: If current time > crash time prediction, signal is stale?
-                    # Or if prediction was for a future time? 
-                    # Assuming h_time_val is the timestamp of predicted crash
                     time_match = bool(h_time_val > 0 and curr_t >= h_time_val)
-                    
-                    # Legacy Logic (Global array check)
-                    # Note: Legacy array corresponds to candle index, but here we iterate sorted time
-                    # We need to trust the backtest signal 'risk_legacy_v2' is aligned.
-                    # Yes, it comes from df row 'i'.
                     panic_legacy = bool(legacy_v2[i] > cfg['legacy_thresh'])
-                    
                     pnl = (p - entry_p) / entry_p
                     
-                    # Combined Exit
-                    # Exit if: Hydra Crash AND Time Match OR Legacy Panic OR TP/SL
-                    should_exit = (crash_hydra and time_match) or panic_legacy or (pnl > 0.04) or (pnl < -0.02)
-                    
-                    if should_exit:
+                    if (crash_hydra and time_match) or panic_legacy or pnl > 0.04 or pnl < -0.02:
                         realized = pnl - (fees_pct * 2)
                         bal += size_val * (1.0 + realized)
                         alloc -= size_val
                         del pos[s]
                         log.append({'pnl': realized})
 
-                # 2. Entry Logic
-                # Use scalar boolean from mask
                 if len(pos) < max_slots and bool(mask[i]):
                     if s not in pos and bal >= 5.0:
                         size = min(10.0, bal * 0.98)
-                        # Store: Entry, HydraRisk, Size, HydraTime
                         pos[s] = (p, hydra[i], size, h_times[i])
-                        bal -= size
-                        alloc += size
+                        bal -= size; alloc += size
             
             final_bal = bal + alloc
             profit = final_bal - initial_capital
-            
-            # Stats
             tot = len(log)
             winning = [x for x in log if x['pnl'] > 0]
             losing = [x for x in log if x['pnl'] <= 0]
-            
-            win_count = len(winning)
-            loss_count = len(losing)
+            win_count = len(winning); loss_count = len(losing)
             win_rate = (win_count/tot*100) if tot > 0 else 0.0
-            
             avg_win = np.mean([x['pnl'] for x in winning]) if winning else 0.0
             avg_loss = np.mean([x['pnl'] for x in losing]) if losing else 0.0
-            
             gross_p = sum([x['pnl'] for x in winning])
             gross_l = abs(sum([x['pnl'] for x in losing]))
             profit_factor = (gross_p / gross_l) if gross_l > 0 else 99.9
-
-            # Streaks
             max_win_s = 0; max_loss_s = 0; curr_w = 0; curr_l = 0
             for t in log:
                 if t['pnl'] > 0:
@@ -592,9 +583,7 @@ class HeavyDutyBacktester:
                 'avg_win': avg_win, 'avg_loss': avg_loss,
                 'max_win_streak': max_win_s, 'max_loss_streak': max_loss_s,
                 'profit_factor': profit_factor,
-                'consensus_agreement_rate': 0.0,
-                'high_consensus_win_rate': 0.0,
-                'high_consensus_avg_pnl': 0.0
+                'consensus_agreement_rate': 0.0, 'high_consensus_win_rate': 0.0, 'high_consensus_avg_pnl': 0.0
             })
         return res
 
@@ -602,29 +591,23 @@ class HeavyDutyBacktester:
         await self.generate_truth_data()
         oracle_r = np.linspace(0.4, 0.7, 3); sniper_r = np.linspace(0.4, 0.7, 3)
         hydra_r = [0.85, 0.95]; l1_r = [10.0]
-        
         combos = []
         for o, s, h, l1 in itertools.product(oracle_r, sniper_r, hydra_r, l1_r):
             combos.append({
                 'w_titan': 0.4, 'w_struct': 0.3, 'thresh': l1, 'l1_thresh': l1, 
                 'oracle_thresh': o, 'sniper_thresh': s, 'hydra_thresh': h, 'legacy_thresh': 0.95
             })
-            
         files = glob.glob(os.path.join(CACHE_DIR, "*.pkl"))
         results_list = self._worker_optimize(combos, files, self.INITIAL_CAPITAL, self.TRADING_FEES, self.MAX_SLOTS)
         if not results_list: return None, {'net_profit': 0.0, 'win_rate': 0.0}
-
         results_list.sort(key=lambda x: x['net_profit'], reverse=True)
         best = results_list[0]
-
-        # Auto-Diagnosis
         diag = []
         if best['total_trades'] > 2000 and best['net_profit'] < 10: diag.append("⚠️ Overtrading")
         if best['win_rate'] > 55 and best['net_profit'] < 0: diag.append("⚠️ Fee Burn")
         if abs(best['avg_loss']) > best['avg_win']: diag.append("⚠️ Risk/Reward Inversion")
         if best['max_loss_streak'] > 10: diag.append("⚠️ Consecutive Loss Risk")
         if not diag: diag.append("✅ System Healthy")
-
         print("\n" + "="*60)
         print(f"🏆 CHAMPION REPORT [{target_regime}]:")
         print(f"   💰 Final Balance:   ${best['final_balance']:,.2f}")
@@ -647,21 +630,19 @@ class HeavyDutyBacktester:
         return best['config'], best
 
 async def run_strategic_optimization_task():
-    print("\n🧪 [STRATEGIC BACKTEST] Full Spectrum Mode...")
+    print("\n🧪 [STRATEGIC BACKTEST] Scalar Enforcement Mode...")
     r2 = R2Service(); dm = DataManager(None, None, r2); proc = MLProcessor(dm)
     try:
         await dm.initialize(); await proc.initialize()
         if proc.guardian_hydra: proc.guardian_hydra.set_silent_mode(True)
         hub = AdaptiveHub(r2); await hub.initialize()
         opt = HeavyDutyBacktester(dm, proc)
-        
         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"}
             ]
-        
         for s in scenarios:
             opt.set_date_range(s["start"], s["end"])
             best_cfg, best_stats = await opt.run_optimization(s["regime"])