""" core/wave_rec.py — Trend-Empfehlungsmodul (EMA-Trend) ====================================================== Signalrichtung folgt dem geglätteten Trend (EMA12 vs EMA50) der gewählten Zeitebene — NICHT mehr dem verrauschten ZigZag-Swing. Grund: Backtests zeigten, dass das alte Momentum-ZigZag ~0 Edge hatte (kaufte die Spitzen von Bounces), während EMA-Trendfolge einen klaren, positiven Edge liefert. Signal: EMA12 > EMA50 → LONG | EMA12 < EMA50 → SHORT |EMA12−EMA50| < _TREND_DEADBAND×ATR → WARTEN (kein klarer Trend / Chop) Kurs > _STRETCH_MAX×ATR über/unter EMA50 → WARTEN (überdehnt, kein Spät-Einstieg) Konfidenz: 60 % Basis, +Trendstärke (EMA-Abstand in ATR), +Einstiegsqualität (nahe der EMA = besseres CRV, kein Hinterherkaufen), +Traders-Union-Bestätigung (5m/15m); stark gegenläufige TU blockt. Arbeitsteilung (Thread-sicher): set_timeframe(tf) → vom KI-Agenten gewählt (M1/M5/M15/M30/H1) refresh_market(sym) → alle ~5 s im Hintergrund (MT5-Call: Bars + EMAs) signal() → aktueller, fertig gefilterter rec-Dict (kein MT5) snapshot() → Trend-Status für die Anzeige (EMA als wave_start, Abstand zur EMA in ATR als move_atr) """ from __future__ import annotations import threading import time from datetime import datetime from zoneinfo import ZoneInfo import MetaTrader5 as mt5 from core.mt5_utils import mt5_lock from core.market_hours import session_state from core.analysis import calc_trend_angle from core.logger import get_logger log = get_logger("wave") _N_BARS = 180 _ATR_PERIOD = 14 _STALE_S = 150 _BASE_CONF = 60 _MIN_CONF = 55 # Signale < 55 % Konfidenz → WARTEN. Gemessen # (backtest_conf.py): Band 40–54 % = NEGATIVER Edge # (−0,025), ≥55 % = +0,058. Gate hebt Edge + Ertrag. # Signalrichtung = geglätteter EMA-Trend (per Backtest mit Edge belegt; das alte # ZigZag-Momentum hatte ~0 Edge / kaufte Spitzen). Totband filtert Chop, die # Anti-Überdehnung verhindert Spät-Einstiege weit weg von der EMA. _EMA_FAST = 12 _EMA_SLOW = 50 _TREND_DEADBAND = 0.15 # |EMA12 − EMA50| < x×ATR → kein klarer Trend → WARTEN _STRETCH_MAX = 3.5 # Kurs > x×ATR über/unter EMA50 → überdehnt → kein Einstieg. # 2,5→3,5 (backtest_stretch.py): +26 % Signale (45→57 % der # Zeit), Bänder 2,5–3,5 noch positiv, Gesamtertrag steigt. # NICHT höher: ab 3,5×ATR kippt der Edge klar negativ (−0,13). _REVERSAL_STRETCH = 3.0 # Bounce/Reversal-Trigger — ENTKOPPELT von _STRETCH_MAX: # ab |stretch|≥3,0×ATR + Winkeldrehung antizyklischer # Einstieg (mehr Bounces als bei 3,5). Gemessen # (backtest_bounce.py, Exit-Sim): Ø-R +0,185, PF 1,35, # Treffer 70 %, ~1,5× mehr Bounces als 3,5. Anti-Über- # dehnung (Trend-Spät-Sperre) bleibt bei _STRETCH_MAX=3,5. # Volatilitäts-Squeeze-Breakout (additiver Setup, gemessen `backtest_breakout_squeeze.py`, # 2-Stichproben-positiv & robust über Box 10–20 / k 0,1–0,2): Box = Spanne der letzten # _SQ_N abgeschlossenen M5-Bars; ist sie ≤ _SQ_MULT×ATR (Kompression) und bricht der # Kurs _SQ_K×ATR über/unter die Box → Ausbruchssignal. Live-Params = die getesteten. _SQ_N = 10 # Box-Länge (M5-Bars) _SQ_MULT = 2.5 # Squeeze: Box ≤ _SQ_MULT×ATR = Kompression _SQ_K = 0.1 # Ausbruch k×ATR über/unter die Box-Grenze _SQ_REFRESH_S = 5.0 # eigener, schneller M5-Fetch NUR für den Squeeze (entkoppelt von # der 15-s-Ampel-Drossel) → Ausbruch-Erkennung ~15 s→5 s, gleiche # validierte M5-Regel (Variante A, 2026-07-21). # Higher-TF-Gegen-Trend-Filter: Signal NUR, wenn der übergeordnete Trend (M30, # EMA12 vs EMA50) nicht klar dagegen steht. Per Backtest belegt: verdoppelt den # Ø-Edge pro Trade (+0,012 → +0,027), Treffer 52→54 %; wirft Gegen-Trend-Signale # raus (z. B. Short im M30-Aufwärtstrend). Nur für Basis-TF < M30 aktiv. _HTF = mt5.TIMEFRAME_M30 _HTF_LABEL = "M30" _HTF_DEADBAND = 0.15 # |EMA12−EMA50| < x×ATR(M30) → HTF gilt als neutral # Multi-TF-Konfluenz (nur Konfidenz/Anzeige): wenn M30 UND H1 die Richtung # bestätigen, ver-3,5-facht sich der Ø-Edge (+0,016→+0,056, backtest_improve.py). _CONFLUENCE_BONUS = 12 # M30 + H1 beide dafür → Top-Setup (⭐⭐) _H1_AGAINST_PEN = 8 # H1 steht gegen die Richtung (M30 ok/neutral) → schwächer # Regressions-Winkel der Basis-TF als schnellerer Wende-Detektor gegen die # nachlaufende EMA (backtest_angle.py: Winkel dafür/neutral +0,055 vs dagegen # +0,025). NUR Konfidenz/Warnung — als hartes Gate senkt es den Gesamtertrag. _ANGLE_LR = 14 # Regressionsfenster (= ANGLE_LR_BARS) _ANGLE_DEAD = 2.0 # Totband um 90° (Grad) → darunter „neutral" _ANGLE_BONUS = 5 # Winkel bestätigt die EMA-Richtung _ANGLE_PENALTY = 10 # Winkel klar GEGEN die EMA-Richtung (mögliche Wende) # Tageszeit-Gate — REAKTIVIERT 2026-07-13 (User-OK) nach Echtkosten-Messung # (`backtest_realcosts.py`, Kosten = Bar-Spread/ATR statt pauschal 0,1): # 0–7 Uhr = Nacht-Kostenfalle (konstanter Spread ÷ niedriger Nacht-ATR = # 0,32–0,50×ATR Kosten → Edge sicher aufgefressen; in BEIDEN Hälften negativ). # 12 & 16 Uhr = 3× unabhängig negativ gemessen (hourly, hourly_split, realcosts). # Gate-Politik „0–7+12" verbesserte BEIDE Hälften (H1 −4038→−1870, H2 +100→+1191). # Robust positiv sind nur 21–22 Uhr (US-Session). (Zwischenzeitlich war das Gate # auf User-Wunsch ganz aus; Rückholung gezielt & gemessen, kein Pauschal-Gate.) _DEAD_HOURS = (0, 1, 2, 3, 4, 5, 6, 7, 12, 16) _BERLIN = ZoneInfo("Europe/Berlin") # Hohe Vola (oberes ATR-Terzil, ~>0,27 bei WTI-M5) = schwächster/negativer Edge # → nur Konfidenz-Abzug (Schwelle regime-abhängig, daher KEIN hartes Gate). _ATR_HIGH = 0.27 _ATR_HIGH_PEN = 8 # S/R-Kontext (vom Engine gesetzt): Konfidenz dämpfen, wenn der Einstieg direkt # in ein Gegen-Level läuft (wenig Raum / schlechtes CRV), anheben bei Rückenwind. _SR_NEAR_ATR = 0.5 # "dicht an" einer Linie = innerhalb x×ATR _SR_PENALTY = 12 # Konfidenz-Abzug: Trade läuft ins Level (wenig Raum) _SR_BONUS = 8 # Konfidenz-Bonus: Trade startet vom Level mit Rückenwind # Börsen-Sessions (Frankfurt 9:00 / US 15:00): Vorsicht direkt nach Open, # Bonus während aktiver Session, Dämpfung in dünnen Zeiten. _SESSION_CAUTION = 15 # Abzug in den ersten Minuten nach einem Open (Whipsaw) _SESSION_BONUS = 5 # Bonus während aktiver US/DE-Session (Liquidität) _SESSION_OFFHOURS = 8 # Abzug außerhalb DE/US-Session (dünne Liquidität) # TradersUnion ist KEIN harter Blocker mehr (blockte zu oft starke EMA-Trends), # sondern ein Konfidenz-Faktor je TF (5m/15m): bestätigt +Bonus, dagegen −Abzug. _TU_BONUS = 6 # je TF, das die Richtung bestätigt _TU_PENALTY = 8 # je TF, das gegen die Richtung steht (beide = −16) _TF_LABELS = { mt5.TIMEFRAME_M1: "M1", mt5.TIMEFRAME_M5: "M5", mt5.TIMEFRAME_M15: "M15", mt5.TIMEFRAME_M30: "M30", mt5.TIMEFRAME_H1: "H1", } def _atr(highs, lows, closes, period=_ATR_PERIOD): trs = [] for i in range(1, len(highs)): trs.append(max(highs[i] - lows[i], abs(highs[i] - closes[i - 1]), abs(lows[i] - closes[i - 1]))) if not trs: return None return sum(trs[-period:]) / min(len(trs), period) def _ema_last(vals, period): """Letzter EMA-Wert der Reihe (genügt für den Trend-Vergleich).""" if not vals: return None k = 2.0 / (period + 1) e = vals[0] for v in vals[1:]: e = v * k + e * (1.0 - k) return e def _ema_series(vals, period): """Komplette EMA-Reihe (für die Kreuzungs-/Wende-Erkennung je TF).""" k = 2.0 / (period + 1) out = [] e = vals[0] if vals else 0.0 for i, v in enumerate(vals): e = v if i == 0 else v * k + e * (1.0 - k) out.append(e) return out class WaveRecommender: """ATR-ZigZag-Wellen-Empfehlung, bestätigt durch Traders-Union-Daten.""" def __init__(self, tu_provider, timeframe: int = mt5.TIMEFRAME_M5): self.tu = tu_provider self._lock = threading.Lock() self._tf = timeframe self._rec: dict | None = None self._snap: dict = {} self._ts: float = 0.0 self._error: str | None = None self._sr_res: float | None = None # nächster Widerstand über Preis self._sr_sup: float | None = None # nächste Unterstützung unter Preis self._turn_due: float = 0.0 # nächster TF-Ampel-Refresh (Throttle) self._sq_due: float = 0.0 # nächster schneller M5-Squeeze-Fetch (~5 s) self._dead_hours: set = set(_DEAD_HOURS) # Tageszeit-Gate (per Config setzbar) self._last_tf_turns: dict = {} # zuletzt berechnete TF-Wenden self._last_bounce: dict = {"state": None, "dir": None, "tf": None} # Multi-TF-Bounce # ATR-Breakout-Bestätigung: ein Richtungssignal wird erst durchgelassen, # wenn der Kurs k×ATR in Signalrichtung gelaufen ist (gemessen ~2× Edge, # `backtest_breakout.py`). 0 = aus. `_pend` = laufende Bestätigung. self._breakout_k = 1.0 self._breakout_timeout_s = 3600.0 self._pend: dict | None = None self._squeeze: dict | None = None # Volatilitäts-Squeeze-Breakout (M5) # Entry-Raum-Gate (gemessen `backtest_entryroom.py`, monoton in BEIDEN # Hälften): kein Entry, wenn das Gegenlevel < X×ATR entfernt ist — der # Ertrag ist durch den S/R-Auto-Close gedeckelt, Kosten fressen den Rest. # 0 = aus. Nur Live-Pfad (refresh_market); Backtests via _build bleiben frei. self._entry_room_atr = 0.0 def set_entry_room(self, x: float): with self._lock: self._entry_room_atr = max(0.0, float(x)) log.info(f"Entry-Raum-Gate {self._entry_room_atr:.2f}×ATR " f"({'aus' if self._entry_room_atr <= 0 else 'aktiv'})") def set_breakout_k(self, k: float): with self._lock: self._breakout_k = max(0.0, float(k)) self._pend = None log.info(f"Breakout-Bestätigung k={self._breakout_k:.2f}×ATR " f"({'aus' if self._breakout_k <= 0 else 'aktiv'})") def set_dead_hours(self, hours): """Tageszeit-Gate setzen (Set/Liste von Stunden 0–23). Leer/None = AUS. Default (Code) = gemessene Negativ-Stunden; per `[trading] dead_hours` steuerbar.""" try: hs = {int(h) for h in (hours or []) if 0 <= int(h) <= 23} except (TypeError, ValueError): hs = set(_DEAD_HOURS) with self._lock: self._dead_hours = hs log.info(f"Tageszeit-Gate: {sorted(hs) if hs else 'AUS'}") def set_sr_context(self, resistance, support): """Vom Engine: nächste S/R-Linien über/unter dem aktuellen Preis.""" with self._lock: self._sr_res = resistance self._sr_sup = support def set_timeframe(self, tf: int): with self._lock: self._tf = tf self._rec = None # alte Welle verwerfen — TF gewechselt self._ts = 0.0 self._pend = None # Breakout-Bestätigung zurücksetzen log.info(f"Wellen-TF: {_TF_LABELS.get(tf, str(tf))}") # ── TU als Konfidenz-Faktor (kein harter Blocker mehr) ─────────────────── def _tu_check(self, direction: str) -> tuple[bool, int, list[str]]: """Rückgabe (immer True, kein Block; Bonus/Abzug; Gründe). TU bestätigt die EMA-Richtung → Bonus, steht dagegen → Abzug — pro TF (5m/15m).""" snap = self.tu.snapshot() iv = snap.get("intervals") or {} s5 = iv.get("5m", {}).get("score") s15 = iv.get("15m", {}).get("score") if s5 is None or s15 is None: return True, 0, ["TU: keine Daten"] d = 1 if direction == "LONG" else -1 bonus, reasons = 0, [] for lbl, s in (("5m", s5), ("15m", s15)): if s * d >= 1: bonus += _TU_BONUS; reasons.append(f"TU {lbl} dafür") elif s * d <= -1: bonus -= _TU_PENALTY; reasons.append(f"TU {lbl} dagegen") if not reasons: reasons.append("TU neutral") return True, bonus, reasons # ── Marktdaten-Refresh (Hintergrund-Thread, ~5 s) ──────────────────────── def refresh_market(self, sym: str): with self._lock: tf = self._tf with mt5_lock(timeout=2) as got: if not got: return bars = mt5.copy_rates_from_pos(sym, tf, 0, _N_BARS) # Higher-TF-Trend (M30) für den Gegen-Trend-Filter — nur für # niedrigere Basis-TFs (M30/H1 filtern sich sonst selbst). hbars = None if tf not in (_HTF, mt5.TIMEFRAME_H1): hbars = mt5.copy_rates_from_pos(sym, _HTF, 0, _N_BARS) # H1-Trend zusätzlich für die Multi-TF-Konfluenz (nur Konfidenz). h1bars = None if tf != mt5.TIMEFRAME_H1: h1bars = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_H1, 0, _N_BARS) # Wende-Erkennung je TF (M1/M5/M15/M30) für die Welle-Ampel: # EMA12/50-Kreuzung mit Totband. Gedrosselt (~15 s) — die Ampel # braucht keine 5-s-Granularität, spart ~⅓ der Lock-Haltezeit. turn_data: dict = {} if time.time() >= self._turn_due: for _lbl, _t in (("M1", mt5.TIMEFRAME_M1), ("M5", mt5.TIMEFRAME_M5), ("M15", mt5.TIMEFRAME_M15), ("M30", mt5.TIMEFRAME_M30), ("H1", mt5.TIMEFRAME_H1)): # M5 tiefer holen (320): daraus werden die P(break)-Pivot-Level # berechnet (Training: rohe M5-Pivots k=3, Lookback 300 Bars). _n = 320 if _lbl == "M5" else 120 _tb = mt5.copy_rates_from_pos(sym, _t, 0, _n) if _tb is not None and len(_tb) > _ATR_PERIOD: _c = [float(b["close"]) for b in _tb] _h = [float(b["high"]) for b in _tb] _l = [float(b["low"]) for b in _tb] turn_data[_lbl] = (_c, _atr(_h, _l, _c)) if _lbl == "M5": self._m5_hl = (_h, _l) self._turn_due = time.time() + 15 # ── Schneller M5-Squeeze-Fetch (Variante A): entkoppelt von der 15-s- # Ampel, kleiner 60-Bar-M5-Fetch ~alle 5 s → Ausbrüche werden ~3× schneller # erkannt (gleiche validierte M5-Regel). Squeeze wird danach neu berechnet. sq_m5 = None if time.time() >= self._sq_due: _sb = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, 60) if _sb is not None and len(_sb) > _ATR_PERIOD: sq_m5 = ([float(b["high"]) for b in _sb], [float(b["low"]) for b in _sb], [float(b["close"]) for b in _sb]) self._sq_due = time.time() + _SQ_REFRESH_S if bars is None or len(bars) < _ATR_PERIOD + 5: with self._lock: self._error = "keine Bars" 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 = _atr(highs, lows, closes) if not atr or atr <= 0: with self._lock: self._error = "ATR=0" return cur = closes[-1] tf_lbl = _TF_LABELS.get(tf, str(tf)) ef = _ema_last(closes, _EMA_FAST) es = _ema_last(closes, _EMA_SLOW) with self._lock: sr_res, sr_sup = self._sr_res, self._sr_sup htf_trend = self._htf_sign(hbars) h1_trend = self._htf_sign(h1bars) angle = calc_trend_angle(closes[-(_ANGLE_LR + 2):], _ANGLE_LR) hour = datetime.now(_BERLIN).hour rec, snap = self._build(ef, es, cur, atr, tf_lbl, 0, sr_res, sr_sup, htf_trend=htf_trend, h1_trend=h1_trend, angle=angle, hour=hour) # Die tatsächlich genutzten Higher-TF-Signale für die Gesamtempfehlung # exponieren (konsistent mit der Empfehlung, nicht mit dem Winkel-Maß). snap["htf_trend"] = htf_trend snap["h1_trend"] = h1_trend # Squeeze aus dem schnellen (~5 s) M5-Fetch neu berechnen — entkoppelt von der # 15-s-Ampel (Variante A). Gleiche validierte M5-Regel, nur öfter geprüft. if sq_m5: _h5s, _l5s, _c5s = sq_m5 _a5s = _atr(_h5s, _l5s, _c5s) if _a5s and _a5s > 0: self._squeeze = self._squeeze_one(_h5s, _l5s, _c5s, max(_a5s, 0.12)) if turn_data: # frisch geholt → neu berechnen, sonst letzten Stand halten self._last_tf_turns = {lbl: self._turn_state(c, a) for lbl, (c, a) in turn_data.items()} # P(break)-Features IMMER aus M5 (Modell wurde auf M5 trainiert/kalibriert # — die Wave-TF wechselt per Heuristik bis M30; damit wäre mom6 ein 3-h- # statt 30-min-Momentum und der ATR 2–3× größer → P(break) völlig falsch # skaliert. Bug gefixt 2026-07-13.) ATR-Floor 0,12 wie im Training. if "M5" in turn_data: _c5, _a5 = turn_data["M5"] if _a5 and _a5 > 0 and len(_c5) >= _EMA_SLOW + 2: _a5f = max(_a5, 0.12) self._pb_feats = { "mom6": round((_c5[-1] - _c5[-7]) / _a5f, 3), "mom3": round((_c5[-1] - _c5[-4]) / _a5f, 3), "ema_diff": round((_ema_last(_c5, _EMA_FAST) - _ema_last(_c5, _EMA_SLOW)) / _a5f, 3), "atr": round(_a5f, 4), } # Pivot-Level trainingsgleich: rohe M5-Pivots (k=3) über die # letzten ~300 Bars — NICHT die geclusterten M15-Level (die sind # in frischen Trends oft LEER → Hint verschwand; Bugfix). _h5, _l5 = getattr(self, "_m5_hl", (None, None)) if _h5 and len(_h5) >= 10: _K = 3 ph = [round(_h5[j], 3) for j in range(_K, len(_h5) - _K) if _h5[j] == max(_h5[j - _K:j + _K + 1])] pl = [round(_l5[j], 3) for j in range(_K, len(_l5) - _K) if _l5[j] == min(_l5[j - _K:j + _K + 1])] self._pb_levels = {"ph": ph, "pl": pl} # (Squeeze läuft jetzt auf dem schnellen ~5-s-M5-Fetch oben, # nicht mehr hier auf dem 15-s-Ampel-Pfad — Variante A 2026-07-21.) # Multi-TF-Bounce: stärksten Status über M1/M5/M15/M30 wählen (H1 ist NUR # für die Ampel im turn_data, NICHT im gemessen-neutralen Bounce → skip). cands = [] for lbl, (c, a) in turn_data.items(): if lbl == "H1": continue b = self._bounce_one(c, a) if b: b["tf"] = lbl; cands.append(b) if cands: rank = {"active": 2, "expected": 1} best = max(cands, key=lambda b: (rank[b["state"]], b["stretch"])) self._last_bounce = {"state": best["state"], "dir": best["dir"], "tf": best["tf"]} else: self._last_bounce = {"state": None, "dir": None, "tf": None} snap["tf_turns"] = dict(self._last_tf_turns) snap["bounce"] = dict(self._last_bounce) # Multi-TF überschreibt Basis-TF snap["pb_feats"] = dict(self._pb_feats) if getattr(self, "_pb_feats", None) else None snap["pb_levels"] = dict(self._pb_levels) if getattr(self, "_pb_levels", None) else None snap["squeeze"] = dict(self._squeeze) if getattr(self, "_squeeze", None) else \ {"state": None, "dir": None, "level": None, "box_atr": None} # ── Entry-Raum-Gate (gemessen `backtest_entryroom.py`, beide Hälften # monoton): Richtungssignal → WARTEN, wenn das GEGENLEVEL (M5-Pivot in # Trade-Richtung, trainingsgleich zum S/R-Auto-Close) näher als # `_entry_room_atr`×ATR_M5 liegt — der Ertrag ist dort durch den Auto- # Close gedeckelt (63 % Containment), die Echtkosten fressen den Rest. # Verhindert die Sofort-/Klein-Close-Trades an der QUELLE (Exit bleibt). rec = self._room_gate(rec, cur) rec = self._confirm_breakout(rec, cur, atr, snap) # k×ATR-Bestätigung with self._lock: self._rec = rec self._snap = snap self._ts = time.time() self._error = None @staticmethod def _htf_sign(hbars) -> int: """Vorzeichen des Higher-TF-Trends (M30, EMA12 vs EMA50): +1 auf, −1 ab, 0 neutral/keine Daten (Totband _HTF_DEADBAND×ATR). Fail-open: ohne Bars → 0 → kein Filter.""" if hbars is None or len(hbars) < _EMA_SLOW + 5: return 0 h = [float(b["high"]) for b in hbars] l = [float(b["low"]) for b in hbars] c = [float(b["close"]) for b in hbars] atr = _atr(h, l, c) if not atr or atr <= 0: return 0 ef = _ema_last(c, _EMA_FAST) es = _ema_last(c, _EMA_SLOW) if ef is None or es is None: return 0 d = ef - es if abs(d) < _HTF_DEADBAND * atr: return 0 return 1 if d > 0 else -1 @staticmethod def _turn_state(closes, atr: float | None = None) -> dict: """Wende-Status einer TF aus der EMA12/50-Kreuzung — mit Totband, damit bei seitwärts laufenden EMAs nicht jede Mikro-Kreuzung als „Wende" blinkt. dir: +1 auf / −1 ab / 0 unklar (im Totband) · bars_ago: Bars seit letzter Kreuzung (None = keine im Fenster) · fresh: echte Kreuzung ≤3 Bars her.""" if not closes or len(closes) < _EMA_SLOW + 5: return {"dir": 0, "bars_ago": None, "fresh": False} ef = _ema_series(closes, _EMA_FAST) es = _ema_series(closes, _EMA_SLOW) diff = [a - b for a, b in zip(ef, es)] dead = (_TREND_DEADBAND * atr) if atr else 0.0 # Substanz-Schwelle last = diff[-1] cur = 1 if last > dead else -1 if last < -dead else 0 bars_ago = None for i in range(len(diff) - 1, _EMA_SLOW, -1): # Warmup-Bereich überspringen if diff[i] != 0 and (diff[i] > 0) != (diff[i - 1] > 0): bars_ago = (len(diff) - 1) - i # Kreuzung bei Bar i break return {"dir": cur, "bars_ago": bars_ago, "fresh": bars_ago is not None and bars_ago <= 3 and cur != 0} def _confirm_breakout(self, rec, cur, atr, snap): """ATR-Breakout-Bestätigung (stateful): ein Richtungssignal wird erst durchgelassen, wenn der Kurs **k×ATR in Signalrichtung** gelaufen ist (= „X dynamisch"). Läuft er vorher k×ATR DAGEGEN oder Timeout → neu verankern (WARTEN). Gemessen ~2× Edge/Trade (`backtest_breakout.py`). k=0 → Filter aus (Sofort-Einstieg).""" k = self._breakout_k sig = rec.get("signal") snap["breakout"] = {"pending": False, "dir": None, "need": None} if not k or k <= 0 or atr <= 0 or sig == "WARTEN": if sig == "WARTEN": self._pend = None return rec d = 1 if sig == "LONG" else -1 p = self._pend if p is None or p.get("dir") != d: # frisches Signal → verankern p = {"dir": d, "level": cur + d * k * atr, "invalid": cur - d * k * atr, "confirmed": False, "start": time.time()} self._pend = p if p["confirmed"]: return rec # schon bestätigt → durchlassen broke = (cur >= p["level"]) if d > 0 else (cur <= p["level"]) against = (cur <= p["invalid"]) if d > 0 else (cur >= p["invalid"]) if broke: p["confirmed"] = True return rec if against or (time.time() - p["start"]) > self._breakout_timeout_s: p = {"dir": d, "level": cur + d * k * atr, "invalid": cur - d * k * atr, "confirmed": False, "start": time.time()} # neu verankern self._pend = p need = abs(p["level"] - cur) snap["breakout"] = {"pending": True, "dir": sig, "level": round(p["level"], 3), "need": round(need, 3)} return {"signal": "WARTEN", "conf_pct": 0, "score": 0.0, "setup": "WAVE", "regime": None, "rsi": None, "reasons": [f"warte auf Breakout (+{k:.1f}×ATR {sig}, noch {need:.2f})"]} def _room_gate(self, rec, cur): """Entry-Raum-Gate: Signal → WARTEN, wenn das Gegenlevel < X×ATR_M5 entfernt ist (gemessen `backtest_entryroom.py`: Raum <0,6×ATR in BEIDEN Hälften klar negativ — 79–82 % WR, aber PF<1 = Klein-Close-Falle). Level/ATR trainings- gleich aus `_pb_levels`/`_pb_feats` (M5). Fail-open: ohne Daten kein Gate.""" x = self._entry_room_atr sig = rec.get("signal") if x <= 0 or sig == "WARTEN": return rec lv = getattr(self, "_pb_levels", None) or {} pf = getattr(self, "_pb_feats", None) or {} atr5 = pf.get("atr") if not atr5: return rec d = 1 if sig == "LONG" else -1 if d > 0: cands = [p for p in (lv.get("ph") or []) if p > cur] lvl = min(cands) if cands else None else: cands = [p for p in (lv.get("pl") or []) if p < cur] lvl = max(cands) if cands else None if lvl is None: # freie Bahn (bestes gemessenes Segment) return rec dist = (lvl - cur) * d / atr5 if dist >= x: return rec return {"signal": "WARTEN", "conf_pct": 0, "score": 0.0, "setup": "WAVE", "regime": None, "rsi": None, "reasons": [f"kein Raum: {'Widerstand' if d > 0 else 'Support'} " f"{lvl:.2f} nur {dist:.2f}×ATR entfernt (Gate {x:.1f}) — " f"Ertrag gedeckelt, Kosten fressen den Edge"]} @staticmethod def _squeeze_one(highs, lows, closes, atr): """Volatilitäts-Squeeze-Breakout (M5, gemessen `backtest_breakout_squeeze.py`). Box = Spanne der letzten _SQ_N ABGESCHLOSSENEN Bars. Ist sie ≤ _SQ_MULT×ATR (Kompression): - Kurs bricht _SQ_K×ATR über/unter die Box → `active` (LONG/SHORT), level = Ausbruchsgrenze — das getestete Einstiegssignal. - sonst → `armed` (komprimiert, Ausbruch steht bevor; dir = nähere Grenze). Nicht komprimiert → state None. Rein transient (der Ausbruch weitet die Box → Signal klärt sich von selbst, sobald der Move läuft).""" none = {"state": None, "dir": None, "level": None, "box_atr": None} if not atr or atr <= 0 or len(closes) < _SQ_N + 2: return none hb = highs[-_SQ_N - 1:-1]; lb = lows[-_SQ_N - 1:-1] # N abgeschlossene Bars if not hb or not lb: return none box_hi = max(hb); box_lo = min(lb) box_atr = round((box_hi - box_lo) / atr, 2) if box_atr > _SQ_MULT: # keine Kompression return {"state": None, "dir": None, "level": None, "box_atr": box_atr} px = closes[-1] up = box_hi + _SQ_K * atr; dn = box_lo - _SQ_K * atr if px >= up: return {"state": "active", "dir": "LONG", "level": round(up, 3), "box_atr": box_atr} if px <= dn: return {"state": "active", "dir": "SHORT", "level": round(dn, 3), "box_atr": box_atr} dir_ = "LONG" if (up - px) <= (px - dn) else "SHORT" # armed → nähere Grenze return {"state": "armed", "dir": dir_, "level": round(up if dir_ == "LONG" else dn, 3), "box_atr": box_atr} @staticmethod def _bounce_one(closes, atr): """Bounce-Status EINER TF: überdehnt (|stretch|≥_REVERSAL_STRETCH von EMA50) → expected; + Winkel gedreht → active. dir=LONG bei überverkauft (unter EMA), SHORT bei überkauft. None, wenn nicht überdehnt.""" if not closes or len(closes) < _EMA_SLOW + 5 or not atr or atr <= 0: return None es = _ema_series(closes, _EMA_SLOW)[-1] stretch = (closes[-1] - es) / atr if abs(stretch) < _REVERSAL_STRETCH: return None ad = calc_trend_angle(closes[-(_ANGLE_LR + 2):], _ANGLE_LR) - 90.0 turned = (stretch < 0 and ad >= _ANGLE_DEAD) or (stretch > 0 and ad <= -_ANGLE_DEAD) return {"state": "active" if turned else "expected", "dir": "LONG" if stretch < 0 else "SHORT", "stretch": abs(stretch)} def _build(self, ef, es, cur, atr, tf_lbl, n_pivots, sr_res=None, sr_sup=None, htf_trend=0, h1_trend=0, angle=90.0, hour=None): """Signalrichtung aus dem EMA12/50-Trend; Einstieg nur, wenn der Trend klar ist (Totband) und der Kurs nicht überdehnt von der EMA weg ist.""" diff = (ef - es) if (ef is not None and es is not None) else 0.0 sep = diff / atr if atr else 0.0 # Trendstärke in ATR stretch = (cur - es) / atr if (es is not None and atr) else 0.0 snap = {"tf": tf_lbl, "atr": atr, "direction": ("up" if diff > 0 else "down" if diff < 0 else None), "wave_start": round(es, 3) if es is not None else None, "move_atr": round(stretch, 2), "n_pivots": n_pivots, "trend_sep": round(sep, 2)} # ── Bounce-Status (für die Anzeige), unabhängig vom Signal-Flow: # "expected" = überdehnt (≥_REVERSAL_STRETCH), Winkel noch NICHT gedreht # "active" = überdehnt UND Winkel gedreht (= der Reversal-Trigger feuert) # dir = LONG bei überverkauft (unter EMA), SHORT bei überkauft. _ad = angle - 90.0 if abs(stretch) >= _REVERSAL_STRETCH: _turned = (stretch < 0 and _ad >= _ANGLE_DEAD) or (stretch > 0 and _ad <= -_ANGLE_DEAD) snap["bounce"] = {"state": "active" if _turned else "expected", "dir": "LONG" if stretch < 0 else "SHORT"} else: snap["bounce"] = {"state": None, "dir": None} wait = {"signal": "WARTEN", "conf_pct": 0, "score": 0.0, "setup": "WAVE", "regime": None, "rsi": None, "reasons": []} # Totband: kein klarer Trend → kein Trade (Chop) if abs(sep) < _TREND_DEADBAND: wait["reasons"] = [f"kein klarer Trend ({tf_lbl}, EMA-Abstand {sep:+.2f}×ATR)"] return wait, snap # Tageszeit-Gate (per Config `set_dead_hours`; Default _DEAD_HOURS = Nacht- # Kostenfalle + 12/16 Uhr. Leer = AUS, User-Vorgabe 2026-07-22 trotz Messung). if hour is not None and hour in self._dead_hours: why = "Nacht-Spread frisst den Edge" if hour <= 7 else "gemessen negativer Edge" wait["reasons"] = [f"Zeit-Gate {hour}:00 Uhr — {why}, kein Trade"] return wait, snap # EIA-Blackout (gemessen, `backtest_events.py`): Mittwoch 15:30–16:30 Berlin # = Vorlauf der EIA-Lagerdaten (16:30). In BEIDEN History-Hälften netto # negativ (−0,147/−0,088 vs rest) — Positionierungs-Chop vor den Zahlen. # Kausales Event-Fenster, kein Pauschal-Gate. Nur im Live-Pfad (hour≠None). if hour is not None: _now_b = datetime.now(_BERLIN) if _now_b.weekday() == 2 and (15, 30) <= (_now_b.hour, _now_b.minute) < (16, 30): wait["reasons"] = ["EIA-Blackout Mi 15:30–16:30 — Lagerdaten-Vorlauf, " "gemessen negativ, kein Trade"] return wait, snap sig = "LONG" if diff > 0 else "SHORT" reversal = False # ── Reversal/BOUNCE (antizyklisch, markiertes ZWEITsignal): überdehnt # (|stretch|≥_REVERSAL_STRETCH=3,0) + Regressions-Winkel hat GEDREHT → # Einstieg in Winkel-Richtung, gegen die EMA. Exit-Sim (backtest_bounce.py): # Ø-R +0,185, PF 1,35, Treffer 70 %, Worst −2×ATR (SL-gedeckelt) — profitabel, # aber schwächer als Trend & regime-anfällig (blutet in starken Trends). # Hebt Anti-Überdehnung UND M30-Filter bewusst auf (per Definition gegen # die nachlaufende EMA/HTF). ad = angle - 90.0 if stretch <= -_REVERSAL_STRETCH and ad >= _ANGLE_DEAD: sig, reversal = "LONG", True # überverkauft + Winkel auf → Bounce elif stretch >= _REVERSAL_STRETCH and ad <= -_ANGLE_DEAD: sig, reversal = "SHORT", True # überkauft + Winkel ab → Bounce d_sig = 1 if sig == "LONG" else -1 if not reversal: # Higher-TF-Gegen-Trend-Filter: kein Short im M30-Aufwärtstrend (und # umgekehrt). Per Backtest belegt (Ø-Edge ×2). Nur reguläre Trendsignale. if htf_trend != 0 and htf_trend != d_sig: wait["reasons"] = [ f"gegen {_HTF_LABEL}-Trend " f"({'auf' if htf_trend > 0 else 'ab'}) — kein Gegen-Trade"] return wait, snap # Anti-Überdehnung: nicht weit weg von der EMA hinterherkaufen/-shorten if sig == "LONG" and stretch > _STRETCH_MAX: wait["reasons"] = [f"überdehnt: {stretch:+.1f}×ATR über EMA — kein Spät-Long"] return wait, snap if sig == "SHORT" and stretch < -_STRETCH_MAX: wait["reasons"] = [f"überdehnt: {stretch:+.1f}×ATR unter EMA — kein Spät-Short"] return wait, snap if reversal: reasons = [f"🔄 Reversal {sig}: {abs(stretch):.1f}×ATR überdehnt + Winkel gedreht"] else: reasons = [f"{tf_lbl}-Trend {'auf' if sig == 'LONG' else 'ab'} " f"(EMA-Abstand {sep:+.2f}×ATR)"] conf = _BASE_CONF # Trendstärke if abs(sep) >= 0.5: conf += 10; reasons.append("starker Trend") elif abs(sep) >= 0.25: conf += 5 # Einstiegsqualität nach gemessenem Edge (backtest_pullback.py): ein # TIEFER Pullback (Kurs durch die EMA zurück, near<0) trägt mit Abstand # am besten; die laue Zone direkt an der EMA (0–0,3) ist der schwächste # Edge; weit gelaufenes Trend-Momentum (1–2×ATR) trägt wieder ordentlich. # Beeinflusst nur Konfidenz/Score (Anzeige + Auto-Dry-Run), NICHT die # Signalrichtung. near = stretch if sig == "LONG" else -stretch if near < 0: conf += 15; reasons.append("⭐ tiefer Pullback (bestes CRV)") elif near < 0.3: conf -= 3; reasons.append("laue Zone an EMA (schwächster Edge)") elif near < 1.0: conf += 3 elif near < 2.0: conf += 8; reasons.append("Trend-Momentum trägt") else: conf += 5; reasons.append("weit gelaufen — Vorsicht Überdehnung") # ── Multi-TF-Konfluenz (nur Konfidenz): M30 UND H1 dafür = Top-Setup; # H1 dagegen = schwächer. M30-Gegen-Trend ist oben schon WARTEN, hier # ist htf_trend also nur =Richtung oder neutral. (Backtest: Edge ×3,5.) if htf_trend == d_sig and h1_trend == d_sig: conf += _CONFLUENCE_BONUS reasons.append("⭐⭐ Konfluenz M30+H1") elif h1_trend != 0 and h1_trend != d_sig: conf -= _H1_AGAINST_PEN reasons.append("H1 gegen Richtung — schwächeres Setup") # ── Regressions-Winkel vs nachlaufende EMA (nur Konfidenz/Warnung) ─── # Steht der Winkel der Basis-TF klar GEGEN die EMA-Richtung, ist die EMA # evtl. am Nachlaufen (Wende) → Warnung. Backtest: Winkel-dafür trägt # klar besser; als Gate aber Gesamtertrag-negativ → nur Konfidenz. # (ad = angle − 90 wurde oben in der Reversal-Prüfung bereits berechnet.) if (d_sig > 0 and ad < -_ANGLE_DEAD) or (d_sig < 0 and ad > _ANGLE_DEAD): conf -= _ANGLE_PENALTY reasons.append("⚠ Winkel gegen EMA (mögliche Wende)") elif (d_sig > 0 and ad > _ANGLE_DEAD) or (d_sig < 0 and ad < -_ANGLE_DEAD): conf += _ANGLE_BONUS reasons.append("Winkel bestätigt") # Hohe Vola dämpfen (oberes ATR-Terzil = schwächster/negativer Edge) — # nur Konfidenz (Schwelle regime-abhängig, kein hartes Gate). if atr and atr >= _ATR_HIGH: conf -= _ATR_HIGH_PEN reasons.append("hohe Vola — schwächster Edge") # ── S/R-Kontext: Raum bis zur nächsten Linie in Trade-Richtung ─────── if atr: thr = _SR_NEAR_ATR * atr if sig == "LONG": if sr_res is not None and 0 <= (sr_res - cur) < thr: conf -= _SR_PENALTY reasons.append(f"dicht unter Widerstand {sr_res:.3f} (wenig Raum)") if sr_sup is not None and 0 <= (cur - sr_sup) < thr: conf += _SR_BONUS reasons.append(f"an Unterstützung {sr_sup:.3f} (Rückenwind)") else: # SHORT if sr_sup is not None and 0 <= (cur - sr_sup) < thr: conf -= _SR_PENALTY reasons.append(f"dicht über Unterstützung {sr_sup:.3f} (wenig Raum)") if sr_res is not None and 0 <= (sr_res - cur) < thr: conf += _SR_BONUS reasons.append(f"an Widerstand {sr_res:.3f} (Rückenwind)") # ── Börsen-Session: Vorsicht nach Open, Bonus in aktiver Session ───── ss = session_state() if ss["just_opened"]: conf -= _SESSION_CAUTION reasons.append(f"{ss['just_opened']}-Open frisch — volatil, Vorsicht") elif ss["active"]: conf += _SESSION_BONUS reasons.append(f"{'+'.join(ss['active'])}-Session aktiv") elif not ss["weekend"]: conf -= _SESSION_OFFHOURS reasons.append("außerhalb DE/US-Session (dünn)") # TU aus der Empfehlung ENTFERNT (User-Vorgabe 2026-07-06): Standard-Indikator- # Konfluenz = kein Edge (gemessen `backtest_confluence.py`); TU lagt (5m „Strong # Buy" während 4h/1d „Strong Sell") + ist nicht backtestbar (Live-Scrape, keine # History). `_tu_check`/`_TU_*` bleiben als Code, fließen aber NICHT mehr in die # Empfehlungs-Konfidenz. TU ist nur noch reine Anzeige (Snapshot). conf = max(0, min(conf, 90)) # Mindest-Konfidenz-Gate: schwache Setups (zu viele Strafen gestapelt) # tragen negativen Edge (gemessen Band 40–54 %) → kein Trade. if conf < _MIN_CONF: wait["reasons"] = [f"Konfidenz {conf}% < {_MIN_CONF}% — Setup zu schwach"] + reasons[:2] return wait, snap if reversal: setup = "WAVE_REV_LONG" if sig == "LONG" else "WAVE_REV_SHORT" else: setup = "WAVE_LONG" if sig == "LONG" else "WAVE_SHORT" rec = {"signal": sig, "conf_pct": conf, "score": 0.8 if sig == "LONG" else -0.8, "setup": setup, "regime": None, "rsi": None, "reasons": reasons} return rec, snap # ── Signal für Panel + Auto-Trader ─────────────────────────────────────── def signal(self) -> dict: with self._lock: rec, ts, err = self._rec, self._ts, self._error if rec is None: return {"signal": "WARTEN", "conf_pct": 0, "score": 0.0, "setup": "WAVE", "regime": None, "rsi": None, "reasons": [err or "keine Wellen-Daten"]} if not ts or time.time() - ts > _STALE_S: return {"signal": "WARTEN", "conf_pct": 0, "score": 0.0, "setup": "WAVE", "regime": None, "rsi": None, "reasons": ["Wellen-Daten veraltet"]} return dict(rec) def snapshot(self) -> dict: with self._lock: d = dict(self._snap) d["error"] = self._error d["last_update"] = self._ts return d