trade_manager.py

# trade_manager.py (Updated to V7.0 - 5m Detector Profit-Saver)
import asyncio
import json
import time
import traceback
import os 
from datetime import datetime, timedelta
from typing import Dict, Any, List
from collections import deque, defaultdict 

import pandas as pd
try:
    import pandas_ta as ta
except ImportError:
    print("⚠️ مكتبة pandas_ta غير موجودة، مؤشرات الحارس (Sentry 1m) ستفشل.")
    ta = None

try:
    import ccxt.async_support as ccxtasync
    CCXT_ASYNC_AVAILABLE = True
except ImportError:
    print("❌❌❌ خطأ فادح: فشل استيراد 'ccxt.async_support'. ❌❌❌")
    CCXT_ASYNC_AVAILABLE = False

import numpy as np
from helpers import safe_float_conversion

# 🔴 --- START OF CHANGE (V7.0) --- 🔴
# (استيراد المحللات لتشغيل الكاشف المصغر 5m)
from ml_engine.indicators import AdvancedTechnicalAnalyzer
from ml_engine.patterns import ChartPatternAnalyzer
# 🔴 --- END OF CHANGE --- 🔴

class TacticalData:
    """
    (محدث V7.0)
    لتخزين بيانات 1m (للدخول) و 5m (لحماية الأرباح).
    """
    def __init__(self, symbol):
        self.symbol = symbol
        self.order_book = None 
        self.trades = deque(maxlen=100) 
        self.cvd = 0.0 
        self.large_trades = []
        self.last_update = time.time()
        
        self.confirmation_trades = defaultdict(lambda: deque(maxlen=50))
        self.confirmation_cvd = defaultdict(float)
        
        self.last_kucoin_trade_id = None
        self.last_confirmation_trade_ids = defaultdict(lambda: None)
        
        # (بيانات 1-دقيقة لزناد الدخول)
        self.ohlcv_1m = deque(maxlen=100) 
        self.indicators_1m = {} 
        self.last_1m_candle_timestamp = None
        
        # 🔴 --- START OF CHANGE (V7.0) --- 🔴
        # (بيانات 5-دقائق لحماية الأرباح)
        self.ohlcv_5m = deque(maxlen=100) # (لتخزين 100 شمعة 5-دقائق)
        self.last_5m_candle_timestamp = None
        self.new_5m_data_added = False # (مؤشر لتشغيل الكاشف 5m)
        # 🔴 --- END OF CHANGE --- 🔴

    def add_trade(self, trade):
        trade_id = trade.get('id')
        if trade_id and trade_id == self.last_kucoin_trade_id:
            return 
        self.last_kucoin_trade_id = trade_id

        self.trades.append(trade)
        self.last_update = time.time()
        try:
            trade_amount = float(trade['amount'])
            if trade['side'] == 'buy': self.cvd += trade_amount
            else: self.cvd -= trade_amount
            trade_cost_usd = float(trade.get('cost', 0))
            if trade_cost_usd == 0 and 'price' in trade:
                trade_cost_usd = float(trade['price']) * trade_amount
            if trade_cost_usd > 20000:
                self.large_trades.append(trade)
                if len(self.large_trades) > 20: self.large_trades.pop(0)
        except Exception: pass

    def add_confirmation_trade(self, exchange_id: str, trade: Dict):
        trade_id = trade.get('id')
        if trade_id and trade_id == self.last_confirmation_trade_ids[exchange_id]:
            return 
        self.last_confirmation_trade_ids[exchange_id] = trade_id
        
        self.confirmation_trades[exchange_id].append(trade)
        try:
            trade_amount = float(trade['amount'])
            if trade['side'] == 'buy': self.confirmation_cvd[exchange_id] += trade_amount
            else: self.confirmation_cvd[exchange_id] -= trade_amount
        except Exception: pass

    def set_order_book(self, ob):
        self.order_book = ob
        self.last_update = time.time()

    def analyze_order_book(self):
        if not self.order_book: return {"bids_depth": 0, "asks_depth": 0}
        try:
            bids = self.order_book.get('bids', []); asks = self.order_book.get('asks', [])
            bids_depth = sum(price * amount for price, amount in bids[:10])
            asks_depth = sum(price * amount for price, amount in asks[:10])
            return {"bids_depth": bids_depth, "asks_depth": asks_depth}
        except Exception: return {"bids_depth": 0, "asks_depth": 0}

    def add_1m_ohlcv(self, ohlcv_data: List):
        """إضافة شموع 1-دقيقة وحساب المؤشرات (للدخول)"""
        if not ohlcv_data:
            return
        
        new_candles_added = False
        for candle in ohlcv_data:
            timestamp = candle[0]
            if timestamp and timestamp != self.last_1m_candle_timestamp:
                if self.ohlcv_1m and timestamp < self.ohlcv_1m[-1][0]:
                    continue 
                
                self.ohlcv_1m.append(candle)
                self.last_1m_candle_timestamp = timestamp
                new_candles_added = True
        
        if new_candles_added and len(self.ohlcv_1m) >= 26: 
            self._analyze_1m_indicators()

    def _analyze_1m_indicators(self):
        """حساب مؤشرات 1-دقيقة الحقيقية (للدخول)"""
        if ta is None or len(self.ohlcv_1m) < 26:
            self.indicators_1m = {}
            return
            
        try:
            df = pd.DataFrame(list(self.ohlcv_1m), columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
            df[['open', 'high', 'low', 'close', 'volume']] = df[['open', 'high', 'low', 'close', 'volume']].astype(float)
            close = df['close']

            ema_9 = ta.ema(close, length=9)
            ema_21 = ta.ema(close, length=21)
            macd_data = ta.macd(close, fast=12, slow=26, signal=9)
            
            if ema_9 is not None and not ema_9.empty and \
               ema_21 is not None and not ema_21.empty and \
               macd_data is not None and not macd_data.empty:
                
                self.indicators_1m = {
                    'ema_9': ema_9.iloc[-1],
                    'ema_21': ema_21.iloc[-1],
                    'macd_hist': macd_data['MACDh_12_26_9'].iloc[-1]
                }
            else:
                self.indicators_1m = {}
                
        except Exception as e:
            self.indicators_1m = {}
            
    # 🔴 --- START OF CHANGE (V7.0) --- 🔴
    def add_5m_ohlcv(self, ohlcv_data: List):
        """(جديد V7.0) إضافة شموع 5-دقائق (لحماية الأرباح)"""
        if not ohlcv_data:
            return
        
        for candle in ohlcv_data:
            timestamp = candle[0]
            if timestamp and timestamp != self.last_5m_candle_timestamp:
                if self.ohlcv_5m and timestamp < self.ohlcv_5m[-1][0]:
                    continue
                
                self.ohlcv_5m.append(candle)
                self.last_5m_candle_timestamp = timestamp
                # (تعيين المؤشر لإعلام حلقة التحليل بوجود بيانات جديدة)
                self.new_5m_data_added = True
    # 🔴 --- END OF CHANGE --- 🔴

    def get_tactical_snapshot(self):
        agg_cvd = sum(self.confirmation_cvd.values())
        
        return {
            "cvd_kucoin": self.cvd,
            "cvd_confirmation_sources": dict(self.confirmation_cvd),
            "cvd_confirmation_aggregate": agg_cvd,
            "large_trades_count_5m": len([t for t in self.large_trades if t.get('timestamp') and (time.time() - t['timestamp']/1000) < 300]),
            "indicators_1m": self.indicators_1m,
            "ob_analysis": self.analyze_order_book()
        }


class TradeManager:
    def __init__(self, r2_service, learning_hub=None, data_manager=None, state_manager=None, callback_on_close=None):
        if not CCXT_ASYNC_AVAILABLE:
            raise RuntimeError("مكتبة 'ccxt.async_support' غير متاحة.")

        self.r2_service = r2_service
        self.learning_hub = learning_hub
        self.data_manager = data_manager
        self.state_manager = state_manager
        self.callback_on_close = callback_on_close
        
        self.is_running = False
        self.sentry_watchlist = {}
        self.sentry_tasks = {}
        self.tactical_data_cache = {}
        
        self.sentry_lock = asyncio.Lock()
        
        self.kucoin_rest = None
        self.confirmation_exchanges = {}
        self.polling_interval = 1.5
        self.confirmation_polling_interval = 3.0
        
        # 🔴 --- START OF CHANGE (V7.0) --- 🔴
        # (تهيئة محللات الكاشف المصغر 5m)
        self.sentry_technical_analyzer = AdvancedTechnicalAnalyzer()
        self.sentry_pattern_analyzer = ChartPatternAnalyzer()
        # 🔴 --- END OF CHANGE --- 🔴

    async def initialize_sentry_exchanges(self): 
        try:
            print("🔄 [Sentry] تهيئة منصات التداول (KuCoin REST ومنصات التأكيد)...")
            
            print("   [Sentry] تهيئة KuCoin للبيانات العامة (وضع المحاكاة).")
            self.kucoin_rest = ccxtasync.kucoin()
            await self.kucoin_rest.load_markets()
            print("✅ [Sentry] منصة REST الأساسية (KuCoin) جاهزة (بيانات عامة فقط).")

            self.confirmation_exchanges = {}
            confirmation_exchange_ids = ['bybit', 'okx', 'gateio'] 
            
            print(f"   [Sentry] تهيئة منصات التأكيد (Confirmation Exchanges): {', '.join(confirmation_exchange_ids)}")
            for ex_id in confirmation_exchange_ids:
                try:
                    exchange = getattr(ccxtasync, ex_id)()
                    await exchange.load_markets()
                    self.confirmation_exchanges[ex_id] = exchange
                    print(f"   ✅ [Sentry] منصة التأكيد {ex_id} جاهزة (REST).")
                except Exception as ex_err:
                    print(f"   ⚠️ [Sentry] فشل تهيئة منصة التأكيد {ex_id}: {ex_err}")
                    if ex_id in self.confirmation_exchanges:
                        try:
                            await self.confirmation_exchanges[ex_id].close()
                        except Exception:
                            pass 
                        del self.confirmation_exchanges[ex_id]
            
            if not self.confirmation_exchanges:
                print("   ⚠️ [Sentry] تحذير: لم يتم تهيئة أي منصة تأكيد. سيعمل الحارس على بيانات KuCoin فقط.")

        except Exception as e:
            print(f"❌ [Sentry] فشل فادح في تهيئة KuCoin REST: {e}")
            if self.kucoin_rest: await self.kucoin_rest.close()
            for ex in self.confirmation_exchanges.values(): await ex.close()
            raise

    async def start_sentry_and_monitoring_loops(self):
        self.is_running = True
        print(f"✅ [Sentry] بدء حلقات المراقبة التكتيكية (Layer 2 - API Polling)...")
        while self.is_running:
            try:
                async with self.sentry_lock:
                    watchlist_symbols = set(self.sentry_watchlist.keys())
                
                open_trades = await self.get_open_trades()
                open_trade_symbols = {t['symbol'] for t in open_trades}
                
                symbols_to_monitor = watchlist_symbols.union(open_trade_symbols)
                current_tasks = set(self.sentry_tasks.keys())

                symbols_to_add = symbols_to_monitor - current_tasks
                for symbol in symbols_to_add:
                    print(f"   [Sentry] بدء المراقبة التكتيكية (Polling) لـ {symbol}")
                    
                    strategy_hint = 'generic' 
                    if symbol in watchlist_symbols:
                         async with self.sentry_lock:
                            if symbol in self.sentry_watchlist:
                                strategy_hint = self.sentry_watchlist.get(symbol, {}).get('strategy_hint', 'generic')
                    elif symbol in open_trade_symbols:
                        trade = next((t for t in open_trades if t['symbol'] == symbol), None)
                        if trade:
                            strategy_hint = trade.get('strategy', 'generic')

                    if symbol not in self.tactical_data_cache: 
                        self.tactical_data_cache[symbol] = TacticalData(symbol)
                    
                    task = asyncio.create_task(self._monitor_symbol_activity_polling(symbol, strategy_hint))
                    self.sentry_tasks[symbol] = task

                symbols_to_remove = current_tasks - symbols_to_monitor
                for symbol in symbols_to_remove:
                    print(f"   [Sentry] إيقاف المراقبة التكتيكية (Polling) لـ {symbol}")
                    task = self.sentry_tasks.pop(symbol, None)
                    if task: 
                        task.cancel()
                    if symbol in self.tactical_data_cache: 
                        del self.tactical_data_cache[symbol]

                await asyncio.sleep(15)
                
            except Exception as error:
                print(f"❌ [Sentry] خطأ في الحلقة الرئيسية: {error}"); traceback.print_exc(); await asyncio.sleep(60)

    async def stop_sentry_loops(self):
        self.is_running = False
        print("🛑 [Sentry] إيقاف جميع حلقات المراقبة...")
        for task in self.sentry_tasks.values(): task.cancel()
        self.sentry_tasks.clear()
        try:
            if self.kucoin_rest: 
                await self.kucoin_rest.close()
                print("   ✅ [Sentry] تم إغلاق اتصال KuCoin REST.")
                
            for ex_id, exchange in self.confirmation_exchanges.items():
                try:
                    await exchange.close()
                    print(f"   ✅ [Sentry] تم إغلاق اتصال {ex_id} REST.")
                except Exception as e:
                    print(f"⚠️ [Sentry] خطأ أثناء إغلاق منصة {ex_id}: {e}")
            self.confirmation_exchanges.clear()
            print("✅ [Sentry] تم إغلاق اتصالات التداول (REST).")
        except Exception as e: print(f"⚠️ [Sentry] خطأ أثناء إغلاق الاتصالات: {e}")

    async def update_sentry_watchlist(self, candidates: List[Dict]):
        async with self.sentry_lock: 
            self.sentry_watchlist = {c['symbol']: c for c in candidates}
            print(f"ℹ️ [Sentry] تم تحديث Watchlist. عدد المرشحين: {len(self.sentry_watchlist)}")

    def get_sentry_status(self):
        active_monitoring_count = len(self.sentry_tasks)
        watchlist_symbols_list = list(self.sentry_watchlist.keys())
        
        return {
            'is_running': self.is_running,
            'active_monitoring_tasks': active_monitoring_count,
            'watchlist_symbols': watchlist_symbols_list,
            'monitored_symbols': list(self.sentry_tasks.keys()),
            'confirmation_exchanges_active': list(self.confirmation_exchanges.keys())
        }

    async def _monitor_symbol_activity_polling(self, symbol: str, strategy_hint: str):
        if symbol not in self.tactical_data_cache:
            self.tactical_data_cache[symbol] = TacticalData(symbol)

        tasks_to_gather = [
            self._poll_kucoin_data(symbol), 
            self._poll_confirmation_data(symbol), 
            self._run_tactical_analysis_loop(symbol, strategy_hint)
        ]

        try:
            await asyncio.gather(*tasks_to_gather)
        except asyncio.CancelledError:
            print(f"ℹ️ [Sentry] تم إيقاف المراقبة (Polling) لـ {symbol}.")
        except Exception as e:
            print(f"❌ [Sentry] خطأ فادح في مراقبة (Polling) {symbol}: {e}")
            traceback.print_exc()
        finally:
            print(f"🛑 [Sentry] إنهاء جميع مهام (Polling) {symbol}")
            if symbol in self.sentry_tasks: 
                self.sentry_tasks.pop(symbol, None)
            if symbol in self.tactical_data_cache: 
                del self.tactical_data_cache[symbol]

    async def _poll_kucoin_data(self, symbol):
        """(محدث V7.0) حلقة استقصاء (Polling) لبيانات KuCoin (تتضمن 1m و 5m OHLCV)"""
        while self.is_running:
            try:
                if not self.kucoin_rest:
                    print(f"   [Sentry Polling] KuCoin REST غير متاح لـ {symbol}. الانتظار...")
                    await asyncio.sleep(10)
                    continue

                # 🔴 --- START OF CHANGE (V7.0) --- 🔴
                # (جلب 4 أنواع بيانات بالتوازي)
                tasks = {
                    'ob': asyncio.create_task(self.kucoin_rest.fetch_order_book(symbol, limit=20)),
                    'trades': asyncio.create_task(self.kucoin_rest.fetch_trades(symbol, since=int((time.time() - 60) * 1000), limit=50)),
                    'ohlcv_1m': asyncio.create_task(self.kucoin_rest.fetch_ohlcv(symbol, '1m', limit=50)),
                    'ohlcv_5m': asyncio.create_task(self.kucoin_rest.fetch_ohlcv(symbol, '5m', limit=50))
                }
                # 🔴 --- END OF CHANGE --- 🔴

                await asyncio.wait(tasks.values(), return_when=asyncio.ALL_COMPLETED)
                
                if symbol not in self.tactical_data_cache:
                    continue 

                if not tasks['ob'].exception():
                    self.tactical_data_cache[symbol].set_order_book(tasks['ob'].result())

                if not tasks['trades'].exception():
                    trades = tasks['trades'].result()
                    trades.sort(key=lambda x: x['timestamp'])
                    for trade in trades:
                        self.tactical_data_cache[symbol].add_trade(trade)
                
                if not tasks['ohlcv_1m'].exception():
                    self.tactical_data_cache[symbol].add_1m_ohlcv(tasks['ohlcv_1m'].result())
                
                # 🔴 --- START OF CHANGE (V7.0) --- 🔴
                if not tasks['ohlcv_5m'].exception():
                    self.tactical_data_cache[symbol].add_5m_ohlcv(tasks['ohlcv_5m'].result())
                # 🔴 --- END OF CHANGE --- 🔴

                await asyncio.sleep(self.polling_interval)

            except ccxtasync.RateLimitExceeded as e:
                print(f"⏳ [Sentry Polling] {symbol} KuCoin Rate Limit Exceeded: {e}. زيادة فترة الانتظار...")
                await asyncio.sleep(10)
            except asyncio.CancelledError:
                raise 
            except Exception as e:
                print(f"⚠️ [Sentry Polling] خطأ في {symbol} KuCoin data polling: {e}")
                await asyncio.sleep(5)
    
    async def _poll_confirmation_data(self, symbol):
        if not self.confirmation_exchanges:
            return 

        await asyncio.sleep(self.confirmation_polling_interval / 2) 

        while self.is_running:
            try:
                tasks = []
                for ex_id, exchange in self.confirmation_exchanges.items():
                    tasks.append(self._fetch_confirmation_trades(ex_id, exchange, symbol))
                
                await asyncio.gather(*tasks)
                await asyncio.sleep(self.confirmation_polling_interval)

            except asyncio.CancelledError:
                raise 
            except Exception as e:
                print(f"⚠️ [Sentry Conf] خطأ في حلقة التأكيد لـ {symbol}: {e}")
                await asyncio.sleep(10) 

    async def _fetch_confirmation_trades(self, ex_id: str, exchange: ccxtasync.Exchange, symbol: str):
        try:
            if symbol not in exchange.markets:
                return

            since_timestamp = int((time.time() - 60) * 1000) 
            trades = await exchange.fetch_trades(symbol, since=since_timestamp, limit=50)
            
            if symbol in self.tactical_data_cache:
                trades.sort(key=lambda x: x['timestamp'])
                for trade in trades:
                    self.tactical_data_cache[symbol].add_confirmation_trade(ex_id, trade)
        
        except ccxtasync.RateLimitExceeded:
            print(f"⏳ [Sentry Conf] {ex_id} Rate Limit لـ {symbol}. الانتظار...")
            await asyncio.sleep(15) 
        except asyncio.CancelledError:
            raise 
        except Exception as e:
            pass
    

    async def _run_tactical_analysis_loop(self, symbol: str, strategy_hint: str):
        """(محدث V7.0) (دماغ الحارس) يشغل التحليل التكتيكي كل ثانية."""
        while self.is_running:
            await asyncio.sleep(1) 
            try:
                if self.state_manager.trade_analysis_lock.locked(): continue
                trade = await self.get_trade_by_symbol(symbol)
                tactical_data = self.tactical_data_cache.get(symbol)
                if not tactical_data: continue
                
                snapshot = tactical_data.get_tactical_snapshot()
                
                if trade:
                    # 1. التحقق من SL/TP (الأولوية القصوى)
                    exit_reason = self._check_exit_trigger(trade, snapshot, tactical_data)
                    
                    # 🔴 --- START OF CHANGE (V7.0) --- 🔴
                    # 2. إذا لم يتم ضرب SL/TP، تحقق من "مراقب حماية الأرباح" 5m
                    # (يتم تشغيله فقط إذا وصلت شمعة 5m جديدة)
                    if not exit_reason and tactical_data.new_5m_data_added:
                        exit_reason = await self._run_5m_profit_saver(trade, list(tactical_data.ohlcv_5m), tactical_data)
                        tactical_data.new_5m_data_added = False # (إعادة ضبط المؤشر)
                    # 🔴 --- END OF CHANGE --- 🔴
                        
                    if exit_reason:
                        print(f"🛑 [Sentry] زناد خروج لـ {symbol}: {exit_reason}")
                        
                        current_price_to_close = None
                        if "Take Profit" in exit_reason:
                            current_price_to_close = trade.get('take_profit')
                        # (إذا كان الإغلاق تكتيكياً أو وقف خسارة، استخدم سعر Bid)
                        elif tactical_data.order_book and tactical_data.order_book.get('bids') and len(tactical_data.order_book['bids']) > 0:
                            current_price_to_close = tactical_data.order_book['bids'][0][0]
                        else:
                            if tactical_data.trades:
                                current_price_to_close = tactical_data.trades[-1].get('price')

                        if current_price_to_close: 
                            await self.immediate_close_trade(symbol, current_price_to_close, f"Exit Trigger: {exit_reason}")
                        else:
                            print(f"⚠️ [Sentry] {symbol} زناد خروج ولكن لا يمكن تحديد سعر الإغلاق!")
                else:
                    async with self.sentry_lock:
                        is_still_on_watchlist = symbol in self.sentry_watchlist

                    if is_still_on_watchlist:
                        trigger = self._check_entry_trigger(symbol, strategy_hint, snapshot)
                        if trigger:
                            print(f"✅ [Sentry] زناد دخول تكتيكي لـ {symbol} (استراتيجية: {strategy_hint})")
                            
                            watchlist_entry = None
                            async with self.sentry_lock:
                                watchlist_entry = self.sentry_watchlist.pop(symbol, None)
                            
                            if watchlist_entry:
                                explorer_context = watchlist_entry.get('llm_decision_context', {})
                                await self._execute_smart_entry(symbol, strategy_hint, snapshot, explorer_context)
            except asyncio.CancelledError:
                raise 
            except Exception as e: print(f"❌ [Sentry] خطأ في حلقة التحليل التكتيكي لـ {symbol}: {e}"); traceback.print_exc()

    def _check_entry_trigger(self, symbol: str, strategy_hint: str, data: Dict) -> bool:
        """(محدث V6.7) زناد ثلاثي: CVD + دفتر الطلبات + مؤشرات 1-دقيقة"""
        
        cvd_kucoin = data.get('cvd_kucoin', 0)
        
        ob_analysis = data.get('ob_analysis', {})
        bids_depth = ob_analysis.get('bids_depth', 0)
        asks_depth = ob_analysis.get('asks_depth', 0)

        indicators_1m = data.get('indicators_1m', {})
        ema_9_1m = indicators_1m.get('ema_9')
        ema_21_1m = indicators_1m.get('ema_21')
        
        large_trades = data.get('large_trades_count_5m', 0)

        if strategy_hint in ['breakout_momentum', 'trend_following']:
            
            if ema_9_1m is None or ema_21_1m is None:
                return False
                
            cvd_check = (cvd_kucoin > 0)
            ob_check = (bids_depth > asks_depth)
            ema_check = (ema_9_1m > ema_21_1m)

            if cvd_check and ob_check and ema_check:
                print(f"   [Trigger] {symbol} (Momentum): CVD+, OB+, 1m_EMA+. الدخول!")
                return True
                 
        elif strategy_hint == 'mean_reversion':
            pass 
                
        elif strategy_hint == 'volume_spike':
             if (large_trades > 0): 
                 print(f"   [Trigger] {symbol} Volume Spike: LargeTrades={large_trades}")
                 return True
                 
        return False

    def _check_exit_trigger(self, trade: Dict, data: Dict, tactical_data: TacticalData) -> str:
        """(V6.6) يراقب وقف الخسارة وجني الأرباح (الأولوية القصوى)"""
        
        symbol = trade['symbol']
        hard_stop = trade.get('stop_loss')
        take_profit = trade.get('take_profit')
        
        best_bid_price = None
        if tactical_data.order_book and tactical_data.order_book.get('bids') and len(tactical_data.order_book['bids']) > 0:
            best_bid_price = tactical_data.order_book['bids'][0][0]

        last_trade_price = None
        if tactical_data.trades:
            try:
                last_trade_price = tactical_data.trades[-1].get('price')
            except (IndexError, AttributeError):
                pass 

        if best_bid_price is None and last_trade_price is None:
            return None 

        current_price_for_sl = best_bid_price if best_bid_price is not None else last_trade_price
        
        current_price_for_tp = max(
            filter(None, [best_bid_price, last_trade_price]), 
            default=None
        )

        if hard_stop and current_price_for_sl and current_price_for_sl <= hard_stop: 
            return f"Strategic Stop Loss hit: {current_price_for_sl} <= {hard_stop}"
        
        if take_profit and current_price_for_tp and current_price_for_tp >= take_profit: 
            return f"Strategic Take Profit hit: {current_price_for_tp} >= {take_profit}"
        
        return None 

    # 🔴 --- START OF CHANGE (V7.0) --- 🔴
    def _create_dataframe_5m(self, candles: List) -> pd.DataFrame:
        """(جديد V7.0) دالة مساعدة لإنشاء DataFrame لتحليل 5m"""
        try:
            if not candles: return pd.DataFrame()
            df = pd.DataFrame(candles, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
            df[['open', 'high', 'low', 'close', 'volume']] = df[['open', 'high', 'low', 'close', 'volume']].astype(float)
            df.set_index('timestamp', inplace=True)
            df.sort_index(inplace=True)
            return df
        except Exception:
            return pd.DataFrame()

    async def _run_5m_profit_saver(self, trade: Dict, ohlcv_5m_list: List, tactical_data: TacticalData) -> str:
        """
        (جديد V7.0) "مراقب الانعكاس" 5m.
        يستخدم الكاشف (مؤشرات + أنماط) على 5m لإغلاق الصفقات الرابحة.
        """
        try:
            # --- 1. التحقق من الربحية (كما طلبت) ---
            best_bid_price = None
            if tactical_data.order_book and tactical_data.order_book.get('bids') and len(tactical_data.order_book['bids']) > 0:
                best_bid_price = tactical_data.order_book['bids'][0][0]
            
            if best_bid_price is None:
                return None 
                
            entry_price = trade.get('entry_price')
            if best_bid_price <= entry_price:
                return None # الصفقة ليست رابحة، لا تفعل شيئاً

            # --- 2. الصفقة رابحة، تشغيل الكاشف 5m ---
            if len(ohlcv_5m_list) < 20:
                return None # بيانات غير كافية للتحليل
                
            df_5m = self._create_dataframe_5m(ohlcv_5m_list)
            if df_5m.empty:
                return None

            # (تشغيل المحللات - بدون مونت كارلو كما طلبت)
            indicators_5m = self.sentry_technical_analyzer.calculate_all_indicators(df_5m, '5m')
            pattern_analysis_5m = await self.sentry_pattern_analyzer.detect_chart_patterns({'5m': ohlcv_5m_list})

            # --- 3. حساب درجة الانعكاس (Reversal Score) ---
            reversal_score = 0.0
            
            # (التحقق من الأنماط الهبوطية)
            pattern_name = pattern_analysis_5m.get('pattern_detected', '')
            pattern_conf = pattern_analysis_5m.get('pattern_confidence', 0)
            if pattern_name in ['Double Top', 'Downtrend', 'Breakout Down', 'Near Resistance'] and pattern_conf > 0.6:
                reversal_score += 0.5

            # (التحقق من المؤشرات السلبية)
            rsi_5m = indicators_5m.get('rsi', 50)
            macd_hist_5m = indicators_5m.get('macd_hist', 0)
            
            # (إشارة ضعف 1: RSI تحت 50 + MACD سلبي)
            if rsi_5m < 45 and macd_hist_5m < 0:
                reversal_score += 0.5
            # (إشارة ضعف 2: تباعد هبوطي - RSI يفشل في تسجيل قمة جديدة)
            elif rsi_5m < 60 and macd_hist_5m < 0:
                 reversal_score += 0.3 # (إشارة ضعف متوسطة)

            # --- 4. القرار ---
            if reversal_score >= 0.8: # (يتطلب إشارتي ضعف قويتين)
                return f"Tactical 5m Profit Save: Reversal signal detected (Score: {reversal_score})"
                
            return None
            
        except Exception as e:
            print(f"❌ [Sentry] خطأ في مراقب حماية الأرباح 5m: {e}")
            return None
    # 🔴 --- END OF CHANGE --- 🔴


    async def _execute_smart_entry(self, symbol: str, strategy_hint: str, tactical_data: Dict, explorer_context: Dict):
        """(المنفذ الوهمي - Layer 3) يحاكي تنفيذ الصفقة ويحفظها في R2."""
        print(f"🚀 [Executor] بدء تنفيذ الدخول الذكي (وهمي) لـ {symbol}...")
        
        context_for_retry = explorer_context 
        
        if self.state_manager.trade_analysis_lock.locked(): 
            print(f"⚠️ [Executor] تم إلغاء الدخول لـ {symbol} بسبب قفل التحليل الاستراتيجي."); 
            return
            
        if not self.r2_service.acquire_lock(): 
            print(f"⚠️ [Executor] فشل في الحصول على قفل R2 لـ {symbol}. تم الإلغاء."); 
            return

        try:
            if await self.get_trade_by_symbol(symbol): 
                print(f"ℹ️ [Executor] الصفقة {symbol} مفتوحة بالفعل (وهمياً). تم الإلغاء."); 
                return
            
            all_open_trades = await self.get_open_trades()
            if len(all_open_trades) > 0:
                print(f"❌ [Executor] يوجد صفقة أخرى مفتوحة ({all_open_trades[0]['symbol']}). لا يمكن فتح {symbol}.");
                return

            portfolio_state = await self.r2_service.get_portfolio_state_async()
            available_capital = portfolio_state.get("current_capital_usd", 0)
            
            if available_capital < 1: 
                print(f"❌ [Executor] رأس مال وهمي غير كافٍ لـ {symbol}."); 
                return

            current_ask_price = None
            if symbol in self.tactical_data_cache and self.tactical_data_cache[symbol].order_book:
                 ob = self.tactical_data_cache[symbol].order_book
                 if ob and ob.get('asks') and len(ob['asks']) > 0: 
                     current_ask_price = ob['asks'][0][0]
            
            if not current_ask_price: 
                print(f"❌ [Executor] لا يمكن الحصول على السعر الحالي (من البيانات العامة) لـ {symbol}."); 
                return

            llm_decision = explorer_context.get('decision', {}) 
            stop_loss_price = llm_decision.get("stop_loss", current_ask_price * 0.98)
            take_profit_price = llm_decision.get("take_profit", current_ask_price * 1.03)
            exit_profile = llm_decision.get('exit_profile', 'ATR_TRAILING')
            exit_parameters = llm_decision.get('exit_parameters', {})

            if not (stop_loss_price and take_profit_price):
                print(f"❌ [Executor] {symbol}: بيانات SL/TP غير صالحة من النموذج. تم الإلغاء.")
                return

            if current_ask_price >= take_profit_price:
                print(f"⚠️ [Executor] {symbol}: السعر الحالي ({current_ask_price}) أعلى من هدف الربح ({take_profit_price}). الفرصة ضاعت. تم الإلغاء.")
                return
                
            if current_ask_price <= stop_loss_price:
                print(f"⚠️ [Executor] {symbol}: السعر الحالي ({current_ask_price}) أقل من وقف الخسارة ({stop_loss_price}). الصفقة فاشلة. تم الإلغاء.")
                return
            
            final_entry_price = current_ask_price
            print(f"✅ [Executor] (SIMULATED) تم التنفيذ! {symbol} بسعر {final_entry_price}")

            await self._save_trade_to_r2(
                symbol=symbol, entry_price=final_entry_price, position_size_usd=available_capital,
                strategy=strategy_hint, exit_profile=exit_profile, exit_parameters=exit_parameters,
                stop_loss=stop_loss_price, take_profit=take_profit_price,
                tactical_context=tactical_data, explorer_context=explorer_context
            )
            
            print(f"   [Executor] الصفقة {symbol} فُتحت. مسح باقي قائمة المراقبة (Watchlist)...")
            async with self.sentry_lock:
                self.sentry_watchlist.clear()
                print("   [Sentry] تم مسح Watchlist.")

        except Exception as e: 
            print(f"❌ [Executor] فشل فادح أثناء التنفيذ (SIM) لـ {symbol}: {e}"); 
            traceback.print_exc()
            
            print(f"   [Sentry] إعادة {symbol} إلى Watchlist بعد فشل التنفيذ الوهمي.")
            async with self.sentry_lock:
                self.sentry_watchlist[symbol] = {
                    "symbol": symbol, 
                    "strategy_hint": strategy_hint, 
                    "llm_decision_context": context_for_retry 
                }

        finally:
            if self.r2_service.lock_acquired: 
                self.r2_service.release_lock()


    async def _save_trade_to_r2(self, **kwargs):
        """(دالة داخلية - V6.2) تحفظ فقط البيانات الأساسية للصفقة الوهمية."""
        try:
            symbol = kwargs.get('symbol')
            strategy = kwargs.get('strategy')
            exit_profile = kwargs.get('exit_profile')
            expected_target_time = (datetime.now() + timedelta(minutes=15)).isoformat()

            explorer_context_blob = kwargs.get('explorer_context', {})
            llm_decision_only = explorer_context_blob.get('decision', {})
            
            decision_data = {
                "reasoning": f"Tactical entry by Sentry based on {strategy}",
                "strategy": strategy, 
                "exit_profile": exit_profile,
                "exit_parameters": kwargs.get('exit_parameters', {}),
                "tactical_context_at_decision": kwargs.get('tactical_context', {}),
                "explorer_llm_decision": llm_decision_only 
            }
            
            new_trade = {
                "id": str(int(datetime.now().timestamp())), 
                "symbol": symbol,
                "entry_price": kwargs.get('entry_price'), 
                "entry_timestamp": datetime.now().isoformat(),
                "decision_data": decision_data, 
                "status": "OPEN",
                "stop_loss": kwargs.get('stop_loss'), 
                "take_profit": kwargs.get('take_profit'),
                "dynamic_stop_loss": kwargs.get('stop_loss'), 
                "trade_type": "LONG",
                "position_size_usd": kwargs.get('position_size_usd'),
                "expected_target_minutes": 15, 
                "expected_target_time": expected_target_time,
                "is_monitored": True, 
                "strategy": strategy, 
                "monitoring_started": True
            }
            
            trades = await self.r2_service.get_open_trades_async()
            trades.append(new_trade)
            await self.r2_service.save_open_trades_async(trades)
            
            portfolio_state = await self.r2_service.get_portfolio_state_async()
            portfolio_state["invested_capital_usd"] = kwargs.get('position_size_usd')
            portfolio_state["current_capital_usd"] = 0.0
            portfolio_state["total_trades"] = portfolio_state.get("total_trades", 0) + 1
            await self.r2_service.save_portfolio_state_async(portfolio_state)
            
            await self.r2_service.save_system_logs_async({
                "new_trade_opened_by_sentry": True, "symbol": symbol,
                "position_size": kwargs.get('position_size_usd'),
                "strategy": strategy, "exit_profile": exit_profile
            })
            print(f"✅ [R2] تم حفظ الصفقة الجديدة (الوهمية) لـ {symbol} بنجاح.")
        except Exception as e: 
            print(f"❌ [R2] فشل حفظ الصفقة لـ {symbol}: {e}"); 
            traceback.print_exc()
            raise

    async def close_trade(self, trade_to_close, close_price, reason="System Close"):
        try:
            symbol = trade_to_close.get('symbol'); trade_to_close['status'] = 'CLOSED'
            trade_to_close['close_price'] = close_price; trade_to_close['close_timestamp'] = datetime.now().isoformat()
            trade_to_close['is_monitored'] = False; entry_price = trade_to_close['entry_price']
            position_size = trade_to_close['position_size_usd']; strategy = trade_to_close.get('strategy', 'unknown')
            pnl = 0.0; pnl_percent = 0.0
            if entry_price and entry_price > 0 and close_price and close_price > 0:
                try: pnl_percent = ((close_price - entry_price) / entry_price) * 100; pnl = position_size * (pnl_percent / 100)
                except (TypeError, ZeroDivisionError): pnl = 0.0; pnl_percent = 0.0
            trade_to_close['pnl_usd'] = pnl; trade_to_close['pnl_percent'] = pnl_percent
            await self._archive_closed_trade(trade_to_close); await self._update_trade_summary(trade_to_close)
            portfolio_state = await self.r2_service.get_portfolio_state_async()
            current_capital = portfolio_state.get("current_capital_usd", 0); new_capital = current_capital + position_size + pnl
            portfolio_state["current_capital_usd"] = new_capital; portfolio_state["invested_capital_usd"] = 0.0
            if pnl > 0: portfolio_state["winning_trades"] = portfolio_state.get("winning_trades", 0) + 1; portfolio_state["total_profit_usd"] = portfolio_state.get("total_profit_usd", 0.0) + pnl
            elif pnl < 0: portfolio_state["total_loss_usd"] = portfolio_state.get("total_loss_usd", 0.0) + abs(pnl)
            await self.r2_service.save_portfolio_state_async(portfolio_state)
            open_trades = await self.r2_service.get_open_trades_async()
            trades_to_keep = [t for t in open_trades if t.get('id') != trade_to_close.get('id')]
            await self.r2_service.save_open_trades_async(trades_to_keep)
            
            await self.r2_service.save_system_logs_async({
                "trade_closed": True, "symbol": symbol, "pnl_usd": pnl, "pnl_percent": pnl_percent,
                "new_capital": new_capital, "strategy": strategy, "reason": reason
            })
            if self.learning_hub and self.learning_hub.initialized:
                print(f"🧠 [LearningHub] تشغيل التعلم (Reflector+Stats) لـ {symbol}...")
                await self.learning_hub.analyze_trade_and_learn(trade_to_close, reason)
            else: print("⚠️ [Sentry] LearningHub غير متاح، تم تخطي التعلم.")
            
            if self.callback_on_close:
                print("🔄 [Executor] Trade closed. Scheduling immediate Explorer cycle...")
                asyncio.create_task(self.callback_on_close())
            
            print(f"✅ [Executor] تم إغلاق الصفقة (الوهمية) {symbol} - السبب: {reason} - PnL: {pnl_percent:+.2f}%")
            return True
        except Exception as e: print(f"❌ [Executor] فشل فادح أثناء إغلاق الصفقة (الوهمية) {symbol}: {e}"); traceback.print_exc(); raise

    async def immediate_close_trade(self, symbol, close_price, reason="Immediate Close"):
        if not self.r2_service.acquire_lock(): print(f"⚠️ [Executor] فشل في الحصول على قفل R2 لـ {symbol} (Immediate Close)"); return False
        try:
            open_trades = await self.r2_service.get_open_trades_async()
            trade_to_close = next((t for t in open_trades if t['symbol'] == symbol and t['status'] == 'OPEN'), None)
            if not trade_to_close: print(f"⚠️ [Executor] لا توجد صفقة مفتوحة لـ {symbol} لإغلاقها."); return False
            await self.close_trade(trade_to_close, close_price, reason)
            return True
        except Exception as e: print(f"❌ [Executor] فشل في immediate_close {symbol}: {e}"); return False
        finally:
            if self.r2_service.lock_acquired: self.r2_service.release_lock()

    async def update_trade_strategy(self, trade_to_update, re_analysis_decision):
        try:
            symbol = trade_to_update.get('symbol')
            if re_analysis_decision.get('action') == "UPDATE_TRADE":
                trade_to_update['stop_loss'] = re_analysis_decision['new_stop_loss']
                trade_to_update['take_profit'] = re_analysis_decision['new_take_profit']
                trade_to_update['dynamic_stop_loss'] = re_analysis_decision['new_stop_loss']
                trade_to_update['decision_data']['exit_profile'] = re_analysis_decision['new_exit_profile']
                trade_to_update['decision_data']['exit_parameters'] = re_analysis_decision['new_exit_parameters']
                print(f"   🔄 (Explorer) {symbol}: Exit profile updated to {re_analysis_decision['new_exit_profile']}")
            new_expected_minutes = re_analysis_decision.get('new_expected_minutes', 15)
            trade_to_update['expected_target_minutes'] = new_expected_minutes
            trade_to_update['expected_target_time'] = (datetime.now() + timedelta(minutes=new_expected_minutes)).isoformat()
            trade_to_update['decision_data']['reasoning'] = re_analysis_decision.get('reasoning')
            open_trades = await self.r2_service.get_open_trades_async()
            for i, trade in enumerate(open_trades):
                if trade.get('id') == trade_to_update.get('id'): open_trades[i] = trade_to_update; break
            await self.r2_service.save_open_trades_async(open_trades)
            await self.r2_service.save_system_logs_async({"trade_strategy_updated": True, "symbol": symbol})
            print(f"✅ (Explorer) تم تحديث الأهداف الاستراتيجية لـ {symbol}")
            return True
        except Exception as e: print(f"❌ (Explorer) فشل تحديث استراتيجية {symbol}: {e}"); raise

    async def _archive_closed_trade(self, closed_trade):
        try:
            key = "closed_trades_history.json"; history = []
            try: response = self.r2_service.s3_client.get_object(Bucket="trading", Key=key); history = json.loads(response['Body'].read())
            except Exception: pass
            history.append(closed_trade); history = history[-1000:]
            data_json = json.dumps(history, indent=2).encode('utf-8')
            self.r2_service.s3_client.put_object(Bucket="trading", Key=key, Body=data_json, ContentType="application/json")
        except Exception as e: print(f"❌ Failed to archive trade: {e}")

    async def _update_trade_summary(self, closed_trade):
        try:
            key = "trade_summary.json"; summary = {"total_trades": 0, "winning_trades": 0, "losing_trades": 0, "total_profit_usd": 0.0, "total_loss_usd": 0.0, "win_percentage": 0.0, "avg_profit_per_trade": 0.0, "avg_loss_per_trade": 0.0, "largest_win": 0.0, "largest_loss": 0.0, "last_updated": datetime.now().isoformat()}
            try: response = self.r2_service.s3_client.get_object(Bucket="trading", Key=key); summary = json.loads(response['Body'].read())
            except Exception: pass
            pnl = closed_trade.get('pnl_usd', 0.0); summary['total_trades'] += 1
            if pnl >= 0: summary['winning_trades'] += 1; summary['total_profit_usd'] += pnl; summary['largest_win'] = max(summary.get('largest_win', 0), pnl)
            else: summary['losing_trades'] += 1; summary['total_loss_usd'] += abs(pnl); summary['largest_loss'] = max(summary.get('largest_loss', 0), abs(pnl))
            if summary['total_trades'] > 0: summary['win_percentage'] = (summary['winning_trades'] / summary['total_trades']) * 100
            if summary['winning_trades'] > 0: summary['avg_profit_per_trade'] = summary['total_profit_usd'] / summary['winning_trades']
            if summary['losing_trades'] > 0: summary['avg_loss_per_trade'] = summary['total_loss_usd'] / summary['losing_trades']
            summary['last_updated'] = datetime.now().isoformat()
            data_json = json.dumps(summary, indent=2).encode('utf-8')
            self.r2_service.s3_client.put_object(Bucket="trading", Key=key, Body=data_json, ContentType="application/json")
        except Exception as e: print(f"❌ Failed to update trade summary: {e}")

    async def get_open_trades(self):
        try: return await self.r2_service.get_open_trades_async()
        except Exception as e: print(f"❌ Failed to get open trades: {e}"); return []

    async def get_trade_by_symbol(self, symbol):
        try:
            open_trades = await self.get_open_trades()
            return next((t for t in open_trades if t['symbol'] == symbol and t['status'] == 'OPEN'), None)
        except Exception as e: print(f"❌ Failed to get trade by symbol {symbol}: {e}"); return None


print(f"✅ Trade Manager loaded - V7.0 (5m Detector Profit-Saver + Post-Close Cycle) (ccxt.async_support: {CCXT_ASYNC_AVAILABLE})")