""" core/patterns.py — Visuelle Chartmuster-Erkennung (ANZEIGE, kein Signal) ======================================================================== Erkennt mechanisch definierte Chartmuster aus den M30-Swing-Pivots und liefert sie fürs Dashboard: Typ, Richtung, Nackenlinie/Trigger, Measured-Move-Ziel, Status (bildet sich / bestätigt / ungültig) und eine Qualitäts-Note. Erkannte Muster (v1): • Doppeltop / Doppelboden (2 gleiche Extrema + Nackenlinie) • Kopf-Schulter / invers (SKS) (3 Extrema, Mitte am höchsten/tiefsten) • Tasse+Henkel / invers (heuristisch: abgerundetes Extrem + Henkel) ⚠ REINE ANZEIGE — KEIN Trade-Trigger, KEIN Verdict-Gewicht. Signal-Einfluss ERST nach Backtest (`backtest_patterns.py`, noch zu bauen): dieselbe Klasse (Doppeltop, Struktur) ist im Projekt schon 2× als SIGNAL verworfen (`backtest_doubletop.py`, `backtest_structure.py`) → hohe Beweislast. Das Modul zeigt nur den Ist-Zustand + das (mechanische) Measured-Move-Ziel, damit man sieht, was ein Kommentator meint. Thread-sicher wie structure.py: refresh_market(sym) holt M30-Bars unter mt5_lock (~30 s gedrosselt), snapshot() liefert den letzten Stand ohne MT5-Call. """ from __future__ import annotations import threading import time import MetaTrader5 as mt5 from core.mt5_utils import mt5_lock from core.logger import get_logger log = get_logger("patterns") _TF = mt5.TIMEFRAME_M30 _N_BARS = 240 _PIVOT_K = 3 _REFRESH_S = 30.0 _MIN_HEIGHT_ATR = 1.2 # Muster-Höhe muss ≥ so viel ATR sein (sonst Rauschen) _TOL_ATR = 0.7 # zwei "gleiche" Extrema dürfen so weit auseinander (×ATR) _NECK_TOL_ATR = 0.9 # Nackenlinie "flach": zwei Tröge/Hochs so weit (×ATR) _MAX_LOOKBACK = 9 # nur die letzten N Swings betrachten (aktuelle Muster) _MAX_DIST_ATR = 5.0 # Trigger max. so weit vom AKTUELLEN Kurs (sonst nicht mehr # aktionabel — alte, weit weg gelaufene Muster ausblenden) # Typen, die im Kontrolltest (`backtest_patterns.py` gegen einen generischen # Swing-Bruch, 2026-07-30) tatsaechlich gemessen wurden. NUR sie tragen die # Verdict-Stimme; die 2026-07-31 aus dem Handbuch ergaenzten Typen (Flagge, # Wimpel, Rechteck, Keil, Dreifach) sind REINE ANZEIGE, bis sie gemessen sind. _MEASURED = {"double_top", "double_bottom", "hs", "inv_hs", "cup", "inv_cup", "tri_asc", "tri_desc"} # Fortsetzungsmuster brauchen eine „Stange" (impulsiver Vorlauf) — ab wann gilt # ein Swing als Stange, und wie kurz muss die Konsolidierung dagegen sein. _POLE_MIN_ATR = 2.0 # Vorlauf mindestens so gross _POLE_MAX_BARS = 20 # ... und in hoechstens so vielen Bars (impulsiv) _FLAG_MAX_FRAC = 0.5 # Konsolidierung hoechstens so gross wie die Stange def _atr(highs, lows, closes, p=14): trs = [] for i in range(1, len(closes)): trs.append(max(highs[i]-lows[i], abs(highs[i]-closes[i-1]), abs(lows[i]-closes[i-1]))) return (sum(trs[-p:]) / min(len(trs), p)) if trs else None def _pivots(highs, lows, k): """Alternierende Swing-Punkte → [(index, price, 'H'/'L')] (wie structure.py).""" n = len(highs); raw = [] for i in range(k, n - k): if highs[i] == max(highs[i-k:i+k+1]) and highs[i] > highs[i-1] and highs[i] >= highs[i+1]: raw.append((i, highs[i], "H")) elif lows[i] == min(lows[i-k:i+k+1]) and lows[i] < lows[i-1] and lows[i] <= lows[i+1]: raw.append((i, lows[i], "L")) out = [] for p in raw: if out and out[-1][2] == p[2]: if (p[2] == "H" and p[1] > out[-1][1]) or (p[2] == "L" and p[1] < out[-1][1]): out[-1] = p else: out.append(p) return out def _sim(a, b, atr): """Ähnlichkeit zweier Preise (1 = identisch, 0 = ≥ _TOL_ATR entfernt).""" return max(0.0, 1.0 - abs(a - b) / (_TOL_ATR * atr)) if atr else 0.0 def _status(dir_, trigger, cur, extreme): """bestätigt (Trigger gebrochen) / ungültig (Extrem überschritten) / bildet sich.""" if dir_ == "bear": if cur < trigger: return "confirmed" if cur > extreme: return "invalidated" else: if cur > trigger: return "confirmed" if cur < extreme: return "invalidated" return "forming" def _detect(piv, cur, atr, last_idx): """Sucht Muster in den letzten Swings. → Liste Muster-dicts.""" out = [] P = piv[-_MAX_LOOKBACK:] if len(P) < 3: return out pr = [p[1] for p in P]; kd = [p[2] for p in P]; ix = [p[0] for p in P] def add(typ, name, dir_, quality, trigger, extreme, note): h = abs(extreme - trigger) if h < _MIN_HEIGHT_ATR * atr: # zu flach = Rauschen return target = trigger - h if dir_ == "bear" else trigger + h out.append({ "type": typ, "name": name, "dir": dir_, "quality": round(quality, 2), "trigger": round(trigger, 3), "target": round(target, 3), "height_atr": round(h / atr, 1), "status": _status(dir_, trigger, cur, extreme), "bars_ago": last_idx - ix[-1], "note": note, # ⚠ `measured` = war dieser Typ im Kontrolltest (`backtest_patterns.py` # + generischer Breakout, 2026-07-30)? NUR gemessene Typen dürfen die # Verdict-Stimme tragen — sonst würde ein validiertes Gewicht still mit # unvalidierten Eingaben gefüttert (der Deployment-Drift-Fehler des # Projekts). Gemessen wurden: Doppeltop/-boden, Tasse/inv. Tasse, # auf-/absteigendes Dreieck, SKS/inv. SKS. "measured": typ in _MEASURED, }) # über alle 3er/5er-Fenster am aktuellen Rand for s in range(len(P) - 2): w = list(zip(kd[s:], pr[s:], ix[s:])) # ── Doppeltop: H, L, H (zwei ~gleiche Hochs, Trog dazwischen) ── if len(w) >= 3 and w[0][0] == "H" and w[1][0] == "L" and w[2][0] == "H": Ha, Lm, Hb = w[0][1], w[1][1], w[2][1] q = _sim(Ha, Hb, atr) if q > 0.35 and Lm < min(Ha, Hb): add("double_top", "Doppeltop", "bear", q, Lm, max(Ha, Hb), f"zwei Hochs {Ha:.2f}/{Hb:.2f}, Nackenlinie {Lm:.2f}") # ── Doppelboden: L, H, L ── if len(w) >= 3 and w[0][0] == "L" and w[1][0] == "H" and w[2][0] == "L": La, Hm, Lb = w[0][1], w[1][1], w[2][1] q = _sim(La, Lb, atr) if q > 0.35 and Hm > max(La, Lb): add("double_bottom", "Doppelboden", "bull", q, Hm, min(La, Lb), f"zwei Tiefs {La:.2f}/{Lb:.2f}, Nackenlinie {Hm:.2f}") # ── Kopf-Schulter (SKS): H, L, H, L, H (Kopf = mittleres Hoch am höchsten) ── if len(w) >= 5 and [x[0] for x in w[:5]] == ["H", "L", "H", "L", "H"]: ls, t1, head, t2, rs = (w[i][1] for i in range(5)) if head > ls and head > rs: neck = (t1 + t2) / 2.0 q = _sim(ls, rs, atr) * _sim(t1, t2, atr) if q > 0.25 and abs(t1 - t2) < _NECK_TOL_ATR * atr: add("hs", "Kopf-Schulter", "bear", q, neck, head, f"Kopf {head:.2f}, Schultern {ls:.2f}/{rs:.2f}, Nacken {neck:.2f}") # ── Inverse SKS: L, H, L, H, L ── if len(w) >= 5 and [x[0] for x in w[:5]] == ["L", "H", "L", "H", "L"]: ls, t1, head, t2, rs = (w[i][1] for i in range(5)) if head < ls and head < rs: neck = (t1 + t2) / 2.0 q = _sim(ls, rs, atr) * _sim(t1, t2, atr) if q > 0.25 and abs(t1 - t2) < _NECK_TOL_ATR * atr: add("inv_hs", "Inverse SKS", "bull", q, neck, head, f"Kopf {head:.2f}, Schultern {ls:.2f}/{rs:.2f}, Nacken {neck:.2f}") # ── (Inverse) Tasse+Henkel (heuristisch): Rand-Extrema ~gleich, Boden/Top # dazwischen, breit; Henkel = letzter kleiner Gegen-Swing. ── if len(w) >= 3: span = w[2][2] - w[0][2] # Bars zwischen den Rand-Punkten if w[0][0] == "L" and w[1][0] == "H" and w[2][0] == "L" and span >= 10: rim_l, top, rim_r = w[0][1], w[1][1], w[2][1] # inverse Tasse (abger. Top) q = _sim(rim_l, rim_r, atr) * 0.9 if q > 0.4 and top > max(rim_l, rim_r): add("inv_cup", "Inverse Tasse+Henkel", "bear", q, min(rim_l, rim_r), top, f"abgerundetes Top {top:.2f}, Rand ~{rim_r:.2f} (Henkel-Unterkante)") if w[0][0] == "H" and w[1][0] == "L" and w[2][0] == "H" and span >= 10: rim_l, bot, rim_r = w[0][1], w[1][1], w[2][1] # Tasse (abger. Boden) q = _sim(rim_l, rim_r, atr) * 0.9 if q > 0.4 and bot < min(rim_l, rim_r): add("cup", "Tasse+Henkel", "bull", q, max(rim_l, rim_r), bot, f"abgerundeter Boden {bot:.2f}, Rand ~{rim_r:.2f} (Henkel-Oberkante)") # ══ HANDBUCH-MUSTER (ergänzt 2026-07-31, User-Vorgabe „alle Muster aus dem # Handbuch") — ALLE REINE ANZEIGE (`measured=False`), sie tragen die # Verdict-Stimme NICHT, weil der Kontrolltest sie nicht abdeckt. ═══════════ for s2 in range(len(P) - 2): w = list(zip(kd[s2:], pr[s2:], ix[s2:])) # ── Dreifach-Top / -Boden: drei ~gleiche Extreme (H,L,H,L,H bzw. umgekehrt). # Abgrenzung zur SKS: dort ist das MITTLERE Extrem deutlich weiter draussen. if len(w) >= 5 and [x[0] for x in w[:5]] == ["H", "L", "H", "L", "H"]: h1, t1, h2, t2, h3 = (w[i][1] for i in range(5)) q = _sim(h1, h2, atr) * _sim(h2, h3, atr) if q > 0.30 and h2 <= max(h1, h3) + 0.3 * atr: # kein Kopf → kein SKS neck = min(t1, t2) add("triple_top", "Dreifach-Top", "bear", q, neck, max(h1, h2, h3), f"drei Hochs {h1:.2f}/{h2:.2f}/{h3:.2f}, Nacken {neck:.2f}") if len(w) >= 5 and [x[0] for x in w[:5]] == ["L", "H", "L", "H", "L"]: l1, t1, l2, t2, l3 = (w[i][1] for i in range(5)) q = _sim(l1, l2, atr) * _sim(l2, l3, atr) if q > 0.30 and l2 >= min(l1, l3) - 0.3 * atr: neck = max(t1, t2) add("triple_bottom", "Dreifach-Boden", "bull", q, neck, min(l1, l2, l3), f"drei Tiefs {l1:.2f}/{l2:.2f}/{l3:.2f}, Nacken {neck:.2f}") # ── Flagge / Wimpel (Fortsetzung): impulsive STANGE, danach eine kleine # Konsolidierung GEGEN die Stangenrichtung. Flagge = paralleler Kanal, # Wimpel = konvergierend. Trigger = Ausbruch in Stangenrichtung. if len(w) >= 4: pole_h = abs(w[1][1] - w[0][1]) pole_bars = w[1][2] - w[0][2] if (pole_h >= _POLE_MIN_ATR * atr and 0 < pole_bars <= _POLE_MAX_BARS): up = w[1][1] > w[0][1] # Stange nach oben? cons = [x[1] for x in w[1:5]] rng = max(cons) - min(cons) if rng <= _FLAG_MAX_FRAC * pole_h and len(cons) >= 3: # konvergierend? (Schwankung nimmt ab → Wimpel) d1 = abs(cons[1] - cons[0]); d2 = abs(cons[-1] - cons[-2]) pennant = d2 < d1 * 0.7 trig = max(cons) if up else min(cons) ext = min(cons) if up else max(cons) typ = "pennant" if pennant else "flag" nm = ("Wimpel" if pennant else "Flagge") + (" (bullisch)" if up else " (bärisch)") add(typ, nm, "bull" if up else "bear", 0.55, trig, ext, f"Stange {pole_h/atr:.1f}×ATR in {pole_bars} Bars, " f"{'konvergierende' if pennant else 'flache'} Konsolidierung") # ── Rechteck (Range): zwei ~flache Hochs UND zwei ~flache Tiefs. Das Handbuch # führt es als Fortsetzung — die Ausbruchsrichtung ist aber offen, deshalb # NEUTRAL (wie das symmetrische Dreieck), nur als Kontext. hh = [(x[2], x[1]) for x in zip(kd, pr, ix) if x[0] == "H"][-2:] ll = [(x[2], x[1]) for x in zip(kd, pr, ix) if x[0] == "L"][-2:] if len(hh) == 2 and len(ll) == 2: top = (hh[0][1] + hh[1][1]) / 2.0 bot = (ll[0][1] + ll[1][1]) / 2.0 if (abs(hh[1][1] - hh[0][1]) <= _NECK_TOL_ATR * atr and abs(ll[1][1] - ll[0][1]) <= _NECK_TOL_ATR * atr and (top - bot) >= _MIN_HEIGHT_ATR * atr): out.append({ "type": "rectangle", "name": "Rechteck (Range)", "dir": "neutral", "quality": round(_sim(hh[0][1], hh[1][1], atr) * _sim(ll[0][1], ll[1][1], atr), 2), "trigger": round(top, 3), "target": None, "height_atr": round((top - bot) / atr, 1), "status": "forming", "bars_ago": last_idx - ix[-1], "measured": False, "note": f"Range {bot:.2f}–{top:.2f} — Ausbruchsrichtung offen", }) # ── Keil: beide Linien in DIESELBE Richtung geneigt und konvergierend. # Steigender Keil = bärisch, fallender Keil = bullisch (Handbuch). dh = hh[1][1] - hh[0][1] dl = ll[1][1] - ll[0][1] conv = abs(hh[1][1] - ll[1][1]) < abs(hh[0][1] - ll[0][1]) - 0.3 * atr if conv and dh > 0.3 * atr and dl > 0.3 * atr and dl > dh: add("wedge_rise", "Steigender Keil", "bear", 0.5, min(ll[0][1], ll[1][1]), max(hh[0][1], hh[1][1]), "beide Linien steigen, Tiefs schneller — Erschöpfung nach oben") # ⚠ Bedingung korrigiert (2026-07-31): stand `dh > dl` — bei einem FALLENDEN # Keil fällt aber die OBERE Linie schneller (sonst konvergiert nichts), also # `dh < dl`. Mit der alten Fassung war die Bedingung in sich widersprüchlich # (`conv` verlangt Konvergenz) → der Typ feuerte in 20.000 Bars KEIN EINZIGES # MAL. Beim Spiegelbild (steigender Keil, `dl > dh`) stimmte es. if conv and dh < -0.3 * atr and dl < -0.3 * atr and dh < dl: add("wedge_fall", "Fallender Keil", "bull", 0.5, max(hh[0][1], hh[1][1]), min(ll[0][1], ll[1][1]), "beide Linien fallen, Hochs schneller — Erschöpfung nach unten") # ── Dreiecke (Fortsetzung): letzte 2 Hochs + 2 Tiefs, Steigungen prüfen ── highs = [(x[2], x[1]) for x in zip(kd, pr, ix) if x[0] == "H"][-2:] lows = [(x[2], x[1]) for x in zip(kd, pr, ix) if x[0] == "L"][-2:] if len(highs) == 2 and len(lows) == 2: (ih1, h1p), (ih2, h2p) = highs (il1, l1p), (il2, l2p) = lows flat_h = abs(h2p - h1p) <= _NECK_TOL_ATR * atr flat_l = abs(l2p - l1p) <= _NECK_TOL_ATR * atr rising_l = l2p > l1p + 0.4 * atr falling_h = h2p < h1p - 0.4 * atr wide = abs(max(h1p, h2p) - min(l1p, l2p)) if wide >= _MIN_HEIGHT_ATR * atr: if flat_h and rising_l: # aufsteigendes Dreieck (bullisch) res_ = max(h1p, h2p) add("tri_asc", "Aufsteigendes Dreieck", "bull", 0.6, res_, min(l1p, l2p), f"flacher Widerstand {res_:.2f}, steigende Tiefs") elif flat_l and falling_h: # absteigendes Dreieck (bärisch) sup_ = min(l1p, l2p) add("tri_desc", "Absteigendes Dreieck", "bear", 0.6, sup_, max(h1p, h2p), f"flacher Support {sup_:.2f}, fallende Hochs") elif falling_h and rising_l: # symmetrisch → Ausbruch offen (nur Anzeige) out.append({ "type": "tri_sym", "name": "Symmetrisches Dreieck", "dir": "neutral", "quality": 0.5, "trigger": round((max(h1p, h2p) + min(l1p, l2p)) / 2, 3), "target": None, "height_atr": round(wide / atr, 1), "status": "forming", "bars_ago": last_idx - ix[-1], "note": f"konvergierend (Hochs fallen, Tiefs steigen) — Ausbruchsrichtung offen", "measured": False, # neutral, ohne Richtung → nie stimmberechtigt }) # Duplikate (gleicher Typ+Trigger) zusammenfassen, beste Qualität behalten best = {} for m in out: key = (m["type"], round(m["trigger"], 2)) if key not in best or m["quality"] > best[key]["quality"]: best[key] = m def played_out(m): """Ziel bereits erreicht → Muster abgearbeitet, nicht mehr zeigen.""" t = m.get("target") if t is None: return False return (cur <= t) if m["dir"] == "bear" else (cur >= t) # raus: ungültig · abgearbeitet (Ziel erreicht) · zu weit vom Kurs (nicht mehr # aktionabel — nur AKTIVE, kursnahe Muster zeigen). res = [m for m in best.values() if m["status"] != "invalidated" and not played_out(m) and abs(m["trigger"] - cur) <= _MAX_DIST_ATR * atr] rank = {"confirmed": 2, "forming": 1} res.sort(key=lambda m: (rank.get(m["status"], 0), m["quality"], -m["bars_ago"]), reverse=True) return res class PatternDetector: def __init__(self): self._snap = {"patterns": [], "atr": None, "ts": 0.0} self._last_refresh = 0.0 self._lock = threading.Lock() def refresh_market(self, sym: str): now = time.time() if now - self._last_refresh < _REFRESH_S: return self._last_refresh = now try: with mt5_lock(timeout=2) as got: if not got: return bars = mt5.copy_rates_from_pos(sym, _TF, 0, _N_BARS) if bars is None or len(bars) < 60: return H = [float(b["high"]) for b in bars] L = [float(b["low"]) for b in bars] C = [float(b["close"]) for b in bars] atr = _atr(H, L, C) if not atr or atr <= 0: return piv = _pivots(H, L, _PIVOT_K) patterns = _detect(piv, C[-1], atr, len(H) - 1) with self._lock: self._snap = {"patterns": patterns, "atr": round(atr, 4), "cur": round(C[-1], 3), "ts": now} except Exception as e: log.warning(f"patterns.refresh_market: {e}") def snapshot(self) -> dict: with self._lock: return dict(self._snap)