Update ml_engine/titan_engine.py

ml_engine/titan_engine.py

@@ -1,5 +1,5 @@
 # ml_engine/titan_engine.py
-# V1.4 — No ambiguous Series truth, fully safe TA, aligned Series
+# (V1.0 - Titan Inference Engine)
 
 import os
 import joblib
@@ -7,56 +7,7 @@ import numpy as np
 import pandas as pd
 import pandas_ta as ta
 import xgboost as xgb
-
-
-# ===================== Utils =====================
-
-def _series_like(index, fill=np.nan) -> pd.Series:
-    return pd.Series(index=index, data=fill, dtype="float64")
-
-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:
-        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 _series_like(index, fill)
-
-def _safe_df(obj):
-    """يعيد DataFrame صالح أو None دون أي تقييم منطقي."""
-    if obj is None:
-        return None
-    try:
-        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 ==================
+import json
 
 class TitanEngine:
     def __init__(self, model_dir="ml_models/layer2"):
@@ -67,184 +18,135 @@ class TitanEngine:
         self.initialized = False
 
     async def initialize(self):
-        print(f"🛡️ [Titan] تهيئة من {self.model_path}", flush=True)
+        """تحميل النموذج وقائمة الميزات من القرص"""
+        print(f"🛡️ [Titan] جاري تهيئة المحرك من {self.model_path}...")
         try:
             if os.path.exists(self.model_path) and os.path.exists(self.features_path):
+                # تحميل نموذج XGBoost
                 self.model = xgb.Booster()
                 self.model.load_model(self.model_path)
+                
+                # تحميل قائمة الميزات لضمان الترتيب الصحيح
                 self.feature_names = joblib.load(self.features_path)
+                
                 self.initialized = True
-                print(f"✅ [Titan] جاهز. ميزات: {len(self.feature_names)}", flush=True)
+                print(f"✅ [Titan] تم التحميل بنجاح. جاهز بـ {len(self.feature_names)} ميزة.")
             else:
-                print("❌ [Titan] ملفات النموذج مفقودة!", flush=True)
+                print(f"❌ [Titan] ملفات النموذج مفقودة!")
         except Exception as e:
-            print(f"❌ [Titan] خطأ أثناء التهيئة: {e}", flush=True)
-
-    # --------- 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)
-
-        return df.reset_index(drop=True)
-
-    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)
-
-        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)
-
-        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)
-
-        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)
+            print(f"❌ [Titan] خطأ فادح أثناء التهيئة: {e}")
+
+    def apply_inverted_pyramid(self, df, tf):
+        """نفس منطق هندسة الميزات المستخدم في التدريب تماماً"""
+        df = df.copy().sort_values('timestamp').reset_index(drop=True)
+        # تعيين الفهرس للسهولة في pandas_ta
+        df = df.set_index(pd.DatetimeIndex(pd.to_datetime(df['timestamp'], unit='ms')))
+        
+        # --- المستوى 1: دقيق (5m, 15m) ---
+        if tf in ['5m', '15m']:
+            df['RSI'] = ta.rsi(df['close'], length=14)
+            df['MACD'] = ta.macd(df['close']).iloc[:, 0]
+            df['MACD_h'] = ta.macd(df['close']).iloc[:, 1]
+            df['CCI'] = ta.cci(df['high'], df['low'], df['close'], length=20)
+            df['ADX'] = ta.adx(df['high'], df['low'], df['close'], length=14).iloc[:, 0]
+            for p in [9, 21, 50, 200]:
+                ema = ta.ema(df['close'], length=p)
+                df[f'EMA_{p}_dist'] = (df['close'] / ema) - 1
+            bb = ta.bbands(df['close'], length=20, std=2.0)
+            df['BB_w'] = (bb.iloc[:, 2] - bb.iloc[:, 0]) / bb.iloc[:, 1]
+            df['BB_p'] = (df['close'] - bb.iloc[:, 0]) / (bb.iloc[:, 2] - bb.iloc[:, 0])
+            df['MFI'] = ta.mfi(df['high'], df['low'], df['close'], df['volume'], length=14)
+            vwap = ta.vwap(df['high'], df['low'], df['close'], df['volume'])
+            df['VWAP_dist'] = (df['close'] / vwap) - 1
+
+        # --- المستوى 2: تكتيكي (1h, 4h) ---
+        elif tf in ['1h', '4h']:
+            df['RSI'] = ta.rsi(df['close'], length=14)
+            df['MACD_h'] = ta.macd(df['close']).iloc[:, 1]
+            df['EMA_50_dist'] = (df['close'] / ta.ema(df['close'], length=50)) - 1
+            df['EMA_200_dist'] = (df['close'] / ta.ema(df['close'], length=200)) - 1
+            df['ATR_pct'] = ta.atr(df['high'], df['low'], df['close'], length=14) / df['close']
+
+        # --- المستوى 3: استراتيجي (1d) ---
+        elif tf == '1d':
+            df['RSI'] = ta.rsi(df['close'], length=14)
+            df['EMA_200_dist'] = (df['close'] / ta.ema(df['close'], length=200)) - 1
+            adx = ta.adx(df['high'], df['low'], df['close'])
+            if adx is not None and not adx.empty:
+                 df['Trend_Strong'] = np.where(adx.iloc[:, 0] > 25, 1, 0)
+            else:
+                 df['Trend_Strong'] = 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
-
-    # ---------------- Prediction -------------------
-
     def predict(self, ohlcv_data: dict) -> dict:
