Update ml_engine/titan_engine.py

ml_engine/titan_engine.py

@@ -1,5 +1,5 @@
 # ml_engine/titan_engine.py
-# (V1.2 - Robust indicators for short daily history)
+# V1.4 — No ambiguous Series truth, fully safe TA, aligned Series
 
 import os
 import joblib
@@ -8,45 +8,56 @@ import pandas as pd
 import pandas_ta as ta
 import xgboost as xgb
 
-def _safe_series_like(ref_index, fill=np.nan):
-    return pd.Series(index=ref_index, data=fill, dtype="float64")
 
-def _safe_ema(s: pd.Series, length: int) -> pd.Series:
-    try:
-        ema = ta.ema(s, length=length)
-        if ema is None:
-            return _safe_series_like(s.index)
-        return ema
-    except Exception:
-        return _safe_series_like(s.index)
+# ===================== Utils =====================
 
-def _safe_adx(h, l, c):
-    try:
-        adx = ta.adx(h, l, c)
-        if adx is None or adx.empty:
-            return None
-        return adx
-    except Exception:
-        return None
+def _series_like(index, fill=np.nan) -> pd.Series:
+    return pd.Series(index=index, data=fill, dtype="float64")
 
-def _safe_macd(c):
+def _as_float_series(obj, index, fill=0.0) -> pd.Series:
+    """
+    يحول أي ناتج (Series/DataFrame/ndarray/None) إلى Series float مُحاذاة للفهرس.
+    لا يستخدم أي تحقق منطقي مبهم.
+    """
+    if obj is None:
+        return _series_like(index, fill)
     try:
-        m = ta.macd(c)
-        if m is None or m.empty:
-            return None
-        return m
+        if isinstance(obj, pd.DataFrame):
+            # خذ العمود الأول فقط بشكل صريح
+            s = obj.iloc[:, 0]
+        elif isinstance(obj, pd.Series):
+            s = obj
+        else:
+            # أي نوع آخر إلى Series
+            s = pd.Series(obj, index=index)
+        s = s.astype("float64")
+        # أعِد المحاذاة أولاً ثم الاستبدال/الملء
+        s = s.reindex(index)
+        s = s.replace([np.inf, -np.inf], np.nan).fillna(fill)
+        return s
     except Exception:
-        return None
+        return _series_like(index, fill)
 
-def _safe_bbands(c, length=20, std=2.0):
+def _safe_df(obj):
+    """يعيد DataFrame صالح أو None دون أي تقييم منطقي."""
+    if obj is None:
+        return None
     try:
-        bb = ta.bbands(c, length=length, std=std)
-        if bb is None or bb.empty:
-            return None
-        return bb
+        if isinstance(obj, pd.DataFrame) and obj.shape[0] > 0 and obj.shape[1] > 0:
+            return obj
+        return None
     except Exception:
         return None
 
+def _safe_divide(num: pd.Series, den: pd.Series, fill=0.0) -> pd.Series:
+    den = den.copy()
+    den = den.replace(0, np.nan)
+    out = num / den
+    return _as_float_series(out, num.index, fill=fill)
+
+
+# ================= Titan Engine ==================
+
 class TitanEngine:
     def __init__(self, model_dir="ml_models/layer2"):
         self.model_path = os.path.join(model_dir, "Titan_XGB_V1.json")
@@ -69,94 +80,120 @@ class TitanEngine:
         except Exception as e:
             print(f"❌ [Titan] خطأ أثناء التهيئة: {e}", flush=True)
 
