Update ml_engine/guardian_hydra.py

ml_engine/guardian_hydra.py

@@ -10,17 +10,15 @@ import sys
 
 class GuardianHydra:
     """
-    GuardianHydra V1.3 (X-RAY DIAGNOSTIC MODE)
-    - Puts everything under a microscope.
-    - Prints INPUT Data (OHLCV sample).
-    - Prints CALCULATED Features (RSI, ATR, etc).
-    - Prints RAW MODEL Output (Array).
+    GuardianHydra V1.4 (Configurable Verbosity)
+    - Added set_silent_mode() to suppress logs during Backtesting.
     """
     def __init__(self, model_dir):
         self.model_dir = model_dir
         self.initialized = False
         self.models = {}
         self.feature_cols = []
+        self.verbose = True  # ✅ Default to True for Live System
         
         self.smoothing_buffer = defaultdict(lambda: {
             'crash': deque(maxlen=3), 
@@ -29,43 +27,48 @@ class GuardianHydra:
         })
 
         self.ATR_PERIOD = 14
-        print("🐲 [Hydra X-RAY] Instance Created. Waiting for data...")
+        # ✅ Silent check
+        if self.verbose: print("🐲 [Hydra X-RAY] Instance Created. Waiting for data...")
+
+    def set_silent_mode(self, silent=True):
+        """ ✅ Control Logging Output (True = No Logs, False = X-RAY Logs) """
+        self.verbose = not silent
 
     def initialize(self):
-        print(f"🐲 [Hydra X-RAY] Loading from: {self.model_dir}")
+        if self.verbose: print(f"🐲 [Hydra X-RAY] Loading from: {self.model_dir}")
         
         if not os.path.exists(self.model_dir):
-            print(f"❌ [FATAL] Directory missing: {self.model_dir}")
+            if self.verbose: print(f"❌ [FATAL] Directory missing: {self.model_dir}")
             return False
 
         try:
             # 1. Load Features
             feat_path = os.path.join(self.model_dir, "hydra_features_list.pkl")
             if not os.path.exists(feat_path):
-                print(f"❌ Feature list missing: {feat_path}")
+                if self.verbose: print(f"❌ Feature list missing: {feat_path}")
                 return False
             self.feature_cols = joblib.load(feat_path)
-            print(f"✅ Features List Loaded ({len(self.feature_cols)} items)")
+            if self.verbose: print(f"✅ Features List Loaded ({len(self.feature_cols)} items)")
 
             # 2. Load Models (RAW)
             heads = ['crash', 'giveback', 'stagnation']
             for h in heads:
                 model_path = os.path.join(self.model_dir, f"hydra_head_{h}_raw.json")
                 if not os.path.exists(model_path):
-                    print(f"❌ Model missing: {model_path}")
+                    if self.verbose: print(f"❌ Model missing: {model_path}")
                     return False
                 
                 clf = xgb.XGBClassifier()
                 clf.load_model(model_path)
                 self.models[h] = clf
-                print(f"✅ Loaded Head: {h}")
+                if self.verbose: print(f"✅ Loaded Head: {h}")
 
             self.initialized = True
-            print(f"✅ [Hydra X-RAY] System Ready.")
+            if self.verbose: print(f"✅ [Hydra X-RAY] System Ready.")
             return True
 
         except Exception as e:
-            print(f"❌ [Hydra Init Error] {e}")
+            if self.verbose: print(f"❌ [Hydra Init Error] {e}")
             traceback.print_exc()
             return False
 
@@ -73,17 +76,18 @@ class GuardianHydra:
         try:
             # 1. Check Raw Data Inputs
             if not ohlcv_1m or len(ohlcv_1m) < 1:
