yuclusters-mt5/backend/mt5_collector.py

import asyncio
import logging
import time
import sys
import os
from typing import Callable, Optional, Any

sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))

import MetaTrader5 as mt5
from config import settings
from backend.aggregator import Aggregator, classify_tick

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("mt5_collector")

class MT5Collector:
    def __init__(self, aggregator: Aggregator, on_update_callback: Callable[[dict, Optional[dict], Optional[list]], Any], on_history_ready_callback: Callable = None, on_big_trade_callback: Optional[Callable[[dict], Any]] = None):
        self.aggregator = aggregator
        self.on_update_callback = on_update_callback
        self.on_history_ready_callback = on_history_ready_callback
        self.on_big_trade_callback = on_big_trade_callback
        self.symbol = settings.MT5_SYMBOL
        self.running = False
        self.connected = False
        self.historical_mode = False
        self.last_tick_time_msc = 0
        self.seen_ticks_buffer = set()
        self.last_mid_price = 0.0
        self.last_bid_price = 0.0
        self.last_is_buy = True
        self.last_bid = 0.0
        self.last_ask = 0.0
        self._history_annotated = False
        self._replay_reset_requested = False

    async def connect_mt5(self) -> bool:
        """
        Attempts to initialize and login to the MetaTrader 5 terminal.
        All MT5 calls are blocking, so they run in a thread executor.
        """
        try:
            # Try to connect to any running MT5 terminal without specifying path
            initialized = await asyncio.to_thread(mt5.initialize)
            if not initialized:
                err = await asyncio.to_thread(mt5.last_error)
                logger.error(f"MT5 initialize failed: {err}")
                return False

            if settings.MT5_LOGIN > 0:
                login_success = await asyncio.to_thread(
                    mt5.login,
                    settings.MT5_LOGIN,
                    password=settings.MT5_PASSWORD,
                    server=settings.MT5_SERVER,
                )
                if not login_success:
                    err = await asyncio.to_thread(mt5.last_error)
                    logger.error(f"MT5 login failed: {err}")
                    await asyncio.to_thread(mt5.shutdown)
                    return False

            symbol_info = await asyncio.to_thread(mt5.symbol_info, self.symbol)
            if symbol_info is None:
                logger.error(f"Symbol {self.symbol} not found.")
                await asyncio.to_thread(mt5.shutdown)
                return False

            if not symbol_info.visible:
                selected = await asyncio.to_thread(mt5.symbol_select, self.symbol, True)
                if not selected:
                    logger.error(f"Failed to select/make visible symbol {self.symbol}.")
                    await asyncio.to_thread(mt5.shutdown)
                    return False

            tick_size = symbol_info.trade_tick_size
            if tick_size > 0:
                self.aggregator.tick_size = tick_size
                self.aggregator.active_cluster.tick_size = tick_size
                logger.info(f"Set aggregator tick size to {tick_size}")

            # Subscribe to Depth of Market (DOM)
            book_added = await asyncio.to_thread(mt5.market_book_add, self.symbol)
            if not book_added:
                logger.warning(f"Failed to subscribe to market book (DOM) for {self.symbol}")

            logger.info("Successfully connected to MetaTrader 5 and logged in.")
            self.connected = True
            return True
        except Exception as e:
            logger.error(f"Exception during MT5 connection: {e}")
            return False

    async def disconnect_mt5(self):
        try:
            await asyncio.to_thread(mt5.market_book_release, self.symbol)
            await asyncio.to_thread(mt5.shutdown)
        except Exception as e:
            logger.error(f"Error during MT5 shutdown: {e}")
        self.connected = False

    async def _find_last_session_open(self):
        """
        Detecta o início da última sessão de mercado buscando o maior gap
        nos últimos N bars M1. Um gap > 30min indica fechamento de sessão.
        Retorna o datetime do primeiro bar após o gap (abertura de sessão).
        """
        from datetime import datetime, timedelta
        SESSION_GAP_MINUTES = 30
        LOOKBACK_BARS = 3000  # ~50h de M1

        rates = await asyncio.to_thread(
            mt5.copy_rates_from_pos, self.symbol, mt5.TIMEFRAME_M1, 0, LOOKBACK_BARS
        )

        if rates is None or len(rates) < 2:
            logger.warning("_find_last_session_open: sem bars M1, usando HISTORY_HOURS")
            return datetime.now() - timedelta(hours=settings.HISTORY_HOURS)

        # Percorre de trás para frente procurando o maior gap (fechamento de sessão)
        best_gap = 0
        session_open_ts = None
        for i in range(len(rates) - 1, 0, -1):
            gap_sec = int(rates[i]['time']) - int(rates[i - 1]['time'])
            if gap_sec > best_gap:
                best_gap = gap_sec
                session_open_ts = int(rates[i]['time'])

        if session_open_ts and best_gap >= SESSION_GAP_MINUTES * 60:
            dt = datetime.fromtimestamp(session_open_ts)
            logger.info(f"Última abertura de sessão detectada: {dt} (gap de {best_gap//60}min)")
            return dt
        else:
            logger.warning("Nenhum gap de sessão encontrado, usando HISTORY_HOURS")
            return datetime.now() - timedelta(hours=settings.HISTORY_HOURS)

    def request_replay_reset(self):
        """Signal the polling loop to reset the aggregator and replay from session start."""
        self.aggregator.history.clear()
        self.aggregator.active_cluster = type(self.aggregator.active_cluster)(tick_size=self.aggregator.tick_size)
        self._history_annotated = False
        self._replay_reset_requested = True

    async def start(self):
        self.running = True
        backoff = 1.0

        while self.running:
            # Config changed — restart replay from session start with new settings
            if self._replay_reset_requested:
                self._replay_reset_requested = False
                self.connected = False
                logger.info("Replay reset requested — restarting from session start...")
                continue

            if not self.connected:
                success = await self.connect_mt5()
                if not success:
                    logger.info(f"Reconnecting to MT5 in {backoff:.1f}s...")
                    await asyncio.sleep(backoff)
                    backoff = min(backoff * 2, 60.0)
                    continue
                else:
                    backoff = 1.0

                    from datetime import datetime, timedelta
                    if settings.HISTORY_FROM_DATE:
                        start_time_dt = datetime.strptime(settings.HISTORY_FROM_DATE, "%Y.%m.%d")
                    elif settings.HISTORY_SESSION_START:
                        start_time_dt = await self._find_last_session_open()
                    else:
                        start_time_dt = datetime.now() - timedelta(hours=settings.HISTORY_HOURS)
                    ticks = await asyncio.to_thread(
                        mt5.copy_ticks_from, self.symbol, start_time_dt, 100000, mt5.COPY_TICKS_ALL
                    )
                    if ticks is not None and len(ticks) > 0:
                        self.last_tick_time_msc = ticks[0]['time_msc']
                    else:
                        self.last_tick_time_msc = int(start_time_dt.timestamp() * 1000)

            # Polling loop
            try:
                if self.historical_mode:
                    await asyncio.sleep(1)
                    continue

                from datetime import datetime
                polling_dt = datetime.fromtimestamp(self.last_tick_time_msc / 1000.0)
                ticks = await asyncio.to_thread(
                    mt5.copy_ticks_from,
                    self.symbol,
                    polling_dt,
                    1000,
                    mt5.COPY_TICKS_ALL,
                )

                if ticks is None:
                    err = await asyncio.to_thread(mt5.last_error)
                    logger.error(f"MT5 copy_ticks_from returned None: {err}")
                    self.connected = False
                    await self.disconnect_mt5()
                    continue

                # Annotate history once replay is complete (first under-full batch = caught up)
                if not self._history_annotated and len(ticks) < 1000:
                    self._history_annotated = True
                    await self.annotate_history_bar_volume()
                    if self.on_history_ready_callback:
                        self.on_history_ready_callback()

                if len(ticks) > 0:
                    logger.info(f"Fetched {len(ticks)} ticks starting at {ticks[0]['time_msc']}")
                    for tick in ticks:
                        msc = tick['time_msc']

                        if msc < self.last_tick_time_msc:
                            continue

                        tick_id = (msc, tick['bid'], tick['ask'], tick['last'], tick['volume_real'], tick['flags'])
                        if msc == self.last_tick_time_msc and tick_id in self.seen_ticks_buffer:
                            continue

                        if msc > self.last_tick_time_msc:
                            self.seen_ticks_buffer.clear()
                            self.last_tick_time_msc = msc

                        self.seen_ticks_buffer.add(tick_id)

                        flags = int(tick['flags'])
                        bid_price = float(tick['bid'])
                        ask_price = float(tick['ask'])
                        last_price = float(tick['last'])

                        if bid_price > 0: self.last_bid = bid_price
                        if ask_price > 0: self.last_ask = ask_price
                        mid_price = (bid_price + ask_price) / 2.0 if (bid_price > 0 and ask_price > 0) else 0.0
                        prev_mid = self.last_mid_price
                        prev_bid = self.last_bid_price

                        # Update direction tracker from bid movement (YuCluster uses Bid as price reference)
                        if bid_price > 0:
                            if bid_price > self.last_bid_price:
                                self.last_is_buy = True
                            elif bid_price < self.last_bid_price:
                                self.last_is_buy = False
                            self.last_bid_price = bid_price
                        elif mid_price > 0:
                            if mid_price > self.last_mid_price:
                                self.last_is_buy = True
                            elif mid_price < self.last_mid_price:
                                self.last_is_buy = False
                        if mid_price > 0:
                            self.last_mid_price = mid_price

                        # Determine price (use mid as best proxy for CFD quote feed)
                        price = last_price if last_price > 0 else (mid_price if mid_price > 0 else (bid_price if bid_price > 0 else ask_price))

                        # Volume = bid price movement in tick-size units (YuCluster: "Ticks & Bid")
                        tick_sz = self.aggregator.tick_size if self.aggregator.tick_size > 0 else 0.01
                        if prev_bid > 0 and bid_price > 0:
                            price_steps = abs(bid_price - prev_bid) / tick_sz
                            volume = max(price_steps, 1.0)
                        elif prev_mid > 0 and mid_price > 0:
                            price_steps = abs(mid_price - prev_mid) / tick_sz
                            volume = max(price_steps, 1.0)
                        else:
                            volume = 1.0

                        # Determine direction
                        if flags & 32:
                            is_buy = True
                        elif flags & 64:
                            is_buy = False
                        else:
                            is_buy = self.last_is_buy

                        # Broadcast Big Trades instantly
                        if volume >= settings.BIG_TRADE_THRESHOLD and self.on_big_trade_callback:
                            import time as _time
                            is_live = ((_time.time() * 1000) - msc) < 10_000
                            if is_live:
                                self.on_big_trade_callback({
                                    "price": price,
                                    "volume": volume,
                                    "is_buy": is_buy,
                                    "time_msc": msc
                                })

                        active_json, closed_json = self.aggregator.process_tick(price, volume, is_buy, msc)

                        # Only broadcast during live trading (within 10s of now) to avoid
                        # flooding the WebSocket during historical replay
                        import time as _time
                        is_live = ((_time.time() * 1000) - msc) < 10_000

                        # Annotate live closed clusters with M1 bar volume
                        if is_live and closed_json and closed_json.get('open_time') and closed_json.get('close_time'):
                            bar_vol = await self.fetch_bar_volume(
                                closed_json['open_time'], closed_json['close_time']
                            )
                            closed_json['bar_volume'] = bar_vol
                            # Sync back to history buffer
                            for h in self.aggregator.history:
                                if h['cluster_id'] == closed_json['cluster_id']:
                                    h['bar_volume'] = bar_vol
                                    break

                        if is_live:
                            active_json['bid'] = self.last_bid
                            active_json['ask'] = self.last_ask
                            
                            # Fetch DOM
                            dom_json = None
                            try:
                                book_items = await asyncio.to_thread(mt5.market_book_get, self.symbol)
                                if book_items:
                                    dom_json = [
                                        {"type": item.type, "price": item.price, "volume": item.volume}
                                        for item in book_items
                                    ]
                            except Exception as e:
                                logger.error(f"Error fetching DOM: {e}")

                            self.on_update_callback(active_json, closed_json, dom_json)

                await asyncio.sleep(0.1)

            except Exception as e:
                logger.error(f"Error in polling loop: {e}")
                import traceback
                logger.error(traceback.format_exc())
                await asyncio.sleep(1.0)

    async def load_historical_range(self, start_dt, end_dt):
        """
        Pauses live polling, clears the aggregator, fetches a specific historical date range,
        processes all ticks into clusters, and broadcasts the completed history to clients.
        """
        self.historical_mode = True
        logger.info(f"Loading historical data from {start_dt} to {end_dt} for {self.symbol}...")
        
        # Clear existing data
        self.aggregator.history.clear()
        self.aggregator.active_cluster = None
        self.aggregator._create_new_cluster()
        
        # We need to fetch ticks in range.
        ticks = await asyncio.to_thread(
            mt5.copy_ticks_range,
            self.symbol,
            start_dt,
            end_dt,
            mt5.COPY_TICKS_ALL
        )
        
        if ticks is None or len(ticks) == 0:
            logger.warning(f"No historical ticks found for {self.symbol} in the requested range.")
            self.on_update_callback(self.aggregator.active_cluster.to_json(), None, [])
            return

        logger.info(f"Fetched {len(ticks)} historical ticks. Processing...")
        
        # Process ticks without broadcasting every tick
        for tick in ticks:
            msc = tick['time_msc']
            flags = int(tick['flags'])
            bid_price = float(tick['bid'])
            ask_price = float(tick['ask'])
            volume = float(tick['volume_real'])
            price = float(tick['last'])

            if price <= 0.0 or volume <= 0.0:
                continue

            if bid_price > 0: self.last_bid = bid_price
            if ask_price > 0: self.last_ask = ask_price
            mid_price = (bid_price + ask_price) / 2.0 if (bid_price > 0 and ask_price > 0) else 0.0

            if mid_price > 0:
                if mid_price > self.last_mid_price:
                    self.last_is_buy = True
                elif mid_price < self.last_mid_price:
                    self.last_is_buy = False
                self.last_mid_price = mid_price

            is_buy = self.last_is_buy
            if flags & 32:
                is_buy = True
            elif flags & 64:
                is_buy = False

            # We don't trigger big trade alerts during history load to avoid spam
            self.aggregator.process_tick(price, volume, is_buy, msc)

        # Apply bar-level volume annotation
        await self.annotate_history_bar_volume()
        
        # Broadcast the reconstructed history
        active_json = self.aggregator.active_cluster.to_json() if self.aggregator.active_cluster else None
        closed_json = [c.to_json() for c in self.aggregator.history]
        self.on_update_callback(active_json, None, closed_json)
        logger.info("Historical data loaded and broadcasted.")

    def return_to_live(self):
        """Resumes live polling from the current time."""
        self.historical_mode = False
        from datetime import datetime
        self.last_tick_time_msc = int(datetime.now().timestamp() * 1000)
        self.aggregator.history.clear()
        self.aggregator.active_cluster = None
        self.aggregator._create_new_cluster()
        self._history_annotated = False
        logger.info("Returned to live polling.")

    async def fetch_bar_volume(self, open_time_msc: int, close_time_msc: int) -> int:
        """Sum tick_volume of M1 bars that overlap with the cluster's time range."""
        from datetime import datetime
        # Expand range by 1 minute on each side to capture partial bars
        open_dt  = datetime.fromtimestamp((open_time_msc  - 60_000) / 1000.0)
        close_dt = datetime.fromtimestamp((close_time_msc + 60_000) / 1000.0)
        rates = await asyncio.to_thread(
            mt5.copy_rates_range, self.symbol, mt5.TIMEFRAME_M1, open_dt, close_dt
        )
        if rates is None or len(rates) == 0:
            return 0
        total = 0
        for r in rates:
            bar_start_msc = int(r['time']) * 1000
            bar_end_msc   = bar_start_msc + 60_000
            # Count bar if it overlaps with cluster period
            if bar_start_msc < close_time_msc and bar_end_msc > open_time_msc:
                total += int(r['tick_volume'])
        return total

    async def annotate_history_bar_volume(self):
        """Batch-fetch M1 bars and annotate all history clusters with bar_volume."""
        if not self.aggregator.history:
            return
        first_open  = self.aggregator.history[0].get('open_time') or 0
        last_close  = self.aggregator.history[-1].get('close_time') or self.aggregator.history[-1].get('open_time') or 0
        if not first_open or not last_close:
            return
        from datetime import datetime
        open_dt  = datetime.fromtimestamp((first_open  - 60_000) / 1000.0)
        close_dt = datetime.fromtimestamp((last_close  + 60_000) / 1000.0)
        all_rates = await asyncio.to_thread(
            mt5.copy_rates_range, self.symbol, mt5.TIMEFRAME_M1, open_dt, close_dt
        )
        if all_rates is None or len(all_rates) == 0:
            logger.warning("annotate_history_bar_volume: no M1 bars returned")
            return
        for cluster in self.aggregator.history:
            c_open  = cluster.get('open_time')  or 0
            c_close = cluster.get('close_time') or c_open
            total = 0
            for r in all_rates:
                bar_start_msc = int(r['time']) * 1000
                bar_end_msc   = bar_start_msc + 60_000
                if bar_start_msc < c_close and bar_end_msc > c_open:
                    total += int(r['tick_volume'])
            cluster['bar_volume'] = total
        logger.info(f"annotate_history_bar_volume: annotated {len(self.aggregator.history)} clusters")

    async def stop(self):
        self.running = False
        await self.disconnect_mt5()
        logger.info("MT5 Collector stopped.")