-    def apply_inverted_pyramid(self, df_in, tf):
-        df = df_in.copy().sort_values('timestamp').reset_index(drop=True)
-        # فهرس زمني لسهولة عمل pandas_ta
-        df = df.set_index(pd.DatetimeIndex(pd.to_datetime(df['timestamp'], unit='ms')))
-        close = df['close']; high = df['high']; low = df['low']; vol = df['volume']
+    # --------- Features per timeframe, no boolean checks ---------
+
+    def _feat_5m_15m(self, df: pd.DataFrame) -> pd.DataFrame:
+        idx = pd.DatetimeIndex(pd.to_datetime(df['timestamp'], unit='ms'))
+        df = df.set_index(idx)
+
+        c = df['close']; h = df['high']; l = df['low']; v = df['volume']
+
+        rsi = ta.rsi(c, length=14)
+        df['RSI'] = _as_float_series(rsi, df.index, fill=50)
+
+        macd = _safe_df(ta.macd(c))
+        if macd is not None and macd.shape[1] >= 2:
+            # عمود 0: خط MACD, عمود 1: الهستوغرام غالباً
+            df['MACD']   = _as_float_series(macd.iloc[:, 0], df.index, fill=0)
+            df['MACD_h'] = _as_float_series(macd.iloc[:, 1], df.index, fill=0)
+        else:
+            df['MACD']   = _series_like(df.index, 0)
+            df['MACD_h'] = _series_like(df.index, 0)
+
+        cci = ta.cci(h, l, c, length=20)
+        df['CCI'] = _as_float_series(cci, df.index, fill=0)
+
+        adx = _safe_df(ta.adx(h, l, c))
+        if adx is not None:
+            df['ADX'] = _as_float_series(adx.iloc[:, 0], df.index, fill=0)
+        else:
+            df['ADX'] = _series_like(df.index, 0)
+
+        for p in [9, 21, 50, 200]:
+            ema = _as_float_series(ta.ema(c, length=p), df.index, fill=np.nan)
+            df[f'EMA_{p}_dist'] = _as_float_series(_safe_divide(c, ema, fill=0) - 1.0, df.index, fill=0)
+
+        bb = _safe_df(ta.bbands(c, length=20, std=2.0))
+        if bb is not None and bb.shape[1] >= 3:
+            lower = _as_float_series(bb.iloc[:, 0], df.index, fill=np.nan)
+            mid   = _as_float_series(bb.iloc[:, 1], df.index, fill=np.nan)
+            upper = _as_float_series(bb.iloc[:, 2], df.index, fill=np.nan)
+            width = _safe_divide(upper - lower, mid, fill=0)
+            pos   = _safe_divide(c - lower, (upper - lower), fill=0)
+            df['BB_w'] = _as_float_series(width, df.index, fill=0)
+            df['BB_p'] = _as_float_series(pos, df.index, fill=0)
+        else:
+            df['BB_w'] = _series_like(df.index, 0)
+            df['BB_p'] = _series_like(df.index, 0)
+
+        vwap = _as_float_series(ta.vwap(h, l, c, v), df.index, fill=np.nan)
+        df['VWAP_dist'] = _as_float_series(_safe_divide(c, vwap, fill=0) - 1.0, df.index, fill=0)
 
-        # مؤشرات عامة آمنة
-        def safe_fill(s):
-            return s.replace([np.inf, -np.inf], np.nan).astype("float64")
+        return df.reset_index(drop=True)
 
-        try:
-            if tf in ['5m', '15m']:
-                df['RSI'] = safe_fill(ta.rsi(close, length=14) or _safe_series_like(close.index)).fillna(50)
-                macd = _safe_macd(close)
-                if macd is not None:
-                    df['MACD']   = safe_fill(macd.iloc[:, 0]).fillna(0)
-                    df['MACD_h'] = safe_fill(macd.iloc[:, 1]).fillna(0)
-                else:
-                    df['MACD'] = _safe_series_like(close.index, 0)
-                    df['MACD_h'] = _safe_series_like(close.index, 0)
+    def _feat_1h_4h(self, df: pd.DataFrame) -> pd.DataFrame:
+        idx = pd.DatetimeIndex(pd.to_datetime(df['timestamp'], unit='ms'))
+        df = df.set_index(idx)
 
-                df['CCI'] = safe_fill(ta.cci(high, low, close, length=20) or _safe_series_like(close.index)).fillna(0)
+        c = df['close']; h = df['high']; l = df['low']
 
-                adx = _safe_adx(high, low, close)
-                if adx is not None:
-                    df['ADX'] = safe_fill(adx.iloc[:, 0]).fillna(0)
-                else:
-                    df['ADX'] = _safe_series_like(close.index, 0)
+        rsi = ta.rsi(c, length=14)
+        df['RSI'] = _as_float_series(rsi, df.index, fill=50)
 
-                for p in [9, 21, 50, 200]:
-                    ema = _safe_ema(close, p)
-                    df[f'EMA_{p}_dist'] = safe_fill((close / ema) - 1).fillna(0)
+        macd = _safe_df(ta.macd(c))
+        if macd is not None and macd.shape[1] >= 2:
+            df['MACD_h'] = _as_float_series(macd.iloc[:, 1], df.index, fill=0)
+        else:
+            df['MACD_h'] = _series_like(df.index, 0)
 
