AH-Oil-Trader/core/mt5data.py

"""
core/mt5data.py — MT5Data
===========================
Daten-Adapter: holt Ticks, Bars, Kontoinfo und Multi-Timeframe-Analyse aus MT5.
"""

from __future__ import annotations
import threading
import time
from datetime import datetime, timezone
from concurrent.futures import ThreadPoolExecutor

import MetaTrader5 as mt5

from core.config import (
    SYMBOL_CANDIDATES, CHART_BARS, ANGLE_LR_BARS,
)
from core.mt5_utils import mt5_lock
from core.analysis import (calc_trend_angle, calc_rsi, calc_atr,
                            M15Analyzer, SRDetector)
from core.logger import get_logger

log_mt5 = get_logger("mt5")


class MT5Data:
    def __init__(self):
        self.symbol = None
        self.bid = self.ask = self.spread = None
        self.change = self.pct = self.day_high = self.day_low = None
        self.balance = self.equity = None
        self.currency = "USD"; self.error = None
        self.trend_angle = 90.0
        self.angles = {"M5": 90.0, "M15": 90.0, "M30": 90.0, "H1": 90.0}
        self._angle_ema: dict = {}   # geglättete Winkel (EMA α=0.4)
        self.rsi_m15:      float | None = None
        self.atr_m15:      float | None = None
        self.server_time:    datetime | None = None
        self.tick_local_ts:  float = 0.0
        self.tick_server_ts: int   = 0
        self._slow_price_ts: float = 0.0   # letzter D1-/Konto-Fetch (Throttle)
        self.sessions: dict = {}
        self._lock = threading.Lock()
        self._connected = False
        self._bad_tick_streak: int = 0
        self.BAD_TICK_LIMIT: int = 10
        self.analyzer = None; self.sr = None
        self._analyze_executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="analyzer")

    def connect(self, preferred_symbol: str | None = None) -> bool:
        if not mt5.initialize():
            with self._lock: self.error = f"MT5: {mt5.last_error()}"; return False
        acc = mt5.account_info()
        if not acc:
            with self._lock: self.error = "MT5 nicht eingeloggt"; return False
        pref       = (preferred_symbol or "").strip()
        candidates = ([pref] if pref else []) + \
                     [s.strip() for s in SYMBOL_CANDIDATES if s.strip() != pref]
        for sym in candidates:
            if not sym: continue
            if mt5.symbol_info(sym) is not None:
                mt5.symbol_select(sym, True)
                with self._lock:
                    self.symbol = sym; self.currency = acc.currency
                self._connected = True
                self.analyzer = M15Analyzer(sym)
                self.sr       = SRDetector(sym)
                log_mt5.info(f"Konto {acc.login} | {acc.currency} | Symbol: {sym}")
                self._load_sessions(sym)
                return True
        with self._lock: self.error = "Kein WTI-/SpotCrude-Symbol"; return False

    def switch_symbol(self, new_sym: str) -> tuple[bool, str]:
        new_sym = (new_sym or "").strip()
        if not new_sym:
            return False, "Symbol-Name leer"
        try:
            with mt5_lock(timeout=10) as got:
                if not got:
                    return False, "MT5-Lock belegt"
                info = mt5.symbol_info(new_sym)
                if info is None:
                    return False, f"Symbol '{new_sym}' nicht beim Broker"
                if not mt5.symbol_select(new_sym, True):
                    return False, f"symbol_select('{new_sym}') fehlgeschlagen"
                new_analyzer = M15Analyzer(new_sym)
                new_sr       = SRDetector(new_sym)
            with self._lock:
                self.symbol = new_sym; self.analyzer = new_analyzer; self.sr = new_sr
                self.m15_bars = []; self.ema_fast = []; self.ema_slow = []
                self.trend_angle = 90.0
                self.angles = {"M5": 90.0, "M15": 90.0, "M30": 90.0, "H1": 90.0}
                self.rsi_m15 = None; self.atr_m15 = None
                self.coc = None; self.sma50_h1 = None; self.vwap = None
                self.error = None; self._bad_tick_streak = 0
            self._load_sessions(new_sym)
            log_mt5.info(f"Symbol gewechselt: {new_sym}")
            return True, f"Symbol jetzt: {new_sym}"
        except Exception as e:
            log_mt5.error(f"switch_symbol({new_sym}): {e}", exc_info=True)
            return False, f"Fehler: {e}"

    # Fallback-Sessionsplan für Rohöl (UTC) — greift wenn MT5 keine Sessions liefert.
    # Mon–Fr 00:00–22:00, So ab 22:00. Broker-Zeiten können minimal abweichen.
    _OIL_SESSIONS_UTC = {
        0: [(79200, 86400)],  # So  22:00–24:00
        1: [(0, 79200)],      # Mo  00:00–22:00
        2: [(0, 79200)],      # Di
        3: [(0, 79200)],      # Mi
        4: [(0, 79200)],      # Do
        5: [(0, 79200)],      # Fr  00:00–22:00
        6: [],                # Sa  geschlossen
    }

    def _load_sessions(self, sym: str):
        sessions = {}
        api_ok = False
        for day in range(7):
            try:
                raw = mt5.symbol_info_sessions_trade(sym, day)
                if raw is not None:
                    sessions[day] = [(int(s.from_), int(s.to)) for s in raw]
                    api_ok = True
                else:
                    sessions[day] = []
            except Exception:
                sessions[day] = []
        if not api_ok:
            sessions = dict(self._OIL_SESSIONS_UTC)
            log_mt5.debug("Sessions-API nicht verfügbar — Fallback auf Öl-Standard (UTC)")
        with self._lock:
            self.sessions = sessions

    def reconnect(self) -> bool:
        # ALLE MT5-Calls (account_info/shutdown/initialize) MÜSSEN unter dem
        # globalen Lock laufen — sonst racet ein shutdown()/initialize() gegen
        # copy_rates/positions_get der anderen Loops (MT5-Lib ist nicht
        # thread-safe → Crash/Garbage). reconnect() wird stets OHNE gehaltenen
        # Lock aufgerufen (aus fetch_price/fetch_trend vor deren with-Block).
        with mt5_lock(timeout=10) as got:
            if not got:
                with self._lock: self.error = "Reconnect: MT5-Lock belegt"
                return False
            try:
                if mt5.account_info():
                    with self._lock: self._connected = True; self.error = None
                    log_mt5.info("MT5 noch verbunden — kein Hard-Reconnect nötig")
                    return True
            except Exception:
                pass
            log_mt5.info("MT5 Hard-Reconnect …")
            try: mt5.shutdown()
            except Exception: pass
            if not mt5.initialize():
                with self._lock:
                    self.error = f"Reconnect fehlgeschlagen: {mt5.last_error()}"
                    self._connected = False
                return False
            if not mt5.account_info():
                with self._lock:
                    self.error = "MT5 nach Reconnect nicht eingeloggt"; self._connected = False
                return False
        with self._lock: self._connected = True; self.error = None
        log_mt5.info("MT5-Reconnect erfolgreich")
        return True

    def fetch_price(self):
        if not self._connected:
            self.reconnect(); return
        with mt5_lock() as got:
            if not got: return
            self._fetch_price_locked()

    def _fetch_price_locked(self):
        # Hot-Path (500 ms): nur den Tick lesen — minimale Lock-Haltezeit,
        # damit Bid/Ask (und damit die live-P&L) auch unter Lock-Konkurrenz
        # (Trailing-order_send, fetch_trend) frisch bleiben.
        sym  = self.symbol
        tick = mt5.symbol_info_tick(sym)
        if not tick or tick.bid <= 0:
            with self._lock:
                self._bad_tick_streak += 1; self.error = f"Kein Tick: {sym}"
                if self._bad_tick_streak >= self.BAD_TICK_LIMIT:
                    self._connected = False; self._bad_tick_streak = 0
                    log_mt5.warning(f"{self.BAD_TICK_LIMIT} Bad-Ticks → reconnect")
            return
        bid    = float(tick.bid); ask = float(tick.ask); mid = (bid + ask) / 2
        spread = round(ask - bid, 5)
        srv_time = datetime.fromtimestamp(tick.time, tz=timezone.utc)

        # Cold-Path (alle ~2 s): D1-Bars (Change/Tageshoch/-tief) + Kontoinfo.
        # Nicht P&L-kritisch → seltener holen, hält den Lock kürzer frei.
        now = time.time()
        do_slow = (now - self._slow_price_ts) > 2.0
        prev = dh = dl = acc = None
        if do_slow:
            bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_D1, 0, 2)
            if bars is not None and len(bars) >= 2:
                prev = float(bars[0]["close"])
                dh   = float(bars[1]["high"]); dl = float(bars[1]["low"])
            acc = mt5.account_info()
            self._slow_price_ts = now

        with self._lock:
            self._bad_tick_streak = 0
            self.bid = bid; self.ask = ask; self.spread = spread
            _prev_srv = self.server_time
            if do_slow:
                if prev:
                    self.change = mid - prev
                    self.pct    = (self.change / prev * 100) if prev else None
                self.day_high = dh if dh is not None else self.day_high
                self.day_low  = dl if dl is not None else self.day_low
                if acc:
                    self.balance  = float(acc.balance)
                    self.equity   = float(acc.equity)
                    self.currency = acc.currency
            prev_srv_ts   = int(_prev_srv.timestamp()) if _prev_srv else 0
            tick_advanced = int(tick.time) > prev_srv_ts
            self.server_time    = srv_time
            self.tick_server_ts = int(tick.time)   # echter Broker-Timestamp
            if tick_advanced: self.tick_local_ts = now
            self.error = None

    def fetch_trend(self):
        if not self._connected:
            self.reconnect(); return
        with mt5_lock() as got:
            if not got: return
            self._fetch_trend_locked()

    def _fetch_trend_locked(self):
        # Nur noch die tatsächlich konsumierten Werte berechnen:
        # Trendwinkel (Trailing/Reversal), RSI/ATR (Auto-Trader/Logging),
        # M15-Analyzer (Reversal) und S/R (Trailing). Die früheren
        # ICT-/Chart-Berechnungen (Fib, BOS, FVG, OB, Ichimoku, VWAP, EMAs …)
        # fütterten nur die entfernte Empfehlungs-Engine.
        sym    = self.symbol
        needed = CHART_BARS + 20
        bars   = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M15, 0, needed)
        if bars is not None and len(bars) >= CHART_BARS:
            closes = [float(b["close"]) for b in bars]
            highs  = [float(b["high"])  for b in bars]
            lows   = [float(b["low"])   for b in bars]
            ang    = calc_trend_angle(closes)
            rsi    = calc_rsi(closes, 14)
            atr    = calc_atr(highs, lows, closes, 14)
            with self._lock:
                self.trend_angle = ang; self.rsi_m15 = rsi; self.atr_m15 = atr
        tf_map = [("M5", mt5.TIMEFRAME_M5, ANGLE_LR_BARS),
                  ("M15", mt5.TIMEFRAME_M15, ANGLE_LR_BARS),
                  ("M30", mt5.TIMEFRAME_M30, ANGLE_LR_BARS),
                  ("H1",  mt5.TIMEFRAME_H1,  ANGLE_LR_BARS)]
        raw_angles = {}
        for label, tf, lr in tf_map:
            b = mt5.copy_rates_from_pos(sym, tf, 0, lr + 5)
            raw_angles[label] = calc_trend_angle([float(x["close"]) for x in b], lr) \
                                if b is not None and len(b) >= lr else 90.0
        # EMA-Glättung α=0.4 — dämpft Tick-Rauschen ohne echte Trendwenden zu verzögern
        alpha = 0.4
        with self._lock:
            for lbl, raw in raw_angles.items():
                prev = self._angle_ema.get(lbl, raw)
                self._angle_ema[lbl] = alpha * raw + (1 - alpha) * prev
            self.angles = dict(self._angle_ema)

        if self.analyzer:
            _az = self.analyzer
            # analyze() läuft im eigenen Thread → MUSS den globalen mt5_lock selbst
            # nehmen (MT5-Lib ist nicht thread-safe; sonst Race gegen alle anderen
            # MT5-Calls). Eigener Thread, kein Deadlock mit dem hier gehaltenen Lock.
            def _locked_analyze(az=_az):
                with mt5_lock(timeout=3) as got:
                    if got:
                        az.analyze()
            self._analyze_executor.submit(_locked_analyze)
        if self.sr:       self.sr.detect()   # inline unter gehaltenem Lock → ok

    def snapshot(self) -> dict:
        with self._lock:
            d = dict(
                symbol=self.symbol, bid=self.bid, ask=self.ask, spread=self.spread,
                change=self.change, pct=self.pct, day_high=self.day_high, day_low=self.day_low,
                balance=self.balance, equity=self.equity, currency=self.currency, error=self.error,
                trend_angle=self.trend_angle,
                angles=dict(self.angles), rsi_m15=self.rsi_m15, atr_m15=self.atr_m15,
                server_time=self.server_time, tick_local_ts=self.tick_local_ts,
                tick_server_ts=self.tick_server_ts,
                sessions=dict(self.sessions),
            )
        rev, reasons = self.analyzer.snapshot() if self.analyzer else (None, [])
        d["reversal"] = rev; d["reasons"] = reasons
        d["sr"] = self.sr.snapshot() if self.sr else None
        return d

    def disconnect(self):
        mt5.shutdown()