-                print("⚠️ [X-RAY] 1m Data is EMPTY!")
+                if self.verbose: print("⚠️ [X-RAY] 1m Data is EMPTY!")
                 return None
                 
             df_1m = pd.DataFrame(ohlcv_1m, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
             
             # [DIAGNOSTIC 1] Print Input Sample
             last_close = df_1m['close'].iloc[-1]
-            print(f"🔍 [Input Check] Last Close: {last_close} | Candles: {len(df_1m)}")
+            if self.verbose: 
+                print(f"🔍 [Input Check] Last Close: {last_close} | Candles: {len(df_1m)}")
             
             if len(df_1m) < 50: 
-                print(f"⚠️ [X-RAY] Not enough history: {len(df_1m)} < 50")
+                if self.verbose: print(f"⚠️ [X-RAY] Not enough history: {len(df_1m)} < 50")
                 return None
             
             # 2. Indicator Calculation
@@ -94,7 +98,7 @@ class GuardianHydra:
             last_rsi = df_1m['rsi'].iloc[-1]
             last_atr = df_1m['atr'].iloc[-1]
             
-            if pd.isna(last_rsi) or pd.isna(last_atr):
+            if (pd.isna(last_rsi) or pd.isna(last_atr)) and self.verbose:
                 print(f"⚠️ [X-RAY] Indicators are NaN! RSI: {last_rsi}, ATR: {last_atr}")
             
             # ... rest of calculations ...
@@ -131,17 +135,12 @@ class GuardianHydra:
             
             duration_mins = trade_context.get('time_in_trade_mins', 10)
             
-            # [GEM-FIX] Extract Highest Price & Calculate Max PnL
-            # This enables the "Giveback" model to see profit retracements.
             highest_price = float(trade_context.get('highest_price', entry_price))
-            if highest_price < entry_price: highest_price = entry_price # Safety
+            if highest_price < entry_price: highest_price = entry_price
             
             max_pnl_amt = highest_price - entry_price
             max_pnl_r = max_pnl_amt / sl_dist_unit if sl_dist_unit > 0 else 0.0
 
-            # [DIAGNOSTIC 3] Context Check
-            # print(f"🔍 [Context Check] Entry: {entry_price} | Current: {last_close} | PnL R: {norm_pnl_r:.2f}")
-
             # Assemble Vector
             feat_dict = {
                 'rsi_1m': last_rsi,
@@ -152,7 +151,7 @@ class GuardianHydra:
                 'dist_ema20_1h': dist_ema20_1h,
                 'atr_pct': atr_val / last_close,
                 'norm_pnl_r': norm_pnl_r,
-                'max_pnl_r': max_pnl_r, # [GEM-FIX] Now uses real calculated value
+                'max_pnl_r': max_pnl_r, 
                 'dist_tp_atr': 0.0,
                 'dist_sl_atr': 0.0,
                 'time_in_trade': float(duration_mins),
@@ -164,22 +163,22 @@ class GuardianHydra:
             
             vector = pd.DataFrame([feat_dict])
             
-            # Fill missing columns
             for col in self.feature_cols:
                 if col not in vector.columns:
                     vector[col] = 0.0
             
-            # [DIAGNOSTIC 4] Check Final Vector for NaNs
             if vector.isnull().values.any():
-                print("⚠️ [X-RAY] Final Vector contains NaNs! Model will fail or output 0.")
-                print(vector.iloc[0].to_dict())
-                vector = vector.fillna(0) # Force fix
+                if self.verbose:
+                    print("⚠️ [X-RAY] Final Vector contains NaNs! Model will fail or output 0.")
+                    print(vector.iloc[0].to_dict())
+                vector = vector.fillna(0) 
             
             return vector[self.feature_cols].astype(float)
 
         except Exception as e:
-            print(f"❌ [X-RAY] Feature Error: {e}")
-            traceback.print_exc()
+            if self.verbose: 
+                print(f"❌ [X-RAY] Feature Error: {e}")
+                traceback.print_exc()
             return None
 
     def analyze_position(self, symbol, ohlcv_1m, ohlcv_5m, ohlcv_15m, trade_data):
@@ -187,40 +186,32 @@ class GuardianHydra:
             return {'action': 'HOLD', 'reason': 'Not Init'}
 
         try:
-            # Time calculation
             features = self._engineer_features(ohlcv_1m, ohlcv_5m, ohlcv_15m, trade_data)
             
             if features is None:
-                print(f"🚫 [X-RAY] {symbol}: Feature Engineering Failed.")
+                if self.verbose: print(f"🚫 [X-RAY] {symbol}: Feature Engineering Failed.")
                 return {'action': 'HOLD', 'reason': 'Feat Fail'}
 
             probs = {}
-            print(f"🔬 [X-RAY] Predicting for {symbol}...")
+            if self.verbose: print(f"🔬 [X-RAY] Predicting for {symbol}...")
             
             for h in ['crash', 'giveback', 'stagnation']:
-                # Predict
                 try:
-                    # XGBoost predict_proba returns [[prob_0, prob_1]]
                     full_pred = self.models[h].predict_proba(features)
-                    
-                    # [DIAGNOSTIC 5] Print Full Prediction Array
-                    # print(f"   👉 {h.upper()} Raw Array: {full_pred}")
-                    
                     raw_prob = full_pred[0][1]
                     probs[h] = raw_prob
                     
-                    if raw_prob > 0.0:
+                    if raw_prob > 0.0 and self.verbose:
                         print(f"   🔥 {h.upper()} Non-Zero Prob: {raw_prob:.4f}")
                     
                 except Exception as e:
-                    print(f"   ❌ Error predicting {h}: {e}")
+                    if self.verbose: print(f"   ❌ Error predicting {h}: {e}")
                     probs[h] = 0.0
 
-            # Pass-through logic (Simplified for test)
             return self._pkg('HOLD', 0.0, "X-RAY DIAGNOSTIC", probs)
 
         except Exception as e:
-            print(f"❌ [X-RAY] Analyze Error: {e}")
+            if self.verbose: print(f"❌ [X-RAY] Analyze Error: {e}")
             return {'action': 'HOLD', 'reason': 'Error'}
 
     def _pkg(self, action, conf, reason, probs):