-                bb = _safe_bbands(close, length=20, std=2.0)
-                if bb is not None:
-                    mid = bb.iloc[:, 1]
-                    df['BB_w'] = safe_fill((bb.iloc[:, 2] - bb.iloc[:, 0]) / (mid.replace(0, np.nan))).fillna(0)
-                    df['BB_p'] = safe_fill((close - bb.iloc[:, 0]) / ((bb.iloc[:, 2] - bb.iloc[:, 0]).replace(0, np.nan))).fillna(0)
-                else:
-                    df['BB_w'] = _safe_series_like(close.index, 0)
-                    df['BB_p'] = _safe_series_like(close.index, 0)
-
-                try:
-                    vwap = ta.vwap(high, low, close, vol)
-                    if vwap is None:
-                        vwap = _safe_series_like(close.index)
-                except Exception:
-                    vwap = _safe_series_like(close.index)
-                df['VWAP_dist'] = safe_fill((close / vwap) - 1).fillna(0)
-
-            elif tf in ['1h', '4h']:
-                df['RSI'] = safe_fill(ta.rsi(close, length=14) or _safe_series_like(close.index)).fillna(50)
-                macd = _safe_macd(close)
-                if macd is not None:
-                    df['MACD_h'] = safe_fill(macd.iloc[:, 1]).fillna(0)
-                else:
-                    df['MACD_h'] = _safe_series_like(close.index, 0)
-
-                ema50  = _safe_ema(close, 50)
-                ema200 = _safe_ema(close, 200)
-                df['EMA_50_dist']  = safe_fill((close / ema50) - 1).fillna(0)
-                df['EMA_200_dist'] = safe_fill((close / ema200) - 1).fillna(0)
-
-                try:
-                    atr = ta.atr(high, low, close, length=14)
-                    if atr is None:
-                        atr = _safe_series_like(close.index, 0)
-                except Exception:
-                    atr = _safe_series_like(close.index, 0)
-                df['ATR_pct'] = safe_fill(atr / close.replace(0, np.nan)).fillna(0)
-
-            elif tf == '1d':
-                df['RSI'] = safe_fill(ta.rsi(close, length=14) or _safe_series_like(close.index)).fillna(50)
-                ema200 = _safe_ema(close, 200)  # قد يكون كله NaN عند 60 شمعة، وهذا مقبول
-                df['EMA_200_dist'] = safe_fill((close / ema200) - 1).fillna(0)
-
-                adx = _safe_adx(high, low, close)
-                if adx is not None:
-                    adx_main = safe_fill(adx.iloc[:, 0]).fillna(0)
-                    df['Trend_Strong'] = (adx_main > 25).astype(int)
-                else:
-                    df['Trend_Strong'] = 0
+        ema50  = _as_float_series(ta.ema(c, length=50),  df.index, fill=np.nan)
+        ema200 = _as_float_series(ta.ema(c, length=200), df.index, fill=np.nan)
+        df['EMA_50_dist']  = _as_float_series(_safe_divide(c, ema50,  fill=0) - 1.0, df.index, fill=0)
+        df['EMA_200_dist'] = _as_float_series(_safe_divide(c, ema200, fill=0) - 1.0, df.index, fill=0)
+
+        atr = _as_float_series(ta.atr(h, l, c, length=14), df.index, fill=0)
+        df['ATR_pct'] = _as_float_series(_safe_divide(atr, c, fill=0), df.index, fill=0)
+
+        return df.reset_index(drop=True)
+
+    def _feat_1d(self, df: pd.DataFrame) -> pd.DataFrame:
+        idx = pd.DatetimeIndex(pd.to_datetime(df['timestamp'], unit='ms'))
+        df = df.set_index(idx)
+
+        c = df['close']; h = df['high']; l = df['low']
+
+        rsi = ta.rsi(c, length=14)
+        df['RSI'] = _as_float_series(rsi, df.index, fill=50)
 
+        ema200 = _as_float_series(ta.ema(c, length=200), df.index, fill=np.nan)
+        df['EMA_200_dist'] = _as_float_series(_safe_divide(c, ema200, fill=0) - 1.0, df.index, fill=0)
+
+        adx = _safe_df(ta.adx(h, l, c))
+        if adx is not None:
+            adx_main = _as_float_series(adx.iloc[:, 0], df.index, fill=0)
+            df['Trend_Strong'] = (adx_main > 25).astype(int)
+        else:
+            df['Trend_Strong'] = _series_like(df.index, 0)
+
+        return df.reset_index(drop=True)
+
+    def apply_inverted_pyramid(self, df_in: pd.DataFrame, tf: str) -> pd.DataFrame:
+        # لا try/except شامل. أي خطأ يُطبع ويُرجع DataFrame كما هو لضمان عدم توقف الباقي.
+        try:
+            df_in = df_in.copy().sort_values('timestamp').reset_index(drop=True)
+            if tf in ('5m', '15m'):
+                return self._feat_5m_15m(df_in)
+            if tf in ('1h', '4h'):
+                return self._feat_1h_4h(df_in)
+            if tf == '1d':
+                return self._feat_1d(df_in)
+            # افتراضي: لا شيء
+            return df_in
         except Exception as e:
             print(f"⚠️ [Titan] مؤشر فشل على tf={tf}: {e}", flush=True)
