# ============================================================ # 🧠 ml_engine/processor.py # (V37.0 - GEM-Architect: Context-Aware Cybernetic Processor) # ============================================================ import asyncio import traceback import logging import os import sys import numpy as np from typing import Dict, Any, List, Optional # --- استيراد المحركات (كما هي) --- try: from .titan_engine import TitanEngine except ImportError: TitanEngine = None try: from .patterns import ChartPatternAnalyzer except ImportError: ChartPatternAnalyzer = None try: from .monte_carlo import MonteCarloEngine except ImportError: MonteCarloEngine = None try: from .oracle_engine import OracleEngine except ImportError: OracleEngine = None try: from .sniper_engine import SniperEngine except ImportError: SniperEngine = None try: from .hybrid_guardian import HybridDeepSteward except ImportError: HybridDeepSteward = None try: from .guardian_hydra import GuardianHydra except ImportError: GuardianHydra = None # ============================================================ # 📂 مسارات النماذج # ============================================================ BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) MODELS_L2_DIR = os.path.join(BASE_DIR, "ml_models", "layer2") MODELS_PATTERN_DIR = os.path.join(BASE_DIR, "ml_models", "xgboost_pattern2") MODELS_UNIFIED_DIR = os.path.join(BASE_DIR, "ml_models", "Unified_Models_V1") MODELS_SNIPER_DIR = os.path.join(BASE_DIR, "ml_models", "guard_v2") MODELS_HYDRA_DIR = os.path.join(BASE_DIR, "ml_models", "guard_v1") MODEL_V2_PATH = os.path.join(BASE_DIR, "ml_models", "DeepSteward_V2_Production.json") MODEL_V3_PATH = os.path.join(BASE_DIR, "ml_models", "DeepSteward_V3_Production.json") MODEL_V3_FEAT = os.path.join(BASE_DIR, "ml_models", "DeepSteward_V3_Features.json") # ============================================================ # 🎛️ SYSTEM LIMITS & THRESHOLDS (Fallback / Global) # ============================================================ class SystemLimits: """ GEM-Architect: The Dynamic Constitution. يتم تحديث هذه القيم آلياً بواسطة AdaptiveHub وتستخدم كقيم احتياطية (Fallback) في حال لم يتم توفير dynamic_limits للعملة. """ # --- Layer 1 (Data Manager Control) --- L1_MIN_AFFINITY_SCORE = 15.0 # --- Layer 2 Weights (Dynamic) --- L2_WEIGHT_TITAN = 0.40 L2_WEIGHT_PATTERNS = 0.30 L2_WEIGHT_MC = 0.10 # إعدادات الأنماط (تتغير حسب الاستراتيجية) PATTERN_TF_WEIGHTS = {'15m': 0.40, '1h': 0.30, '5m': 0.20, '4h': 0.10, '1d': 0.00} PATTERN_THRESH_BULLISH = 0.60 PATTERN_THRESH_BEARISH = 0.40 # --- Layer 3 (Oracle) --- L3_CONFIDENCE_THRESHOLD = 0.65 L3_WHALE_IMPACT_MAX = 0.10 L3_NEWS_IMPACT_MAX = 0.05 L3_MC_ADVANCED_MAX = 0.10 # --- Layer 4 (Sniper & Execution) --- L4_ENTRY_THRESHOLD = 0.40 L4_WEIGHT_ML = 0.60 L4_WEIGHT_OB = 0.40 L4_OB_WALL_RATIO = 0.40 # --- Layer 0: Hydra & Guardian Thresholds --- HYDRA_CRASH_THRESH = 0.60 HYDRA_GIVEBACK_THRESH = 0.70 HYDRA_STAGNATION_THRESH = 0.50 # Legacy Guard Thresholds LEGACY_V2_PANIC_THRESH = 0.95 LEGACY_V3_HARD_THRESH = 0.95 LEGACY_V3_SOFT_THRESH = 0.85 LEGACY_V3_ULTRA_THRESH = 0.98 @classmethod def to_dict(cls) -> Dict[str, Any]: return {k: v for k, v in cls.__dict__.items() if not k.startswith('__') and not callable(v)} @classmethod def update_from_dict(cls, config: Dict[str, Any]): if not config: return for k, v in config.items(): if hasattr(cls, k): setattr(cls, k, v) # ============================================================ # 🧠 MLProcessor Class # ============================================================ class MLProcessor: def __init__(self, data_manager=None): self.data_manager = data_manager self.initialized = False self.titan = TitanEngine(model_dir=MODELS_L2_DIR) if TitanEngine else None self.pattern_engine = ChartPatternAnalyzer(models_dir=MODELS_PATTERN_DIR) if ChartPatternAnalyzer else None self.mc_analyzer = MonteCarloEngine() if MonteCarloEngine else None self.oracle = OracleEngine(model_dir=MODELS_UNIFIED_DIR) if OracleEngine else None self.sniper = SniperEngine(models_dir=MODELS_SNIPER_DIR) if SniperEngine else None self.guardian_hydra = None if GuardianHydra: self.guardian_hydra = GuardianHydra(model_dir=MODELS_HYDRA_DIR) self.guardian_legacy = None if HybridDeepSteward: self.guardian_legacy = HybridDeepSteward( v2_model_path=MODEL_V2_PATH, v3_model_path=MODEL_V3_PATH, v3_features_map_path=MODEL_V3_FEAT ) print(f"🧠 [MLProcessor V37.0] Context-Aware Cybernetics Active.") async def initialize(self): if self.initialized: return print("⚙️ [Processor] Initializing Neural Grid...") try: tasks = [] if self.titan: tasks.append(self.titan.initialize()) if self.pattern_engine: self.pattern_engine.configure_thresholds( weights=SystemLimits.PATTERN_TF_WEIGHTS, bull_thresh=SystemLimits.PATTERN_THRESH_BULLISH, bear_thresh=SystemLimits.PATTERN_THRESH_BEARISH ) tasks.append(self.pattern_engine.initialize()) if self.oracle: if hasattr(self.oracle, 'set_threshold'): self.oracle.set_threshold(SystemLimits.L3_CONFIDENCE_THRESHOLD) tasks.append(self.oracle.initialize()) if self.sniper: if hasattr(self.sniper, 'configure_settings'): self.sniper.configure_settings( threshold=SystemLimits.L4_ENTRY_THRESHOLD, wall_ratio=SystemLimits.L4_OB_WALL_RATIO, w_ml=SystemLimits.L4_WEIGHT_ML, w_ob=SystemLimits.L4_WEIGHT_OB ) tasks.append(self.sniper.initialize()) if tasks: await asyncio.gather(*tasks) if self.guardian_hydra: self.guardian_hydra.initialize() print(" 🛡️ [Guard 1] Hydra X-Ray: Active") if self.guardian_legacy: if asyncio.iscoroutinefunction(self.guardian_legacy.initialize): await self.guardian_legacy.initialize() else: self.guardian_legacy.initialize() self.guardian_legacy.configure_thresholds( v2_panic=SystemLimits.LEGACY_V2_PANIC_THRESH, v3_hard=SystemLimits.LEGACY_V3_HARD_THRESH, v3_soft=SystemLimits.LEGACY_V3_SOFT_THRESH, v3_ultra=SystemLimits.LEGACY_V3_ULTRA_THRESH ) print(f" 🛡️ [Guard 2] Legacy Steward: Active") self.initialized = True print("✅ [Processor] All Systems Operational.") except Exception as e: print(f"❌ [Processor FATAL] Init failed: {e}") traceback.print_exc() async def process_compound_signal(self, raw_data: Dict[str, Any]) -> Optional[Dict[str, Any]]: """ L2 Processing: Uses 'dynamic_limits' from raw_data if available (Per-Asset Overrides), otherwise falls back to SystemLimits (Global). """ if not self.initialized: await self.initialize() symbol = raw_data.get('symbol') ohlcv_data = raw_data.get('ohlcv') current_price = raw_data.get('current_price', 0.0) # ✅ EXTRACT DYNAMIC LIMITS (Priority: Local > Global) limits = raw_data.get('dynamic_limits', {}) if not symbol or not ohlcv_data: return None try: # 1. Titan Engine score_titan = 0.5 titan_res = {} if self.titan: titan_res = await asyncio.to_thread(self.titan.predict, ohlcv_data) score_titan = titan_res.get('score', 0.5) # 2. Pattern Engine score_patterns = 0.5 pattern_res = {} pattern_name = "Neutral" if self.pattern_engine: # Use Global config for pattern internal TFs for now self.pattern_engine.configure_thresholds( weights=SystemLimits.PATTERN_TF_WEIGHTS, bull_thresh=SystemLimits.PATTERN_THRESH_BULLISH, bear_thresh=SystemLimits.PATTERN_THRESH_BEARISH ) pattern_res = await self.pattern_engine.detect_chart_patterns(ohlcv_data) score_patterns = pattern_res.get('pattern_confidence', 0.5) pattern_name = pattern_res.get('pattern_detected', 'Neutral') # 3. Monte Carlo (Light) mc_score = 0.5 if self.mc_analyzer and '1h' in ohlcv_data: closes = [c[4] for c in ohlcv_data['1h']] raw_mc = self.mc_analyzer.run_light_check(closes) mc_score = 0.5 + (raw_mc * 5.0) mc_score = max(0.0, min(1.0, mc_score)) # 4. Hybrid Calculation (USING DYNAMIC WEIGHTS) w_titan = limits.get('w_titan', SystemLimits.L2_WEIGHT_TITAN) w_patt = limits.get('w_patt', SystemLimits.L2_WEIGHT_PATTERNS) w_mc = SystemLimits.L2_WEIGHT_MC total_w = w_titan + w_patt + w_mc if total_w <= 0: total_w = 1.0 hybrid_score = ((score_titan * w_titan) + (score_patterns * w_patt) + (mc_score * w_mc)) / total_w return { 'symbol': symbol, 'current_price': current_price, 'enhanced_final_score': hybrid_score, # Pass limits forward for next layers 'dynamic_limits': limits, 'asset_regime': raw_data.get('asset_regime', 'UNKNOWN'), 'titan_score': score_titan, 'patterns_score': score_patterns, 'mc_score': mc_score, 'components': { 'titan_score': score_titan, 'patterns_score': score_patterns, 'mc_score': mc_score }, 'pattern_name': pattern_name, 'ohlcv': ohlcv_data, 'titan_details': titan_res, 'pattern_details': pattern_res.get('details', {}) } except Exception as e: print(f"❌ [Processor] Error processing {symbol}: {e}") return None async def consult_oracle(self, symbol_data: Dict[str, Any]) -> Dict[str, Any]: """ L3 Processing: Oracle uses specific threshold from dynamic_limits (Per-Asset). """ if not self.initialized: await self.initialize() # ✅ EXTRACT DYNAMIC THRESHOLD limits = symbol_data.get('dynamic_limits', {}) threshold = limits.get('l3_oracle_thresh', SystemLimits.L3_CONFIDENCE_THRESHOLD) if self.oracle: if hasattr(self.oracle, 'set_threshold'): self.oracle.set_threshold(threshold) decision = await self.oracle.predict(symbol_data) conf = decision.get('confidence', 0.0) # Dynamic Veto based on Context if decision.get('action') in ['WATCH', 'BUY'] and conf < threshold: decision['action'] = 'WAIT' decision['reason'] = f"Context Veto: Conf {conf:.2f} < Limit {threshold:.2f} ({limits.get('regime','Global')})" return decision return {'action': 'WAIT', 'reason': 'Oracle Engine Missing'} async def check_sniper_entry(self, ohlcv_1m_data: List, order_book_data: Dict[str, Any], context_data: Dict = None) -> Dict[str, Any]: """ L4 Processing: Sniper uses specific wall ratio and thresholds from dynamic_limits. """ if not self.initialized: await self.initialize() # ✅ EXTRACT DYNAMIC CONFIG limits = context_data.get('dynamic_limits', {}) if context_data else {} thresh = limits.get('l4_sniper_thresh', SystemLimits.L4_ENTRY_THRESHOLD) wall_r = limits.get('l4_ob_wall_ratio', SystemLimits.L4_OB_WALL_RATIO) if self.sniper: # Inject Dynamic Config before check if hasattr(self.sniper, 'configure_settings'): self.sniper.configure_settings( threshold=thresh, wall_ratio=wall_r, w_ml=SystemLimits.L4_WEIGHT_ML, w_ob=SystemLimits.L4_WEIGHT_OB ) return await self.sniper.check_entry_signal_async(ohlcv_1m_data, order_book_data) return {'signal': 'WAIT', 'reason': 'Sniper Engine Missing'} def consult_dual_guardians(self, symbol, ohlcv_1m, ohlcv_5m, ohlcv_15m, trade_context, order_book_snapshot=None): """ L0 Guardians: Ideally, trade_context should also carry 'dynamic_limits' if we want per-asset guarding. For now, we use Global SystemLimits which are updated by AdaptiveHub to reflect 'General Market State'. """ response = {'action': 'HOLD', 'detailed_log': '', 'probs': {}} # 1. Hydra hydra_result = {'action': 'HOLD', 'reason': 'Disabled', 'probs': {}} if self.guardian_hydra and self.guardian_hydra.initialized: hydra_result = self.guardian_hydra.analyze_position(symbol, ohlcv_1m, ohlcv_5m, ohlcv_15m, trade_context) h_probs = hydra_result.get('probs', {}) p_crash = h_probs.get('crash', 0.0) p_giveback = h_probs.get('giveback', 0.0) # Using Global SystemLimits (updated by Hub) if hydra_result['action'] == 'HOLD': if p_crash >= SystemLimits.HYDRA_CRASH_THRESH: hydra_result['action'] = 'EXIT_HARD' hydra_result['reason'] = f"Hydra Crash Risk {p_crash:.2f}" elif p_giveback >= SystemLimits.HYDRA_GIVEBACK_THRESH: hydra_result['action'] = 'EXIT_SOFT' hydra_result['reason'] = f"Hydra Giveback Risk {p_giveback:.2f}" # 2. Legacy (Volume-Aware Veto) legacy_result = {'action': 'HOLD', 'reason': 'Disabled', 'scores': {}} if self.guardian_legacy and self.guardian_legacy.initialized: self.guardian_legacy.configure_thresholds( v2_panic=SystemLimits.LEGACY_V2_PANIC_THRESH, v3_hard=SystemLimits.LEGACY_V3_HARD_THRESH, v3_soft=SystemLimits.LEGACY_V3_SOFT_THRESH, v3_ultra=SystemLimits.LEGACY_V3_ULTRA_THRESH ) entry_price = float(trade_context.get('entry_price', 0.0)) vol_30m = trade_context.get('volume_30m_usd', 0.0) legacy_result = self.guardian_legacy.analyze_position( ohlcv_1m, ohlcv_5m, ohlcv_15m, entry_price, order_book=order_book_snapshot, volume_30m_usd=vol_30m ) # 3. Final Arbitration h_probs = hydra_result.get('probs', {}) l_scores = legacy_result.get('scores', {}) h_c = h_probs.get('crash', 0.0) h_g = h_probs.get('giveback', 0.0) h_s = h_probs.get('stagnation', 0.0) l_v2 = l_scores.get('v2', 0.0) l_v3 = l_scores.get('v3', 0.0) stamp_str = f"🐲[C:{h_c:.0%}|G:{h_g:.0%}|S:{h_s:.0%}] 🕸️[V2:{l_v2:.0%}|V3:{l_v3:.0%}]" final_action = 'HOLD' final_reason = f"Safe. {stamp_str}" if hydra_result['action'] in ['EXIT_HARD', 'EXIT_SOFT', 'TIGHTEN_SL', 'TRAIL_SL']: final_action = hydra_result['action'] final_reason = f"🐲 HYDRA: {hydra_result['reason']} | {stamp_str}" elif legacy_result['action'] in ['EXIT_HARD', 'EXIT_SOFT']: final_action = legacy_result['action'] final_reason = f"🕸️ LEGACY: {legacy_result['reason']} | {stamp_str}" return { 'action': final_action, 'reason': final_reason, 'detailed_log': f"{final_action} | {stamp_str}", 'probs': h_probs, 'scores': l_scores } async def run_advanced_monte_carlo(self, symbol, timeframe='1h'): if self.mc_analyzer and self.data_manager: try: ohlcv = await self.data_manager.get_latest_ohlcv(symbol, timeframe, limit=300) if ohlcv: return self.mc_analyzer.run_advanced_simulation([c[4] for c in ohlcv]) except Exception: pass return 0.0