""" core/analysis/m15.py — M15Analyzer und SRDetector """ from __future__ import annotations import threading import time import MetaTrader5 as mt5 from core.config import ( M15_BARS, EMA_FAST, EMA_SLOW, PIVOT_WINDOW, ANGLE_LR_BARS, SR_LOOKBACK, SR_PIVOT_WIN, SR_MIN_TOUCHES, SR_TOL_ATR_FACTOR, SR_MAX_LINES, TL_MIN_PIVOTS, CHART_BARS, ) from core.analysis.indicators import _ema class M15Analyzer: """ Erkennt M15-Trendwenden anhand von 3 Indikatoren: 1. EMA(5)/EMA(13)-Crossover 2. Bullish/Bearish Engulfing 3. Frische Swing-Highs/Lows Mindestens 2 Indikatoren müssen in dieselbe Richtung zeigen. """ def __init__(self, sym): self.symbol = sym self.result = None self.reasons = [] self._lock = threading.Lock() def analyze(self): bars = mt5.copy_rates_from_pos(self.symbol, mt5.TIMEFRAME_M15, 0, M15_BARS) if bars is None or len(bars) < max(EMA_SLOW + 2, PIVOT_WINDOW * 2 + 2): return closes = [float(b["close"]) for b in bars] opens = [float(b["open"]) for b in bars] highs = [float(b["high"]) for b in bars] lows = [float(b["low"]) for b in bars] n = len(bars); w = PIVOT_WINDOW; signals = [] ef = _ema(closes, EMA_FAST) es = _ema(closes, EMA_SLOW) if ef[-2] < es[-2] and ef[-1] > es[-1]: signals.append(("bullish", f"EMA{EMA_FAST}/{EMA_SLOW}")) if ef[-2] > es[-2] and ef[-1] < es[-1]: signals.append(("bearish", f"EMA{EMA_FAST}/{EMA_SLOW}")) if (closes[-2] < opens[-2] and closes[-1] > opens[-1] and closes[-1] > opens[-2] and opens[-1] < closes[-2]): signals.append(("bullish", "Bullish Engulfing")) if (closes[-2] > opens[-2] and closes[-1] < opens[-1] and closes[-1] < opens[-2] and opens[-1] > closes[-2]): signals.append(("bearish", "Bearish Engulfing")) for i in range(n - w - 2, n - 1): h = highs[i] if (all(h > highs[j] for j in range(max(0, i - w), i)) and all(h > highs[j] for j in range(i + 1, min(n, i + w + 1)))): signals.append(("bearish", f"Swing-High (Bar -{n - 1 - i})")) break for i in range(n - w - 2, n - 1): l = lows[i] if (all(l < lows[j] for j in range(max(0, i - w), i)) and all(l < lows[j] for j in range(i + 1, min(n, i + w + 1)))): signals.append(("bullish", f"Swing-Low (Bar -{n - 1 - i})")) break bull = [r for d, r in signals if d == "bullish"] bear = [r for d, r in signals if d == "bearish"] direction = None; reasons = [] if len(bull) >= 2: direction = "bullish"; reasons = bull elif len(bear) >= 2: direction = "bearish"; reasons = bear with self._lock: self.result = direction self.reasons = reasons def snapshot(self): with self._lock: return self.result, list(self.reasons) class SRDetector: """ Findet horizontale Support-/Resistance-Zonen und Trendlinien auf M15. Algorithmus: 1. Swing-Pivots erkennen (lokale Hochs/Tiefs mit Fenster ±SR_PIVOT_WIN) 2. Pivots clustern: Preise innerhalb 0.5×ATR werden zu einer Zone 3. Zonen mit >= SR_MIN_TOUCHES Berührungen → gültiges S/R-Level 4. Trendlinien: lineare Regression durch jüngste Pivot-Lows/Highs """ SR_DETECT_INTERVAL_S = 60 def __init__(self, symbol: str): self.symbol = symbol self.supports = [] self.resistances = [] self.trendline_up = None self.trendline_dn = None self._lock = threading.Lock() self._last_detect_ts: float = 0 @staticmethod def _find_pivots(highs, lows, win): n = len(highs) ph, pl = [], [] for i in range(win, n - win): h = highs[i]; l = lows[i] if (all(h >= highs[j] for j in range(i - win, i)) and all(h >= highs[j] for j in range(i + 1, i + win + 1))): ph.append((i, h)) if (all(l <= lows[j] for j in range(i - win, i)) and all(l <= lows[j] for j in range(i + 1, i + win + 1))): pl.append((i, l)) return ph, pl @staticmethod def _atr(highs, lows, closes, period=14): if len(highs) < period + 1: return None trs = [] for i in range(1, len(highs)): tr = max(highs[i] - lows[i], abs(highs[i] - closes[i - 1]), abs(lows[i] - closes[i - 1])) trs.append(tr) return sum(trs[-period:]) / period @staticmethod def _cluster(pivots, tolerance): if not pivots: return [] prices = sorted(p[1] for p in pivots) clusters = [] current = [prices[0]] for p in prices[1:]: if abs(p - current[-1]) <= tolerance: current.append(p) else: clusters.append(current) current = [p] clusters.append(current) return [(sum(c) / len(c), len(c)) for c in clusters] @staticmethod def _trendline(pivots_recent): if len(pivots_recent) < TL_MIN_PIVOTS: return None xs = [p[0] for p in pivots_recent] ys = [p[1] for p in pivots_recent] n = len(xs) xm = sum(xs) / n ym = sum(ys) / n num = sum((xs[i] - xm) * (ys[i] - ym) for i in range(n)) den = sum((xs[i] - xm) ** 2 for i in range(n)) if den == 0: return None slope = num / den intercept = ym - slope * xm x0 = xs[0]; x1 = xs[-1] return { "start_idx": x0, "end_idx": x1, "start_price": slope * x0 + intercept, "end_price": slope * x1 + intercept, "slope": slope, } def detect(self): now = time.monotonic() if now - self._last_detect_ts < self.SR_DETECT_INTERVAL_S: return self._last_detect_ts = now bars = mt5.copy_rates_from_pos( self.symbol, mt5.TIMEFRAME_M15, 0, SR_LOOKBACK) if bars is None or len(bars) < 2 * SR_PIVOT_WIN + 5: return highs = [float(b["high"]) for b in bars] lows = [float(b["low"]) for b in bars] closes = [float(b["close"]) for b in bars] atr = self._atr(highs, lows, closes) tol = (atr or (max(highs) - min(lows)) / 50) * SR_TOL_ATR_FACTOR ph, pl = self._find_pivots(highs, lows, SR_PIVOT_WIN) zones_high = [(p, n) for p, n in self._cluster(ph, tol) if n >= SR_MIN_TOUCHES] zones_low = [(p, n) for p, n in self._cluster(pl, tol) if n >= SR_MIN_TOUCHES] cur = closes[-1] resistances = sorted( [{"price": p, "touches": n} for p, n in zones_high if p > cur], key=lambda z: z["price"])[:SR_MAX_LINES] supports = sorted( [{"price": p, "touches": n} for p, n in zones_low if p < cur], key=lambda z: -z["price"])[:SR_MAX_LINES] recent_cutoff = len(bars) - 50 recent_lows = [p for p in pl if p[0] >= recent_cutoff] recent_highs = [p for p in ph if p[0] >= recent_cutoff] tl_up = self._trendline(recent_lows[-3:]) if len(recent_lows) >= 2 else None tl_dn = self._trendline(recent_highs[-3:]) if len(recent_highs) >= 2 else None if tl_up and tl_up["slope"] <= 0: tl_up = None if tl_dn and tl_dn["slope"] >= 0: tl_dn = None n_bars = len(bars) offset = n_bars - CHART_BARS def _remap(tl): if tl is None: return None si = tl["start_idx"] - offset ei = tl["end_idx"] - offset if tl["slope"] is not None and ei < CHART_BARS - 1: extra = (CHART_BARS - 1) - tl["end_idx"] ep = tl["end_price"] + tl["slope"] * extra ei = CHART_BARS - 1 else: ep = tl["end_price"] if si < 0: sp = tl["start_price"] + tl["slope"] * (offset - tl["start_idx"]) si = 0 else: sp = tl["start_price"] return {"start_idx": si, "end_idx": ei, "start_price": sp, "end_price": ep} with self._lock: self.supports = supports self.resistances = resistances self.trendline_up = _remap(tl_up) self.trendline_dn = _remap(tl_dn) def snapshot(self): with self._lock: return { "supports": list(self.supports), "resistances": list(self.resistances), "trendline_up": dict(self.trendline_up) if self.trendline_up else None, "trendline_dn": dict(self.trendline_dn) if self.trendline_dn else None, }