AH-Oil-Trader/core/candles.py

"""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)