""" 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 _keil(a1, a2, b1, b2, last_idx, max_extra=40): """Zwei konvergierende Linien -> DREIECK: linke Kante + Apex. `a`/`b` je zwei (bar_index, preis). Der Apex ist der Schnittpunkt. ⚠ Er wird auf `last_idx + max_extra` GEDECKELT: ein fast paralleles Linienpaar schneidet sich irgendwo im Unendlichen und wuerde sonst ein Dreieck ueber den halben Chart aufspannen. Nach links begrenzt auf mindestens einen Bar hinter dem letzten Stuetzpunkt, damit der Apex nicht INNERHALB des Musters landet und die Form sich selbst durchdringt. """ (x1, y1), (x2, y2) = a1, a2 (x3, y3), (x4, y4) = b1, b2 ma = (y2 - y1) / (x2 - x1) if x2 != x1 else 0.0 mb = (y4 - y3) / (x4 - x3) if x4 != x3 else 0.0 x0 = min(x1, x3) if abs(ma - mb) < 1e-12: xa = last_idx + max_extra else: xa = (y3 - mb * x3 - y1 + ma * x1) / (ma - mb) xa = max(min(xa, last_idx + max_extra), max(x2, x4) + 1) return [(int(x0), y1 + ma * (x0 - x1)), (int(x0), y3 + mb * (x0 - x3)), (int(round(xa)), y1 + ma * (xa - x1))] def _detect(piv, cur, atr, last_idx, T=None, bar_s=0): """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 _anker(punkte): """[(bar_index, preis)] -> [[epoch, preis]] fuer den MT5-Export. ⚠ Ein Index JENSEITS des letzten Bars (der Dreiecks-Apex liegt in der Zukunft) wird ueber die Bar-Dauer hochgerechnet — `T` hat dafuer keinen Eintrag. Ohne das waere der Apex auf den letzten Bar geklemmt und jedes Dreieck haette einen senkrechten rechten Rand. """ if not T: return None aus = [] for i, pr_ in punkte: i = int(i) if 0 <= i < len(T): t = int(T[i]) elif i >= len(T): t = int(T[-1] + (i - len(T) + 1) * max(bar_s, 60)) else: t = int(T[0]) aus.append([t, round(float(pr_), 3)]) return aus def add(typ, name, dir_, quality, trigger, extreme, note, form=None, punkte=None): 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, # ⚠ Geometrie fuer den MT5-Export (User-Vorgabe 2026-08-21: die # Muster als gefuellte Formen zeichnen). REIN ADDITIV — kein # bestehender Schluessel aendert sich; das Verdict-Gewicht liest # ohnehin nur type/dir/status/quality. "form": form, "anker": _anker(punkte) if punkte else None, }) # ü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}", "dreieck", [(w[0][2], Ha), (w[1][2], Lm), (w[2][2], Hb)]) # ── 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}", "dreieck", [(w[0][2], La), (w[1][2], Hm), (w[2][2], Lb)]) # ── 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}", "dreieck2", [(w[i][2], w[i][1]) for i in range(5)]) # ── 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}", "dreieck2", [(w[i][2], w[i][1]) for i in range(5)]) # ── (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)", "ellipse", [(w[0][2], rim_l), (w[2][2], rim_r), (w[1][2], top)]) 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)", "ellipse", [(w[0][2], rim_l), (w[2][2], rim_r), (w[1][2], bot)]) # ══ 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}", "dreieck2", [(w[i][2], w[i][1]) for i in range(5)]) 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}", "dreieck2", [(w[i][2], w[i][1]) for i in range(5)]) # ── 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)") _ci = [x[2] for x in w[1:5]] # Wimpel = konvergierend -> Dreieck; Flagge = paralleler # Kanal -> Rechteck (zwei diagonale Ecken). _pk = ([(_ci[0], max(cons)), (_ci[0], min(cons)), (_ci[-1], cons[-1])] if pennant else [(_ci[0], max(cons)), (_ci[-1], min(cons))]) 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", "dreieck" if pennant else "rechteck", _pk) # ── 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", "form": "rechteck", "anker": _anker([(hh[0][0], top), (ll[1][0], bot)]), }) # ── 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", "dreieck", _keil(hh[0], hh[1], ll[0], ll[1], last_idx)) # ⚠ 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", "dreieck", _keil(hh[0], hh[1], ll[0], ll[1], last_idx)) # ── 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", "dreieck", _keil(highs[0], highs[1], lows[0], lows[1], last_idx)) 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", "dreieck", _keil(highs[0], highs[1], lows[0], lows[1], last_idx)) 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 "form": "dreieck", "anker": _anker(_keil(highs[0], highs[1], lows[0], lows[1], last_idx)), }) # 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 # ⚠⚠ BILD je Mustertyp (User-Vorgabe 2026-08-20: das gerade entstehende Muster # als Bild zeigen). Die Dateien liegen verkleinert in `web/muster/` — die # Originale in `Chartmuster/` sind 436-611 KB je Stueck; das Dashboard # uebertraegt sonst ~14 KB je Snapshot, ein Originalbild waere das 40-fache. # Verkleinert: 7,5 MB -> 379 KB fuer alle 15. # ⚠ Doppeltop und Doppelboden teilen sich ein Bild (die Vorlage zeigt beide), # ebenso Dreifach-Top/-Boden und Flagge bullisch/baerisch. # ⚠ `inv_cup` hat seit dem 20.08. ein eigenes Bild (vorher bewusst keins). BILD = { "double_top": "double", "double_bottom": "double", "triple_top": "triple", "triple_bottom": "triple", "hs": "hs", "inv_hs": "inv_hs", "cup": "cup", # ⚠ Am 2026-08-20 nachgereicht (User: "scanne neue Bilder"). Bis dahin # stand hier ausdruecklich KEIN Eintrag — lieber kein Bild als ein # falsches; die normale Tasse zu zeigen waere die falsche Richtung gewesen. "inv_cup": "inv_cup", "flag": "flag", "pennant": "pennant", "wedge_rise": "wedge_rise", "wedge_fall": "wedge_fall", "tri_asc": "tri_asc", "tri_desc": "tri_desc", "tri_sym": "tri_sym", "rect": "rect", "rounding_bottom": "rounding_bottom", "diamond": "diamond", } class PatternDetector: def __init__(self, tf=None, label: str = "M30", refresh_s: float = _REFRESH_S, bars: int = _N_BARS): """⚠ `tf=None` = M30, der bisherige Stand — der Default bleibt bitgleich. Die Zeitebene ist seit dem 2026-08-20 ein PARAMETER (User-Wunsch: das gerade entstehende Muster je Minuten-Chart anzeigen). Vorher war sie die Modul-Konstante `_TF`; die bleibt als Default stehen, damit der bestehende Aufruf `PatternDetector()` sich NICHT aendert — er speist das Verdict-Gewicht 0,25, und eine stille Verschiebung dort waere eine Strategie-Aenderung durch die Hintertuer. """ self._tf = _TF if tf is None else tf self._label = label self._refresh_s = float(refresh_s) self._bars = int(bars) self._snap = {"patterns": [], "atr": None, "ts": 0.0, "tf": label} self._last_refresh = 0.0 self._lock = threading.Lock() def refresh_market(self, sym: str): now = time.time() if now - self._last_refresh < self._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, self._tf, 0, self._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 T = [int(b["time"]) for b in bars] # ⚠ Bar-Dauer aus den DATEN, nicht aus der TF-Konstante — so # stimmt sie auch, wenn der Broker Bars auslaesst. _bs = (T[-1] - T[-2]) if len(T) > 1 else 0 piv = _pivots(H, L, _PIVOT_K) patterns = _detect(piv, C[-1], atr, len(H) - 1, T, _bs) with self._lock: self._snap = {"patterns": patterns, "atr": round(atr, 4), "cur": round(C[-1], 3), "ts": now, "tf": self._label} except Exception as e: log.warning(f"patterns.refresh_market: {e}") def snapshot(self) -> dict: with self._lock: return dict(self._snap)