# ml_engine/monte_carlo.py # (V11.8 - GEM-Architect: Dynamic Sensitivity Logic) # - Tier 1: Light Check -> Keep as is (Micro-scores). # - Tier 2: Advanced Simulation -> Dynamic Range (-0.10 to +0.10). import numpy as np import pandas as pd from scipy.stats import norm from typing import Dict, Any, List, Optional import traceback class MonteCarloEngine: """ Advanced Stochastic Simulation Engine. Provides two tiers of analysis: 1. Light Check: For rapid screening (All 150 Coins) -> Returns sensitive micro-scores. 2. Advanced Simulation: For deep risk assessment (Top 10 Only). """ def __init__(self): print("✅ [MonteCarlo V11.8] Hybrid Engine Loaded (Dynamic Sensitivity).") # ============================================================================== # ⚡ TIER 1: Light Check (Applied to ALL 150 Coins) # ============================================================================== def run_light_check(self, prices: List[float]) -> float: """ فحص سريع يطبق على الـ 150 عملة. (تم الإبقاء على المنطق كما هو بناء على التحديث السابق) """ try: if not prices or len(prices) < 20: return 0.0 clean_prices = [] for p in prices: try: val = float(p) if val > 0: clean_prices.append(val) except: continue if len(clean_prices) < 20: return 0.0 price_arr = np.array(clean_prices, dtype=np.float64) log_returns = np.diff(np.log(price_arr)) mu = np.mean(log_returns) sigma = np.std(log_returns) if sigma == 0: return 0.0 prob_up = norm.cdf(mu / sigma) # سلم الحساسية (No Dead Zone) if prob_up >= 0.55: return 0.10 elif prob_up >= 0.52: return 0.07 elif prob_up > 0.5001: return 0.03 elif prob_up <= 0.45: return -0.10 elif prob_up <= 0.48: return -0.07 elif prob_up < 0.4999: return -0.03 return 0.00 except Exception as e: return 0.0 # ============================================================================== # 🧠 TIER 2: Advanced Simulation (Applied to Top 10 Only) # ============================================================================== def run_advanced_simulation(self, prices: List[float], num_simulations=5000, time_horizon=24) -> float: """ [تعديل V11.8]: نظام تنقيط ديناميكي (Dynamic Scoring). النطاق: من -0.10 (سلبي جداً) إلى +0.10 (إيجابي جداً). يعتمد على قوة احتمالية الربح (Prob Profit) والعائد المتوقع (Exp Return). """ try: if not prices or len(prices) < 50: return 0.0 clean_prices = [float(p) for p in prices if p is not None] if len(clean_prices) < 50: return 0.0 closing_prices = np.array(clean_prices, dtype=np.float64) log_returns = np.diff(np.log(closing_prices)) mean_daily_return = np.mean(log_returns) daily_volatility = np.std(log_returns) last_price = closing_prices[-1] # GBM Simulation random_shocks = np.random.normal(0, 1, (num_simulations, time_horizon)) drift = (mean_daily_return - 0.5 * daily_volatility ** 2) diffusion = daily_volatility * random_shocks daily_returns_sim = np.exp(drift + diffusion) price_paths = np.zeros_like(daily_returns_sim) price_paths[:, 0] = last_price * daily_returns_sim[:, 0] for t in range(1, time_horizon): price_paths[:, t] = price_paths[:, t-1] * daily_returns_sim[:, t] final_prices = price_paths[:, -1] # Metrics threshold = last_price * 1.002 prob_profit = np.mean(final_prices > threshold) # نسبة المسارات الرابحة var_95 = np.percentile(final_prices, 5) drawdown_risk = (var_95 - last_price) / last_price expected_return = (np.mean(final_prices) - last_price) / last_price # --------------------------------------------------------- # ⚖️ Dynamic Scoring Logic (-0.10 to +0.10) # --------------------------------------------------------- mc_score = 0.0 # 1. القوة الإيجابية (Max +0.10) if prob_profit > 0.50: # إضافة تدريجية بحسب قوة الاحتمالية # مثال: 0.55 -> يضيف نقاط قليلة، 0.70 -> يضيف الكثير strength = (prob_profit - 0.50) * 2 # Scale 0.0 to 1.0 mc_score += strength * 0.08 # Max 0.08 if expected_return > 0.01: # عائد متوقع ممتاز mc_score += 0.02 # 2. القوة السلبية (Max -0.10) elif prob_profit <= 0.50: # خصم تدريجي بحسب ضعف الاحتمالية weakness = (0.50 - prob_profit) * 2 # Scale 0.0 to 1.0 mc_score -= weakness * 0.08 # Max -0.08 if drawdown_risk < -0.05: # مخاطرة عالية mc_score -= 0.02 # Clamp Result (لضمان عدم تجاوز الحدود المطلوبة) final_mc_score = max(-0.10, min(0.10, mc_score)) return float(final_mc_score) except Exception as e: return 0.0