""" 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()