"""core/candles.py — Kerzen-Anatomie (Body/Docht-Klassifikation). REINE ANZEIGE. User-Vorgabe 2026-07-31 („Kerzengröße und Dochtlänge … im Kontext mit Trendrichtung, Volumen und Liquiditätszonen"). Erkennt mechanisch aus OHLCV die sechs Klassen Marubozu · Langer Körper · Langer Docht (oben/unten) · Spinning Top · Doji · Engulfing. ⚠⚠ WICHTIG — DIE BESCHRIFTUNG FOLGT DER MESSUNG, NICHT DEM LEHRBUCH. `backtest_candles.py` (80k M5-Bars, 2 Halbjahre, Forward-Return gegen die DRIFT-NULL gelesen) hat die gängige Deutung in BEIDEN Hälften **invertiert** vorgefunden: * „Langer Körper = Momentum, im Trend = Fortsetzung" → Überschuss −0,059/−0,076; IM Trend sogar −0,132/−0,042. Es ist die **Klimax-/Erschöpfungskerze**, nicht die Fortsetzung. * „Langer Körper mit 2× Volumen = echt" → −0,117/−0,204, monoton schlechter mit MEHR Volumen. Volumen bestätigt hier nicht, es kapituliert. * „Docht am Level = zuverlässig, Range-Mitte = Noise" → am Level in BEIDEN Hälften SCHLECHTER als in der Range-Mitte. * „mit/gegen Trend" beim Docht kippt zwischen den Hälften = Rauschen. Der Folgetest als Handelssignal (`backtest_candle_fade.py`, Live-Exit + Echtkosten) fiel ebenfalls durch (benachbarte Volumen-Schwellen exakt gegenläufig). Deshalb: **KEIN Verdict-Gewicht, kein Trade-Trigger** — die Kachel zeigt nur, WAS gerade gedruckt wurde, mit der gemessenen Lesart als Hinweis. Wer sie als Signal liest, handelt gegen die eigene Datenlage. Warum trotzdem gebaut: dieselbe Rolle wie `structure.py`/`patterns.py` — sichtbar machen, was ein Chart-Leser sieht, ohne es zu bewerten. Der Hinweistext ist bewusst so formuliert, dass er NICHT in die Lehrbuch-Richtung stupst (dieselbe Korrektur wie bei der Bounce-Anzeige, die deshalb heute „Überdehnung" heißt und bernstein ist). """ from __future__ import annotations import threading import time from statistics import median import MetaTrader5 as mt5 from core.logger import get_logger from core.mt5_utils import mt5_lock log = get_logger("candles") _REFRESH_S = 20.0 # Kachel muss nicht schneller sein als eine M5-Kerze _N_BARS = 60 # Referenzfenster (Median-Body, Ø-Volumen) _REF_N = 20 # Fenster für „lang" und „viel Volumen" _LOOK = 6 # so viele letzte Kerzen werden klassifiziert # Schwellen — identisch zu `backtest_candles.py`, damit Anzeige und Messung # dieselbe Sprache sprechen. _MARU_BR = 0.90 # Body/Range ab hier = Marubozu _LONG_BR = 0.60 # Body/Range für „langer Körper" _LONG_X = 1.5 # Body >= X × Median-Body _DOJI_BR = 0.10 _SPIN_BR = 0.30 _SPIN_W = 0.30 # beide Dochte > X × Range _WICK = 0.50 # Docht/Range ab hier = „langer Docht" _VOL_HOT = 2.0 # Volumen >= X × Ø class CandleAnatomy: """Klassifiziert die letzten Kerzen einer TF. Thread-safe, self-throttled.""" def __init__(self, tf: str = "M5"): self.tf = tf self._lock = threading.Lock() self._snap: dict = {} self._ts = 0.0 # ── Klassifikation (rein rechnerisch, ohne MT5) ────────────────────────── @staticmethod def classify(o, h, l, c, body_ref, prev=None): """→ (klasse, richtung, kennzahlen). richtung 0 = keine implizierte Richtung. `prev` = (o,c) der Vorkerze, nur für Engulfing. """ rng = h - l if rng <= 0: return None, 0, {} body = abs(c - o) up_w = h - max(o, c) lo_w = min(o, c) - l br = body / rng d = 1 if c > o else -1 if c < o else 0 m = {"body_pct": round(100 * br), "wick_up_pct": round(100 * up_w / rng), "wick_lo_pct": round(100 * lo_w / rng)} long_body = body_ref > 0 and body >= _LONG_X * body_ref # Engulfing zuerst prüfen (zwei-Kerzen-Muster schlägt Ein-Kerzen-Form) if prev and long_body: po, pc = prev if d > 0 and pc < po and c >= po and o <= pc: return "Engulfing", 1, m if d < 0 and pc > po and c <= po and o >= pc: return "Engulfing", -1, m if br >= _MARU_BR: return "Marubozu", d, m if br < _DOJI_BR: return "Doji", 0, m if br < _SPIN_BR and up_w / rng > _SPIN_W and lo_w / rng > _SPIN_W: return "SpinningTop", 0, m if up_w / rng >= _WICK: return "LangerDochtOben", -1, m if lo_w / rng >= _WICK: return "LangerDochtUnten", 1, m if long_body and br >= _LONG_BR: return "LangerKoerper", d, m return None, 0, m # ── Anzeige-Text: GEMESSENE Lesart, nicht die Lehrbuch-Lesart ──────────── @staticmethod def reading(cls: str, d: int, hot: bool) -> str: if cls in ("LangerKoerper", "Marubozu", "Engulfing"): base = ("⚠ Klimax-Kerze — gemessen läuft der Kurs danach eher GEGEN " "die Kerze (nicht Fortsetzung)") if hot: base += " · hohes Volumen verstärkt das (gemessen −0,12…−0,20×ATR)" return base if cls in ("LangerDochtOben", "LangerDochtUnten"): seite = "oben" if cls.endswith("Oben") else "unten" return (f"Ablehnung nach {seite} — ⚠ am Level gemessen NICHT zuverlässiger " f"als in der Range-Mitte") if cls == "SpinningTop": return "Unentschlossenheit (kleiner Körper, beide Seiten gedrückt)" if cls == "Doji": return "Gleichstand — keine Seite dominiert" return "" LABELS = {"Marubozu": "Marubozu", "LangerKoerper": "Langer Körper", "LangerDochtOben": "Langer Docht oben", "LangerDochtUnten": "Langer Docht unten", "SpinningTop": "Spinning Top", "Doji": "Doji", "Engulfing": "Engulfing"} # ── Datenbeschaffung ───────────────────────────────────────────────────── def refresh_market(self, symbol: str) -> None: """Im Trend-Loop aufrufen; self-throttled. Nimmt den `mt5_lock` selbst.""" if not symbol or time.time() - self._ts < _REFRESH_S: return self._ts = time.time() tf_map = {"M1": mt5.TIMEFRAME_M1, "M5": mt5.TIMEFRAME_M5, "M15": mt5.TIMEFRAME_M15, "M30": mt5.TIMEFRAME_M30, "H1": mt5.TIMEFRAME_H1} try: # ⚠ `mt5_lock` ist ein CONTEXT-MANAGER (kein Lock-Objekt) und liefert # `got` — nicht .acquire()/.release() verwenden. with mt5_lock(3.0) as got: if not got: return bars = mt5.copy_rates_from_pos( symbol, tf_map.get(self.tf, mt5.TIMEFRAME_M5), 0, _N_BARS + 2) if bars is None or len(bars) < _REF_N + 3: return self._build(bars) except Exception as e: log.warning(f"candles.refresh_market: {e}") def _build(self, bars) -> None: O = [float(b["open"]) for b in bars]; H = [float(b["high"]) for b in bars] L = [float(b["low"]) for b in bars]; C = [float(b["close"]) for b in bars] V = [float(b["tick_volume"]) for b in bars] # letzte Kerze ist noch NICHT abgeschlossen → nur bis -1 klassifizieren last = len(C) - 1 out = [] for i in range(max(_REF_N, last - _LOOK), last): bref = median([abs(C[k] - O[k]) for k in range(i - _REF_N, i)]) vref = sum(V[i - _REF_N:i]) / _REF_N cls, d, m = self.classify(O[i], H[i], L[i], C[i], bref, prev=(O[i - 1], C[i - 1])) if not cls: continue vol_x = round(V[i] / vref, 1) if vref > 0 else 0.0 hot = vol_x >= _VOL_HOT out.append({ "cls": cls, "label": self.LABELS.get(cls, cls), "dir": d, "bars_ago": last - i, "vol_x": vol_x, "hot": hot, "reading": self.reading(cls, d, hot), **m, }) out.reverse() # neueste zuerst with self._lock: self._snap = {"tf": self.tf, "items": out, "vol_hot_x": _VOL_HOT, "ts": time.time()} def snapshot(self) -> dict: with self._lock: return dict(self._snap)