+            return df_in
 
-        # ارجع إلى شكل طبيعي
-        return df.reset_index(drop=True)
+    # ---------------- Prediction -------------------
 
     def predict(self, ohlcv_data: dict) -> dict:
         if not self.initialized or not self.model:
@@ -165,50 +202,49 @@ class TitanEngine:
         try:
             processed_tfs = {}
             for tf, data in ohlcv_data.items():
-                if not data:
+                if data is None:
                     continue
                 if isinstance(data, list):
-                    df = pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
+                    df = pd.DataFrame(data, columns=['timestamp','open','high','low','close','volume'])
                 else:
                     df = data.copy()
-                df = self.apply_inverted_pyramid(df, tf)
-                processed_tfs[tf] = df
+                processed_tfs[tf] = self.apply_inverted_pyramid(df, tf)
 
-            if '5m' not in processed_tfs:
+            if '5m' not in processed_tfs or processed_tfs['5m'].empty:
                 print("⚠️ [Titan] Missing 5m base timeframe", flush=True)
                 return {'score': 0.0, 'error': 'Missing 5m base timeframe'}
 
-            latest_5m = processed_tfs['5m'].iloc[-1:].copy()
-            latest_ts = latest_5m['timestamp'].iloc[0]
-            base_row = latest_5m.add_prefix('5m_').rename(columns={'5m_timestamp': 'timestamp'})
+            base5 = processed_tfs['5m'].iloc[-1:].copy()
+            latest_ts = int(base5['timestamp'].iloc[0])
+            row = base5.add_prefix('5m_').rename(columns={'5m_timestamp': 'timestamp'})
 
-            for tf, df in processed_tfs.items():
-                if tf == '5m':
+            # دمج آخر صف متاح من كل إطار ≤ latest_ts
+            for tf, tdf in processed_tfs.items():
+                if tf == '5m' or tdf is None or tdf.empty:
                     continue
-                relevant_row = df[df['timestamp'] <= latest_ts].iloc[-1:].copy()
-                if relevant_row.empty:
+                rel = tdf[tdf['timestamp'] <= latest_ts]
+                if rel.empty:
                     continue
-                cols = [c for c in relevant_row.columns if c not in ['timestamp','open','high','low','close','volume']]
-                for col in cols:
-                    base_row[f"{tf}_{col}"] = relevant_row[col].values[0]
-
-            # ترتيب الميزات
-            input_data = []
-            for feat in self.feature_names:
-                val = base_row.get(feat, np.nan)
-                if isinstance(val, (pd.Series, np.ndarray)):
-                    val = val.iloc[0] if len(val) > 0 else np.nan
+                last = rel.iloc[-1:]
+                cols = [c for c in last.columns if c not in ['timestamp','open','high','low','close','volume']]
+                for c in cols:
+                    row[f"{tf}_{c}"] = last[c].values[0]
+
+            # بناء متجه الميزات بنفس الترتيب
+            feats = []
+            for name in self.feature_names:
+                val = row.get(name, np.nan)
+                if isinstance(val, pd.Series):
+                    val = val.iloc[0] if len(val) else np.nan
                 if pd.isna(val):
                     val = 0.0
-                input_data.append(float(val))
-
-            dtest = xgb.DMatrix([input_data], feature_names=self.feature_names)
-            prediction = float(self.model.predict(dtest)[0])
+                feats.append(float(val))
 
-            print(f"[DEBUG][Titan] pred={prediction:.3f}", flush=True)
-            return {'score': prediction, 'timestamp': int(latest_ts), 'status': 'OK'}
+            dtest = xgb.DMatrix([feats], feature_names=self.feature_names)
+            pred = float(self.model.predict(dtest)[0])
+            print(f"[DEBUG][Titan] pred={pred:.3f}", flush=True)
+            return {'score': pred, 'timestamp': latest_ts, 'status': 'OK'}
 
         except Exception as e:
-            import traceback
-            traceback.print_exc()
+            import traceback; traceback.print_exc()
             return {'score': 0.0, 'error': str(e)}