+        """
+        استقبال البيانات الخام (Dictionary of DataFrames/Lists)،
+        تجهيزها، ثم استدعاء النموذج للتنبؤ.
+        """
         if not self.initialized or not self.model:
-            print("⚠️ [Titan] لم يُهيَّأ. score=0", flush=True)
             return {'score': 0.0, 'error': 'Titan not initialized'}
+
         try:
+            # 1. تجهيز البيانات لكل إطار
             processed_tfs = {}
             for tf, data in ohlcv_data.items():
-                if data is None:
-                    continue
+                if not data: continue
+                # تحويل القوائم إلى DataFrame إذا لزم الأمر
                 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()
-                processed_tfs[tf] = self.apply_inverted_pyramid(df, tf)
+                
+                # تطبيق المؤشرات حسب الإطار
+                df = self.apply_inverted_pyramid(df, tf)
+                processed_tfs[tf] = df
 
-            if '5m' not in processed_tfs or processed_tfs['5m'].empty:
-                print("⚠️ [Titan] Missing 5m base timeframe", flush=True)
+            # 2. الدمج (Alignment) للحصول على آخر لقطة (Latest Snapshot)
+            if '5m' not in processed_tfs:
                 return {'score': 0.0, 'error': 'Missing 5m base timeframe'}
 
-            base5 = processed_tfs['5m'].iloc[-1:].copy()
-            latest_ts = int(base5['timestamp'].iloc[0])
-            row = base5.add_prefix('5m_').rename(columns={'5m_timestamp': 'timestamp'})
-
-            # دمج آخر صف متاح من كل إطار ≤ latest_ts
-            for tf, tdf in processed_tfs.items():
-                if tf == '5m' or tdf is None or tdf.empty:
-                    continue
-                rel = tdf[tdf['timestamp'] <= latest_ts]
-                if rel.empty:
-                    continue
-                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
-                feats.append(float(val))
-
-            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'}
+            # نأخذ آخر صف فقط من الـ 5m كأساس
+            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'})
+
+            # دمج باقي الأطر (نأخذ آخر شمعة أغلقت قبل أو مع شمعة الـ 5m الحالية)
+            for tf, df in processed_tfs.items():
+                if tf == '5m': continue
+                # العثور على الشمعة المناسبة زمنياً
+                relevant_row = df[df['timestamp'] <= latest_ts].iloc[-1:].copy()
+                if relevant_row.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]
+
+            # 3. تجهيز شعاع الإدخال (Feature Vector)
+            # التأكد من وجود كل الميزات المطلوبة بالترتيب الصحيح
+            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
+                input_data.append(val)
+
+            # 4. التنبؤ
+            # تحويل إلى DMatrix (تنسيق XGBoost السريع)
+            dtest = xgb.DMatrix([input_data], feature_names=self.feature_names)
+            prediction = self.model.predict(dtest)[0] # إرجاع الاحتمالية الأولى
+
+            return {
+                'score': float(prediction),
+                'timestamp': int(latest_ts),
+                'status': 'OK'
+            }
 
         except Exception as e:
-            import traceback; traceback.print_exc()
+            # print(f"⚠️ [Titan Error] {e}")
+            import traceback
+            traceback.print_exc()
             return {'score': 0.0, 'error': str(e)}