AH-Oil-Trader/core/history.py

"""
core/history.py — SQLite-basierter Verlaufs-Logger
====================================================
Speichert Trades, KI-Analysen, Empfehlungen und Signale in einer lokalen
SQLite-Datenbank für spätere statistische Auswertung.

Tabellen:
  trades           — jeder geöffnete + geschlossene Trade
  ai_analyses      — jede ChatGPT-Antwort
  recommendations  — Algorithm-Empfehlungen (gefiltert: alle 60 s)
  signals          — Reversal, Breakout, EMA-Cross etc.

Alle Schreiboperationen sind thread-safe und werden in einem internen
Lock serialisiert.  Lese-Queries (für Stats) können parallel laufen.

Wichtig: Diese Klasse macht KEIN Machine Learning. Sie loggt nur.
Auswertung passiert separat über die `stats_*`-Methoden.
"""

from __future__ import annotations

import sqlite3
import threading
import time
import json
from pathlib import Path
from datetime import datetime, timedelta

# history sendet selbst kein Telegram mehr — Close-Pushes laufen über die Engine
# (Flip-Close-Alarm / Notfall-/Gewinn-Auto-Close).


SCHEMA = [
    """
    CREATE TABLE IF NOT EXISTS trades (
        id              INTEGER PRIMARY KEY AUTOINCREMENT,
        ticket          INTEGER UNIQUE,
        symbol          TEXT,
        direction       TEXT,           -- 'BUY' | 'SELL'
        lots            REAL,
        entry_time      INTEGER,        -- unix timestamp (s)
        entry_price     REAL,
        sl_at_entry     REAL,
        tp_at_entry     REAL,
        exit_time       INTEGER,
        exit_price      REAL,
        pnl             REAL,
        closed_by       TEXT,           -- 'manual' | 'sl' | 'tp' | 'trail' | 'unknown'
        ai_sentiment    TEXT,           -- KI-Bewertung beim Einstieg
        ai_confidence   INTEGER,        -- 0-100
        rec_signal      TEXT,           -- 'LONG' | 'SHORT' | 'WARTEN'
        rec_score       REAL,           -- -1.0 .. +1.0
        setup           TEXT,           -- 'TREND_PULLBACK_LONG' | 'BREAKOUT_SHORT' | ...
        regime          TEXT,           -- 'trend_up' | 'trend_down' | 'range' | 'transition'
        rsi_at_entry    REAL,           -- RSI(14) M15 beim Einstieg
        news_score      REAL            -- News-Sentiment -1..+1 beim Einstieg
    )
    """,
    """
    CREATE TABLE IF NOT EXISTS ai_analyses (
        id              INTEGER PRIMARY KEY AUTOINCREMENT,
        timestamp       INTEGER,
        sentiment       TEXT,
        confidence      INTEGER,
        summary         TEXT,
        drivers_json    TEXT,
        cost_estimate   REAL,
        model           TEXT
    )
    """,
    """
    CREATE TABLE IF NOT EXISTS recommendations (
        id              INTEGER PRIMARY KEY AUTOINCREMENT,
        timestamp       INTEGER,
        signal          TEXT,
        score           REAL,
        conf_pct        INTEGER,
        angle_m5        REAL,
        angle_m15       REAL,
        angle_m30       REAL,
        angle_h1        REAL,
        reversal        TEXT,
        ai_sentiment    TEXT,
        ai_confidence   INTEGER,
        setup           TEXT,
        regime          TEXT,
        rsi             REAL,
        news_score      REAL,
        -- Maschinenlesbarer Grund für WARTEN (Stufe-2-Telemetrie 2026-07-31):
        -- deadband · dead_hour · eia · htf_counter · stretch · min_conf ·
        -- breakout_pending · entry_room · no_data · stale. NULL = kein Block.
        -- Ohne den war live nicht feststellbar, WELCHES Gate die 93 % WARTEN
        -- erzeugt (Backtest-Erwartung ~43 %).
        block_reason    TEXT
    )
    """,
    """
    CREATE TABLE IF NOT EXISTS intended_trades (
        id              INTEGER PRIMARY KEY AUTOINCREMENT,
        open_ts         INTEGER,
        open_price      REAL,
        direction       TEXT,           -- 'BUY' | 'SELL'
        sl              REAL,
        tp              REAL,
        setup           TEXT,
        regime          TEXT,
        rsi             REAL,
        score           REAL,
        conf_pct        INTEGER,
        news_score      REAL,
        ai_sentiment    TEXT,
        ai_confidence   INTEGER,
        close_ts        INTEGER,        -- NULL = noch offen
        close_price     REAL,
        close_reason    TEXT,           -- 'sl' | 'tp' | 'flip' | 'expired' | 'disabled'
        pnl_pct         REAL            -- (close-open)/open * 100, vorzeichenrichtig
    )
    """,
    """
    CREATE TABLE IF NOT EXISTS signals (
        id              INTEGER PRIMARY KEY AUTOINCREMENT,
        timestamp       INTEGER,
        signal_type     TEXT,           -- 'reversal' | 'breakout' | 'ema_cross'
        direction       TEXT,           -- 'bullish' | 'bearish'
        price           REAL,
        details_json    TEXT
    )
    """,
    """
    CREATE TABLE IF NOT EXISTS pbreak_predictions (
        id              INTEGER PRIMARY KEY AUTOINCREMENT,
        ts              INTEGER,        -- lokale Epoch (wie trades/recommendations)
        symbol          TEXT,
        direction       TEXT,           -- 'LONG' | 'SHORT' (Positionsrichtung)
        level           REAL,
        p_break         REAL,           -- geglättete P(break) 0..100 beim Touch
        predicted       TEXT,           -- 'break' | 'bounce' (P(break) vs. sr_close_pbreak)
        price_at_pred   REAL,           -- bid beim Touch
        confirm_price   REAL,           -- level ± 0,5×ATR in Richtung Durchbruch
        reject_price    REAL,           -- level ± 0,5×ATR in Richtung Abprall
        atr             REAL,
        outcome         TEXT,           -- NULL bis ausgewertet, dann 'break' | 'bounce'
        outcome_ts      INTEGER,
        correct         INTEGER         -- NULL bis ausgewertet, dann 0/1
    )
    """,
    """
    CREATE TABLE IF NOT EXISTS rec_outcomes (
        id          INTEGER PRIMARY KEY AUTOINCREMENT,
        ts          INTEGER,   -- lokale Epoch, Beginn der EPISODE
        signal      TEXT,      -- 'LONG' | 'SHORT'
        conf_pct    INTEGER,
        tf          TEXT,
        price       REAL,      -- Kurs bei Ausgabe der Empfehlung
        atr         REAL,      -- ATR_M5 zur Normierung
        ret30       REAL,      -- (Kurs[+30min] - price) x Richtung / atr
        ret60       REAL,
        hit30       INTEGER,   -- 1 = Kurs lief in Signalrichtung
        hit60       INTEGER,
        done        INTEGER    -- 1 = beide Horizonte ausgewertet
    )
    """,
    """
    -- M15-KARTE misst sich SELBST (2026-08-08).
    -- ⚠ WARUM: jeder Ausfall, der in diesem Projekt je aufgedeckt wurde, kam aus
    -- einer Rueckkopplung — `pbreak_predictions` entlarvte den Modellausfall LIVE
    -- (was kein Backtest konnte), der B4-Monitor die Squeeze-Drift. Die Karte
    -- hatte nichts davon und koennte monatelang falsch stehen, ohne dass es
    -- auffaellt. Geloggt wird EINE Zeile je STATUSWECHSEL (nicht je Tick — sonst
    -- zaehlt man dieselbe Marktlage dutzendfach; die Lehre aus rec_outcomes und
    -- der Clusterung bei pbreak_predictions).
    CREATE TABLE IF NOT EXISTS m15_states (
        id          INTEGER PRIMARY KEY AUTOINCREMENT,
        ts          INTEGER,   -- lokale Epoch, Beginn des Zustands
        status      TEXT,      -- 'gruen' | 'gelb' | 'rot'
        veto        TEXT,      -- verbotene Richtung ('LONG'/'SHORT'/NULL)
        kosten      REAL,      -- Spread/ATR_M15
        level       REAL,      -- naechstes stehendes Level
        gap_atr     REAL,      -- Abstand Kurs<->Level in ATR_M15
        p_break     INTEGER,   -- Durchbruchwahrscheinlichkeit in %
        lesart      TEXT,      -- 'auf' | 'ab' | 'unklar' (die Richtungs-Lesart)
        price       REAL,
        atr         REAL,      -- ATR_M15 zur Normierung
        ret30       REAL,      -- (Kurs[+30min]-price) x Lesart-Richtung / atr
        ret60       REAL,
        hit30       INTEGER,   -- 1 = Kurs lief in Richtung der Lesart
        hit60       INTEGER,
        done        INTEGER
    )
    """,
    """
    -- KEGEL-ABDECKUNG fuer Adaptive Conformal Inference (2026-08-08).
    -- ⚠ NICHT-UEBERLAPPEND je Horizont geloggt: ein 120-min-Fenster alle 5 min
    -- waere 24-fach ueberlappend, und ueberlappende Beobachtungen haben genau
    -- die Cluster-Verzerrung erzeugt, an der `analyze_hl_funding.py` (06.08.)
    -- und die Faelligkeitsbedingung `pbreak_accuracy_v2` (07.08.) aufgelaufen
    -- sind. Je Horizont startet ein neuer Check erst, wenn der vorige ablief.
    CREATE TABLE IF NOT EXISTS cone_checks (
        id      INTEGER PRIMARY KEY AUTOINCREMENT,
        ts      INTEGER,   -- lokale Epoch
        bars    INTEGER,   -- Horizont in M5-Bars (6/12/24)
        band    INTEGER,   -- Nennband in % (80)
        price   REAL,
        lo      REAL,
        hi      REAL,
        skala   REAL,      -- ACI-Skala, mit der dieses Band gebaut wurde
        drin    INTEGER,   -- 1 = Kurs lag im Band
        done    INTEGER
    )
    """,
    """
    CREATE TABLE IF NOT EXISTS verdict_votes (
        id              INTEGER PRIMARY KEY AUTOINCREMENT,
        ts              INTEGER,        -- lokale Epoch
        headline        TEXT,           -- Wave-Signal (LONG/SHORT/WARTEN)
        conf            INTEGER,        -- conf_pct der Welle
        bias            REAL,           -- gewichteter Konsens-Bias −1..+1
        tf              TEXT,           -- Zeitebene der Empfehlung
        wave            INTEGER,        -- Votes je Modul: +1/−1/0
        m30             INTEGER,
        h1              INTEGER,
        ki              INTEGER,        -- KI-Copilot-Vote (0 auch wenn ohne Aussage)
        ki_w            REAL,           -- Copilot-Gewicht (0 = hatte keine Aussage)
        elliott         INTEGER,
        elliott_w       REAL,
        squeeze         INTEGER,
        news_score      REAL,           -- News-Sentiment zum Zeitpunkt (Kontext)
        -- ab 2026-07-30 (User-Vorgabe, Stimmen mit KLEINEM Gewicht):
        pattern         INTEGER,        -- Chartmuster-Vote (Kontrolltest: kleiner Zusatz)
        pattern_w       REAL,           -- 1,0 bestaetigt / 0,5 bildet sich / 0 keins
        orderbook       INTEGER,        -- HL-Orderbuch-Imbalance (UNGEMESSEN)
        orderbook_w     REAL,
        liqtrend        INTEGER,        -- Liquiditaets-Trend (UNGEMESSEN)
        liqtrend_w      REAL
    )
    """,
]

# Indizes — werden NACH den Migrationen ausgeführt, weil sie Spalten
# referenzieren, die u.U. erst per ALTER TABLE hinzugefügt werden.
INDEXES = [
    "CREATE INDEX IF NOT EXISTS idx_trades_entry     ON trades(entry_time)",
    "CREATE INDEX IF NOT EXISTS idx_trades_exit      ON trades(exit_time)",
    "CREATE INDEX IF NOT EXISTS idx_trades_setup     ON trades(setup)",
    "CREATE INDEX IF NOT EXISTS idx_trades_closed_by ON trades(closed_by)",
    "CREATE INDEX IF NOT EXISTS idx_trades_open      ON trades(exit_time) WHERE exit_time IS NULL",
    "CREATE INDEX IF NOT EXISTS idx_ai_ts            ON ai_analyses(timestamp)",
    "CREATE INDEX IF NOT EXISTS idx_rec_ts           ON recommendations(timestamp)",
    "CREATE INDEX IF NOT EXISTS idx_rec_setup        ON recommendations(setup)",
    "CREATE INDEX IF NOT EXISTS idx_signals_ts       ON signals(timestamp)",
    "CREATE INDEX IF NOT EXISTS idx_intended_open    ON intended_trades(open_ts)",
    "CREATE INDEX IF NOT EXISTS idx_intended_close   ON intended_trades(close_ts)",
    "CREATE INDEX IF NOT EXISTS idx_pbreak_ts        ON pbreak_predictions(ts)",
    "CREATE INDEX IF NOT EXISTS idx_pbreak_open      ON pbreak_predictions(outcome) WHERE outcome IS NULL",
    "CREATE INDEX IF NOT EXISTS idx_verdict_ts       ON verdict_votes(ts)",
]

# Migrations: Spalten, die in alten DBs evtl. fehlen.
# `ALTER TABLE ADD COLUMN` ist in SQLite idempotent über try/except.
MIGRATIONS = [
    ("trades",         "setup",        "TEXT"),
    ("trades",         "regime",       "TEXT"),
    ("trades",         "rsi_at_entry", "REAL"),
    ("trades",         "news_score",   "REAL"),
    ("trades",         "commission",   "REAL"),
    ("recommendations","setup",        "TEXT"),
    ("recommendations","regime",       "TEXT"),
    ("recommendations","rsi",          "REAL"),
    ("recommendations","news_score",   "REAL"),
]


class HistoryLogger:
    """Persistenter SQLite-Logger für Trading-Verlauf."""

    def __init__(self, db_path: Path,
                 rec_min_interval_s: int = 60):
        self.db_path = db_path
        self._lock = threading.Lock()
        self._last_rec_ts = 0          # Throttling für recommendations
        self.rec_min_interval = rec_min_interval_s
        self._telegram: dict = {"enabled": False, "token": "", "chat_id": ""}
        self._last_optimize_ts: float = 0.0
        self._init_db()

    def configure_telegram(self, *, enabled: bool, token: str, chat_id: str):
        """Telegram-Benachrichtigungen konfigurieren (nach Config-Load aufrufen)."""
        self._telegram = {"enabled": enabled, "token": token, "chat_id": chat_id}

    # ── Verbindung & Schema ─────────────────
    def _connect(self):
        # Eine neue Verbindung pro Thread vermeidet SQLite-Threading-Issues.
        # Da wir alles unter Lock haben, ist eine Connection auch ok – aber wir
        # bleiben safer mit Thread-Local Verbindungen.
        conn = sqlite3.connect(str(self.db_path), timeout=5.0)
        conn.row_factory = sqlite3.Row
        conn.execute("PRAGMA journal_mode = WAL")
        now = time.time()
        if now - self._last_optimize_ts > 86400:
            conn.execute("PRAGMA optimize")
            self._last_optimize_ts = now
        return conn

    def _init_db(self):
        with self._connect() as conn:
            # 1. Tabellen anlegen (nur Spalten, keine Indizes)
            for stmt in SCHEMA:
                conn.execute(stmt)
            # 2. Migrationen: fehlende Spalten in bestehenden DBs nachziehen.
            #    MUSS vor den CREATE-INDEX-Statements laufen, weil manche
            #    Indizes auf Spalten zeigen, die erst hier hinzukommen.
            for table, col, ctype in MIGRATIONS:
                try:
                    conn.execute(f"ALTER TABLE {table} ADD COLUMN {col} {ctype}")
                except sqlite3.OperationalError:
                    pass  # Spalte existiert bereits
            # 3. Jetzt sind alle Spalten garantiert da → Indizes anlegen
            for stmt in INDEXES:
                conn.execute(stmt)
            conn.commit()

    # ══════════════════════════════════════════
    #  WRITE-METHODEN
    # ══════════════════════════════════════════

    def log_trade_open(self, *, ticket: int, symbol: str, direction: str,
                       lots: float, entry_price: float,
                       sl_at_entry: float | None = None,
                       tp_at_entry: float | None = None,
                       ai_sentiment: str | None = None,
                       ai_confidence: int | None = None,
                       rec_signal: str | None = None,
                       rec_score: float | None = None,
                       setup: str | None = None,
                       regime: str | None = None,
                       rsi_at_entry: float | None = None,
                       news_score: float | None = None,
                       ctx: dict | None = None):
        """Loggt das Öffnen eines Trades. Idempotent über UNIQUE(ticket).

        `ctx` = vollständiger Entscheidungszustand beim Einstieg (Stufe 1 des
        „Manuelle Trades"-Moduls, 2026-08-04). Bis dahin wurden nur 5 Felder
        gespeichert — davon `ai_sentiment` in 1,4 % und `news_score` in 34 % der
        Fälle befüllt. Damit war NICHT rekonstruierbar, in welcher Marktlage der
        User eingestiegen ist, und jede Auswertung seiner Diskretion blieb
        Spekulation. Alle `ctx_*`-Werte werden im Snapshot ohnehin berechnet;
        hier werden sie nur mitgeschrieben (reine Telemetrie, kein Verhalten).
        Unbekannte Schlüssel werden ignoriert — so bricht ein späterer Ausbau
        nichts."""
        # Guard (Fix 2026-07-19): 0-Lot-/Preis-lose Einträge sind Reconcile-
        # Artefakte (real: Ticket 47966457, 0.0L, closed_by=unknown) — sie
        # verfälschen WR/Statistik. Nicht loggen.
        if not lots or lots <= 0 or not entry_price or entry_price <= 0:
            return
        now_ts = int(time.time())
        # Kontext-Spalten dynamisch anhängen — nur solche, die es in der Tabelle
        # wirklich gibt. So läuft eine ältere DB ohne Migration weiter (fail-open),
        # statt beim INSERT zu werfen.
        ctx = ctx or {}
        with self._lock, self._connect() as conn:
            vorhanden = {r[1] for r in conn.execute("PRAGMA table_info(trades)")}
            extra = [(f"ctx_{k}", v) for k, v in ctx.items()
                     if f"ctx_{k}" in vorhanden and v is not None]
            spalten = ("ticket, symbol, direction, lots, entry_time, entry_price, "
                       "sl_at_entry, tp_at_entry, ai_sentiment, ai_confidence, "
                       "rec_signal, rec_score, setup, regime, rsi_at_entry, news_score")
            werte = [ticket, symbol, direction, lots, now_ts, entry_price,
                     sl_at_entry, tp_at_entry, ai_sentiment, ai_confidence,
                     rec_signal, rec_score, setup, regime, rsi_at_entry, news_score]
            for name, v in extra:
                spalten += f", {name}"
                werte.append(v)
            platz = ",".join("?" * len(werte))
            conn.execute(f"INSERT OR IGNORE INTO trades ({spalten}) VALUES ({platz})",
                         werte)
            conn.commit()
            # Kein Telegram bei Einstieg — nur Ergebnis (Close) wird gesendet

    def log_trade_close(self, *, ticket: int, exit_price: float,
                        pnl: float, closed_by: str = "manual",
                        exit_ts: int | None = None,
                        commission: float = 0.0):
        """Schreibt Exit-Daten zu einem bestehenden Trade.
        exit_ts kann gesetzt werden für Reconciliation alter Trades.

        Sanity-Check: wenn exit_ts < entry_time (kaputter Deal-Lookup),
        wird stattdessen time.time() benutzt.
        """
        ts = int(exit_ts) if exit_ts else int(time.time())
        with self._lock, self._connect() as conn:
            # Defensive: exit kann nicht vor entry liegen.
            row = conn.execute(
                "SELECT entry_time FROM trades "
                "WHERE ticket = ? AND exit_time IS NULL", (ticket,)
            ).fetchone()
            if row and row["entry_time"] and ts < row["entry_time"]:
                ts = int(time.time())
            cur = conn.execute("""
                UPDATE trades
                   SET exit_time = ?, exit_price = ?, pnl = ?,
                       closed_by = ?, commission = ?
                 WHERE ticket = ? AND exit_time IS NULL
            """, (ts, exit_price, pnl, closed_by, commission, ticket))
            # ⚠⚠ KORREKTUR EINER FEHLBUCHUNG (2026-08-12). Bis dahin galt strikt
            #    `exit_time IS NULL` — eine einmal geschriebene Zeile war endgueltig.
            #    Das machte einen Folgefehler dauerhaft: buchte der Fallback-Pfad
            #    eine Position faelschlich als geschlossen (real #49852362:
            #    +5,87 statt −95,38 € und auf dem falschen Tag), lief der SPAETERE,
            #    ECHTE Close still ins Leere.
            # ⚠ ENG GEFASST, damit keine gute Zeile ueberschrieben wird:
            #    korrigiert wird NUR, wenn die bestehende Zeile eine erkennbare
            #    Fehlbuchung ist (`closed_by='unknown'`) UND die neue Buchung
            #    echte Deal-Daten mitbringt (`closed_by != 'unknown'`).
            if cur.rowcount == 0 and closed_by != "unknown":
                cur = conn.execute("""
                    UPDATE trades
                       SET exit_time = ?, exit_price = ?, pnl = ?,
                           closed_by = ?, commission = ?
                     WHERE ticket = ? AND closed_by = 'unknown'
                """, (ts, exit_price, pnl, closed_by, commission, ticket))
                if cur.rowcount:
                    self._korrigiert = getattr(self, "_korrigiert", 0) + 1
            conn.commit()
            # KEIN Trade-Abschluss-Telegram mehr (User-Vorgabe): Telegram zum Thema
            # Schließen kommt nur noch beim echten Signal-Flip (engine._check_close_
            # alert „🔔 CLOSE-Signal") sowie bei Notfall-/Gewinn-Auto-Close. Die terse
            # „Sym +X"-Bestätigung bei jedem Close ist entfernt.

    # ── TELEMETRIE-PULS ──────────────────────────────────────────────────
    # ⚠⚠ ANLASS (2026-08-11): `rec_outcomes` stand seit dem Einbau am 06.08. bei
    #    0 Zeilen — ein falsches dict liess die Schreibbedingung nie wahr werden.
    #    KEIN NameError (der Linter sieht nichts), und das umgebende `except`
    #    loggt auf DEBUG. Fuenf Tage still tot; aufgefallen nur, weil die Tabelle
    #    neben ihren Schwestern mit 0 Zeilen stand.
    # ⚠ Warum hier und nicht im Reminder-Skript: der Puls ist eine DB-Frage, und
    #   BEIDE Nutzer (Reminder-Popup + Tagesreport) brauchen ihn. Zwei Kopien
    #   waeren genau die Divergenz, die im Projekt dreimal zugeschlagen hat.
    #   Richtungsregel: Skripte duerfen `core` importieren, nicht umgekehrt.
    PULS = (
        # (Tabelle, Zeitspalte, erwartete Kadenz in Minuten, wozu)
        ("recommendations",    "timestamp", 1,   "Empfehlungs-Log"),
        ("candles_m1",         "time",      1,   "M1-Kerzen-Sammlung"),
        ("verdict_votes",      "ts",        1,   "Modul-Stimmen"),
        ("m15_states",         "ts",        60,  "M15-Karte Selbstmessung"),
        ("cone_checks",        "ts",        60,  "Kegel-Kalibrierung (ACI)"),
        ("pbreak_predictions", "ts",        60,  "P(break)-Prognose-Tracking"),
        ("rec_outcomes",       "ts",        180, "Selbst-Kalibrierung Empfehlung"),
    )

    def telemetrie_puls(self) -> list[str]:
        """Welche Telemetrie-Tabelle schreibt nicht mehr? → Klartext-Zeilen.

        ⚠ Toleranz bewusst grosszuegig (Faktor 20 der Kadenz): ein ruhiger Markt
        oder ein Wochenende darf keinen Fehlalarm ausloesen — ein Waechter, der
        oft warnt, wird ignoriert (Lehre vom 02.08.).
        """
        out: list[str] = []
        try:
            jetzt = time.time()
            with self._lock, self._connect() as conn:
                for tab, sp, kadenz, wozu in self.PULS:
                    try:
                        n = conn.execute(f"SELECT COUNT(*) FROM {tab}").fetchone()[0]
                        letzt = conn.execute(f"SELECT MAX({sp}) FROM {tab}").fetchone()[0]
                    except Exception:
                        continue                 # Tabelle gibt es (noch) nicht
                    if not n:
                        out.append(f"{tab}: 0 Zeilen — Logger hat NIE geschrieben ({wozu})")
                        continue
                    # ⚠ `candles_m1.time` ist ROHE Brokerzeit (UTC+3) — ohne den
                    #   Abzug meldet der Puls dort dauerhaft 3 h Rueckstand.
                    ts = letzt - 10800 if tab == "candles_m1" else letzt
                    alter_min = (jetzt - ts) / 60.0
                    if alter_min > kadenz * 20:
                        out.append(f"{tab}: letzte Zeile vor {alter_min/60:.1f} h "
                                   f"(erwartet ~alle {kadenz} min) — {wozu}")
        except Exception:
            return out                            # fail-open wie der Rest der Schicht
        return out

    def trade_by_ticket(self, ticket: int) -> dict | None:
        """Einen Trade komplett lesen — für die Close-Benachrichtigung (2026-08-11).

        ⚠ Der Aufrufer muss `exit_time` prüfen: die Engine bemerkt den Flat-Zustand
        über `trader.snapshot()` und kann dem DB-Schreiber um einen Tick voraus sein.
        Ein Datensatz mit `exit_time IS NULL` heisst „noch nicht fertig", NICHT
        „kein Trade" — wer das verwechselt, meldet einen Close ohne P&L.
        """
        try:
            with self._lock, self._connect() as conn:
                row = conn.execute(
                    "SELECT * FROM trades WHERE ticket = ?", (int(ticket),)).fetchone()
                return dict(row) if row else None
        except Exception:
            return None            # fail-open wie der Rest dieser Schicht

    def log_ai(self, *, sentiment: str, confidence: int,
               summary: str, drivers: list,
               cost_estimate: float | None,
               model: str):
        with self._lock, self._connect() as conn:
            conn.execute("""
                INSERT INTO ai_analyses
                  (timestamp, sentiment, confidence, summary,
                   drivers_json, cost_estimate, model)
                VALUES (?,?,?,?,?,?,?)
            """, (int(time.time()), sentiment, confidence,
                  summary[:500], json.dumps(drivers, ensure_ascii=False),
                  cost_estimate, model))
            conn.commit()

    def log_recommendation(self, *, signal: str, score: float, conf_pct: int,
                           angles: dict, reversal: str | None,
                           ai_sentiment: str | None,
                           ai_confidence: int | None,
                           setup: str | None = None,
                           regime: str | None = None,
                           rsi: float | None = None,
                           news_score: float | None = None,
                           block: str | None = None):
        """
        Empfehlungs-Logging mit Throttling: nur jede N Sekunden, um die DB
        nicht mit identischen Snapshots zu fluten.
        """
        now = int(time.time())
        if now - self._last_rec_ts < self.rec_min_interval:
            return
        self._last_rec_ts = now
        with self._lock, self._connect() as conn:
            conn.execute("""
                INSERT INTO recommendations
                  (timestamp, signal, score, conf_pct,
                   angle_m5, angle_m15, angle_m30, angle_h1,
                   reversal, ai_sentiment, ai_confidence,
                   setup, regime, rsi, news_score, block_reason)
                VALUES (?,?,?,?, ?,?,?,?, ?,?,?, ?,?,?,?,?)
            """, (now, signal, score, conf_pct,
                  angles.get("M5"),  angles.get("M15"),
                  angles.get("M30"), angles.get("H1"),
                  reversal, ai_sentiment, ai_confidence,
                  setup, regime, rsi, news_score, block))
            conn.commit()

    def log_signal(self, *, signal_type: str, direction: str,
                   price: float | None = None, details: dict | None = None):
        with self._lock, self._connect() as conn:
            conn.execute("""
                INSERT INTO signals
                  (timestamp, signal_type, direction, price, details_json)
                VALUES (?,?,?,?,?)
            """, (int(time.time()), signal_type, direction, price,
                  json.dumps(details or {}, ensure_ascii=False)))
            conn.commit()

    def log_pbreak_prediction(self, *, symbol: str, direction: str, level: float,
                              p_break: float, predicted: str, price_at_pred: float,
                              confirm_price: float, reject_price: float,
                              atr: float) -> int:
        """Loggt EINE Abprall/Durchbruch-Vorhersage bei frischem Level-Touch
        (`predicted` = 'break'|'bounce', aus dem live-P(break) vs. `sr_close_pbreak`).
        Auswertung passiert separat (`engine._evaluate_pbreak_predictions`, gegen
        `candles_m1`). Rückgabe = Zeilen-ID."""
        with self._lock, self._connect() as conn:
            cur = conn.execute("""
                INSERT INTO pbreak_predictions
                  (ts, symbol, direction, level, p_break, predicted, price_at_pred,
                   confirm_price, reject_price, atr)
                VALUES (?,?,?,?,?,?,?,?,?,?)
            """, (int(time.time()), symbol, direction, level, p_break, predicted,
                  price_at_pred, confirm_price, reject_price, atr))
            conn.commit()
            return cur.lastrowid

    def log_rec_outcome(self, *, signal: str, conf_pct: int, tf: str,
                        price: float, atr: float) -> int:
        """Loggt EINE Empfehlungs-EPISODE zur spaeteren Selbst-Kalibrierung.

        ⚠ WARUM EPISODEN UND NICHT ZEILEN: die Empfehlung wird ~1x/Minute
        geloggt; eine Marktlage erzeugt dutzende identische Zeilen. Wer die zaehlt,
        haelt 20.000 Faelle fuer eine Stichprobe, obwohl es 1.000 sind
        (`analyze_signal_live.py`). Hier wird genau EINE Zeile je Richtungswechsel
        geschrieben.

        Ausgewertet wird separat (`engine._evaluate_rec_outcomes`) gegen
        `candles_m1` — dasselbe Muster wie `pbreak_predictions`, das den Ausfall
        des P(break)-Modells live aufgedeckt hat. Rueckgabe = Zeilen-ID."""
        with self._lock, self._connect() as conn:
            cur = conn.execute("""
                INSERT INTO rec_outcomes (ts, signal, conf_pct, tf, price, atr)
                VALUES (?,?,?,?,?,?)
            """, (int(time.time()), signal, int(conf_pct or 0), tf or "",
                  float(price or 0.0), float(atr or 0.0)))
            conn.commit()
            return cur.lastrowid

    def log_cone_check(self, *, bars: int, band: int, price: float,
                       lo: float, hi: float, skala: float) -> int:
        """Legt EINEN Abdeckungs-Check des Kegels an (nicht-ueberlappend)."""
        try:
            with self._lock, self._connect() as conn:
                cur = conn.execute("""
                    INSERT INTO cone_checks (ts,bars,band,price,lo,hi,skala)
                    VALUES (?,?,?,?,?,?,?)
                """, (int(time.time()), int(bars), int(band), float(price),
                      float(lo), float(hi), float(skala)))
                conn.commit()
                return cur.lastrowid
        except Exception:
            return 0

    def cone_coverage(self, bars: int, last_n: int = 100) -> dict:
        """Real gemessene Abdeckung der letzten `last_n` Checks eines Horizonts."""
        try:
            with self._lock, self._connect() as conn:
                rows = conn.execute(
                    "SELECT drin FROM cone_checks WHERE bars=? AND done=1 "
                    "ORDER BY id DESC LIMIT ?", (int(bars), int(last_n))).fetchall()
        except Exception:
            return {"n": 0}
        n = len(rows)
        if not n:
            return {"n": 0}
        return {"n": n, "abdeckung": round(100 * sum(r[0] for r in rows) / n, 1)}

    def log_m15_state(self, *, status: str, veto, kosten, level, gap_atr,
                      p_break, lesart, price, atr) -> int:
        """Loggt EINEN Zustandswechsel der M15-Karte.

        ⚠ Nur bei WECHSEL aufrufen, nicht je Tick — sonst entsteht dieselbe
        Cluster-Verzerrung, die am 07.08. eine Untersuchung mit elf widerlegten
        Hypothesen ausgeloest hat (roh 922 Zeilen mit 52,6 %, entkoppelt 93 mit
        45,2 %). Ausgewertet wird separat gegen `candles_m1`.
        ⚠ Fail-open wie der Rest dieser Klasse: ein Logging-Fehler darf den
        Handelspfad nie stoeren."""
        try:
            with self._lock, self._connect() as conn:
                cur = conn.execute("""
                    INSERT INTO m15_states
                      (ts,status,veto,kosten,level,gap_atr,p_break,lesart,price,atr)
                    VALUES (?,?,?,?,?,?,?,?,?,?)
                """, (int(time.time()), status or "", veto,
                      None if kosten is None else float(kosten),
                      None if level is None else float(level),
                      None if gap_atr is None else float(gap_atr),
                      None if p_break is None else int(p_break),
                      lesart, float(price or 0.0), float(atr or 0.0)))
                conn.commit()
                return cur.lastrowid
        except Exception:
            return 0

    def m15_accuracy(self, last_n: int = 200) -> dict:
        """Trefferquote der LESART (nicht des Status) ueber die letzten
        ausgewerteten Zustaende mit gerichteter Lesart.

        ⚠ Gegen 50 % zu lesen, nicht gegen 0. Und ⚠ die Karte behauptet KEINE
        freie Richtung — gemessen wird nur, ob die BEDINGTE Lesart am Level
        („Abprall -> eher abwaerts") getroffen hat.

        ⚠⚠ ENTKOPPELT seit 2026-08-12. Vorher nahm die Abfrage schlicht die
        letzten N Zeilen (`ORDER BY id DESC LIMIT ?`). Die sind NICHT unabhaengig:
        der Zustand wechselt ~990x/Tag, 200 Zeilen sind also rund fuenf Stunden
        DERSELBEN Marktlage, vielfach gezaehlt. Die Karte zeigte dadurch **60 %**,
        entkoppelt sind es **54,4 %** mit dem Intervall [46 … 63] — also nicht von
        der Muenze zu unterscheiden.
        ⚠ Dieselbe Cluster-Falle hat im Projekt schon die Faelligkeitsbedingung
        `pbreak_accuracy_v2` um das ZEHNFACHE danebenliegen lassen (07.08.) und
        elf Hypothesen ausgeloest (07.08.). Sie ist hier dieselbe.
        ⚠ Schluessel = (Level auf 2 Stellen, Stunde) — identisch zu
        `core.stichprobe.schluessel_level_stunde`, damit beide Auswertungen
        dieselbe Definition benutzen.
        """
        try:
            with self._lock, self._connect() as conn:
                # ⚠ Grosszuegig lesen und DANACH entkoppeln — sonst waeren nach der
                #   Entkopplung nur noch eine Handvoll Faelle uebrig.
                raw = conn.execute("""
                    SELECT hit30, hit60, ret30, level, ts FROM m15_states
                    WHERE done=1 AND lesart IN ('auf','ab')
                    ORDER BY id DESC LIMIT ?
                """, (int(last_n) * 40,)).fetchall()
                offen = conn.execute(
                    "SELECT COUNT(*) FROM m15_states WHERE done IS NULL "
                    "AND lesart IN ('auf','ab')").fetchone()[0]
        except Exception:
            return {"n": 0}
        seen, rows = set(), []
        for r in raw:
            k = (round(r[3] or 0.0, 2), int((r[4] or 0) // 3600))
            if k in seen:
                continue
            seen.add(k)
            rows.append(r)
            if len(rows) >= int(last_n):
                break
        n = len(rows)
        if not n:
            return {"n": 0, "offen": offen, "roh": len(raw)}
        h30 = [r[0] for r in rows if r[0] is not None]
        h60 = [r[1] for r in rows if r[1] is not None]
        r30 = [r[2] for r in rows if r[2] is not None]
        return {"n": n, "offen": offen, "roh": len(raw),
                "wr30": round(100 * sum(h30) / len(h30)) if h30 else None,
                "wr60": round(100 * sum(h60) / len(h60)) if h60 else None,
                "avg30": round(sum(r30) / len(r30), 3) if r30 else None}

    def rec_accuracy(self, last_n: int = 200) -> dict:
        """Trefferquote der letzten `last_n` ausgewerteten Empfehlungs-Episoden.

        ⚠ Die Trefferquote ist gegen 50 % zu lesen, nicht gegen 0 — und der
        Ø-Return gegen die Drift des Zeitraums. Gemessen ueber 1.071 Episoden lag
        die Richtungstreffer-Quote bei 48–51 % (`analyze_signal_live.py`); diese
        Zeile macht laufend sichtbar, ob sich daran etwas aendert."""
        with self._connect() as conn:
            rows = conn.execute("""
                SELECT signal, ret30, ret60, hit30, hit60 FROM rec_outcomes
                 WHERE done = 1 ORDER BY ts DESC LIMIT ?
            """, (int(last_n),)).fetchall()
        if not rows:
            return {"n": 0}
        def q(feld, hit):
            v = [r[feld] for r in rows if r[feld] is not None]
            h = [r[hit] for r in rows if r[hit] is not None]
            return ((round(100.0 * sum(h) / len(h), 1) if h else None),
                    (round(sum(v) / len(v), 3) if v else None), len(h))
        wr30, avg30, n30 = q("ret30", "hit30")
        wr60, avg60, n60 = q("ret60", "hit60")
        return {"n": len(rows), "wr30": wr30, "avg30": avg30, "n30": n30,
                "wr60": wr60, "avg60": avg60, "n60": n60}

    def pbreak_accuracy(self, period: str = "all") -> dict:
        """Trefferquote der Abprall/Durchbruch-Prognose (nur ausgewertete Zeilen)."""
        since, until = self._range_to_ts(period)
        with self._connect() as conn:
            rows = conn.execute("""
                SELECT predicted, outcome, correct FROM pbreak_predictions
                 WHERE outcome IS NOT NULL AND ts BETWEEN ? AND ?
            """, (since, until)).fetchall()
            pending = conn.execute(
                "SELECT COUNT(*) FROM pbreak_predictions WHERE outcome IS NULL"
            ).fetchone()[0]
        n = len(rows)
        n_correct = sum(1 for r in rows if r["correct"])
        n_break_pred   = sum(1 for r in rows if r["predicted"] == "break")
        n_bounce_pred  = sum(1 for r in rows if r["predicted"] == "bounce")
        n_break_correct  = sum(1 for r in rows if r["predicted"] == "break"  and r["correct"])
        n_bounce_correct = sum(1 for r in rows if r["predicted"] == "bounce" and r["correct"])
        return {
            "n": n, "n_correct": n_correct,
            "accuracy": round(100 * n_correct / n, 1) if n else None,
            "n_break_pred": n_break_pred, "n_break_correct": n_break_correct,
            "break_accuracy": round(100 * n_break_correct / n_break_pred, 1) if n_break_pred else None,
            "n_bounce_pred": n_bounce_pred, "n_bounce_correct": n_bounce_correct,
            "bounce_accuracy": round(100 * n_bounce_correct / n_bounce_pred, 1) if n_bounce_pred else None,
            "pending": pending,
        }

    def log_verdict_votes(self, *, headline: str, conf: int, bias: float, tf: str,
                          wave: int, m30: int, h1: int, ki: int, ki_w: float,
                          elliott: int, elliott_w: float, squeeze: int,
                          news_score=None, pattern: int = 0, pattern_w: float = 0.0,
                          orderbook: int = 0, orderbook_w: float = 0.0,
                          liqtrend: int = 0, liqtrend_w: float = 0.0) -> None:
        """Verdict-Modul-Stimmen für die spätere Kalibrier-Auswertung loggen
        (Copilot/Elliott sind unbelegte Stimmen — nach ein paar Wochen kann
        `analyze_verdict_calibration.py`-artig entschieden werden, ob sie
        Gewicht behalten). Intern gedrosselt auf max. 1×/60 s."""
        now = time.time()
        if now - getattr(self, "_verdict_log_ts", 0.0) < 60:
            return
        self._verdict_log_ts = now
        with self._lock, self._connect() as conn:
            conn.execute("""
                INSERT INTO verdict_votes
                  (ts, headline, conf, bias, tf, wave, m30, h1, ki, ki_w,
                   elliott, elliott_w, squeeze, news_score,
                   pattern, pattern_w, orderbook, orderbook_w, liqtrend, liqtrend_w)
                VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
            """, (int(now), headline, conf, bias, tf, wave, m30, h1, ki, ki_w,
                  elliott, elliott_w, squeeze, news_score,
                  pattern, pattern_w, orderbook, orderbook_w, liqtrend, liqtrend_w))
            conn.commit()

    def manual_stats(self, tage: int = 30) -> dict:
        """Kennzahlen NUR der manuellen Trades — Datengrundlage der Dashboard-Karte
        „Manuelle Trades" (2026-08-04).

        ⚠ Bewusst DESKRIPTIV, nicht prädiktiv. Der volle Entscheidungszustand wird
        erst seit dem 04.08. mitgeschrieben (`ctx_*`); bis genug davon vorliegt,
        wäre jede Vorhersage aus diesen Zahlen Rauschen. Die Karte zeigt deshalb,
        was die eigene Historie SAGT, und rechnet nichts hoch.
        ⚠ `ctx_abdeckung` macht sichtbar, wie weit der neue Kontext schon reicht —
        ohne diese Zahl könnte man die Karte für aussagekräftiger halten, als sie ist.
        """
        MAN = ("(setup IS NULL OR (setup NOT LIKE 'AUTOSIG%' "
               "AND setup NOT LIKE 'SQUEEZE%'))")

        def agg(row):
            n = row["n"] or 0
            return {"n": n, "wr": round(100.0 * (row["w"] or 0) / n, 1) if n else 0.0,
                    "sum": round(row["s"] or 0.0, 2),
                    "avg": round((row["s"] or 0.0) / n, 2) if n else 0.0}

        out = {"tage": tage}
        with self._connect() as conn:
            conn.row_factory = sqlite3.Row
            q = (f"SELECT COUNT(*) n, SUM(CASE WHEN pnl>0 THEN 1 ELSE 0 END) w, "
                 f"SUM(pnl) s FROM trades WHERE pnl IS NOT NULL AND {MAN}")
            seit = f"AND exit_time >= strftime('%s','now','-{int(tage)} days')"
            out["fenster"] = agg(conn.execute(f"{q} {seit}").fetchone())
            out["gesamt"] = agg(conn.execute(q).fetchone())
            out["heute"] = agg(conn.execute(
                f"{q} AND exit_time >= strftime('%s','now','start of day')").fetchone())
            # Mit / ohne Signal-Deckung — die dokumentierte Kern-Frage
            for lbl, cond in (("mit_signal", "rec_signal IN ('LONG','SHORT')"),
                              ("ohne_signal", "rec_signal='WARTEN'")):
                out[lbl] = agg(conn.execute(f"{q} {seit} AND {cond}").fetchone())
            # Bot zum Vergleich (dieselbe Fensterlogik)
            qb = ("SELECT COUNT(*) n, SUM(CASE WHEN pnl>0 THEN 1 ELSE 0 END) w, "
                  "SUM(pnl) s FROM trades WHERE pnl IS NOT NULL AND "
                  "(setup LIKE 'AUTOSIG%' OR setup LIKE 'SQUEEZE%')")
            out["bot"] = agg(conn.execute(f"{qb} {seit}").fetchone())
            # Abdeckung des NEUEN Kontexts — ehrlicher Reifegrad-Indikator
            r = conn.execute(
                f"SELECT COUNT(*) n, SUM(CASE WHEN ctx_atr_m5 IS NOT NULL THEN 1 ELSE 0 END) c "
                f"FROM trades WHERE {MAN}").fetchone()
            out["ctx_abdeckung"] = {
                "n": r["n"] or 0, "mit_ctx": r["c"] or 0,
                "pct": round(100.0 * (r["c"] or 0) / (r["n"] or 1), 1)}
        return out

    def alignment_stats(self, last_n: int = 0) -> dict:
        """Ausrichtungs-Split: liefen die Trades MIT, GEGEN oder OHNE die Empfehlung?

        ⚠⚠ ZWEI AENDERUNGEN AM 2026-08-06, beide gemessen begruendet:
        (1) **Volle Historie statt der letzten 40 Trades** (`last_n=0`). Ueber
            1.216 geschlossene Trades ist der Befund beidhaelftig robust; ueber 40
            war er eine Momentaufnahme. Das ist die einzige Richtungsaussage ueber
            die Gesamtempfehlung, die den 2-Stichproben-Test besteht — sie sollte
            auf der ganzen Stichprobe stehen.
        (2) **Zusaetzlich JE LOT normiert** (`pnl_lot`). Absolute Euro-Vergleiche
            ueber Zeitraeume mit unterschiedlicher Positionsgroesse sind ungueltig
            — am 04.08. sah ein reiner Groessenzuwachs wie ein Strategie-Effekt aus
            (Ø +55 € gegen +19 €, je Lot aber identisch). Das Order-Dialog-Feld
            zeigt deshalb beides.

        Gemessener Stand (06.08., ganze Historie, je Lot):
            MIT   n=354  WR 56,5 %  +3,22 €/Lot   (H1 +0,37 / H2 +5,62)
            GEGEN n= 87  WR 40,2 %  −6,60 €/Lot   (H1 −7,01 / H2 −5,68)
            OHNE  n=775  WR 56,6 %  +0,36 €/Lot   (kippt zwischen den Haelften)
        MIT und OHNE sind praktisch gleich — der Empfehlung zu FOLGEN bringt
        gemessen nichts, nur sie zu VERLETZEN kostet. Sie ist ein Veto, keine
        Prognose.
        """
        sql = ("SELECT direction, rec_signal, pnl, lots FROM trades "
               "WHERE exit_time IS NOT NULL AND pnl IS NOT NULL AND lots > 0 "
               "ORDER BY exit_time DESC")
        args: tuple = ()
        if last_n and last_n > 0:
            sql += " LIMIT ?"
            args = (last_n,)
        with self._connect() as conn:
            rows = conn.execute(sql, args).fetchall()

        def fasse(grp: list) -> dict:
            n = len(grp)
            if not n:
                return {"n": 0, "wr": None, "pnl": 0.0, "pnl_lot": None}
            pnl = sum(r["pnl"] for r in grp)
            lots = sum(r["lots"] for r in grp) or 0.0
            return {"n": n,
                    "wr": round(100 * sum(1 for r in grp if r["pnl"] > 0) / n),
                    "pnl": round(pnl, 2),
                    "pnl_lot": round(pnl / lots, 2) if lots > 0 else None}

        mit, gegen, ohne = [], [], []
        for r in rows:
            rec = (r["rec_signal"] or "").upper()
            if rec not in ("LONG", "SHORT"):
                ohne.append(r)
            elif (rec == "LONG") == (r["direction"] == "BUY"):
                mit.append(r)
            else:
                gegen.append(r)
        return {"with": fasse(mit), "against": fasse(gegen), "none": fasse(ohne)}

    # ══════════════════════════════════════════
    #  STATISTIK-QUERIES
    # ══════════════════════════════════════════

    @staticmethod
    def _range_to_ts(period: str) -> tuple[int, int]:
        """'today' | 'week' | 'month' | 'all' → (since_ts, now_ts)."""
        now = int(time.time())
        if period == "today":
            today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
            return (int(today.timestamp()), now)
        if period == "yesterday":
            today = datetime.now().replace(hour=0, minute=0, second=0, microsecond=0)
            return (int((today - timedelta(days=1)).timestamp()), int(today.timestamp()))
        if period == "week":
            since = datetime.now() - timedelta(days=7)
            return (int(since.timestamp()), now)
        if period == "month":
            since = datetime.now() - timedelta(days=30)
            return (int(since.timestamp()), now)
        return (0, now)

    def tag_bot_trade(self, ticket: int, setup: str) -> bool:
        """Setup einer bereits geloggten Position NACHTRAGEN. → True bei Erfolg.

        ⚠ WARUM DAS NOETIG IST: eine gefuellte PENDING-Order laeuft nicht durch
        `engine._open`, sondern wird vom Positions-Abgleich als „magic-match"
        adoptiert und mit `log_trade_open(...)` OHNE Setup geschrieben. Am 06.08.
        fuellten drei `SQZ-STOP`-Orders und landeten dadurch als `setup=NULL` in der
        DB — ununterscheidbar von manuellen Trades. Folge: B4-Monitor und die
        Reminder `squeeze_b5` / `squeeze_entry_gap` zaehlten 0, obwohl der Bot
        gehandelt hatte. Genau die Messpipeline, die den Pending-Umbau
        kontrollieren soll, war blind fuer ihn.
        Nur setzen, wenn noch KEIN Setup steht — ein manuell/regulaer getaggter
        Trade darf nicht ueberschrieben werden."""
        try:
            with self._connect() as conn:
                cur = conn.execute(
                    "UPDATE trades SET setup=? WHERE ticket=? AND setup IS NULL",
                    (setup, int(ticket)))
                return cur.rowcount > 0
        except Exception:
            # ⚠ `core/history.py` fuehrt bewusst keinen Logger (Konvention der
            # Datei: stilles Abfangen). Ein fehlgeschlagenes Nachtragen ist
            # unkritisch — der Trade steht in der DB, nur ohne Setup-Tag; der
            # Aufrufer sieht das am Rueckgabewert.
            return False

    def stats_overview(self, period: str = "all") -> dict:
        """Liefert ein Dict mit den Hauptkennzahlen für den gewählten Zeitraum."""
        since, until = self._range_to_ts(period)
        with self._connect() as conn:
            # Trades (nur abgeschlossene)
            rows = conn.execute("""
                SELECT direction, lots, entry_time, exit_time, entry_price,
                       exit_price, pnl, closed_by, ai_sentiment, ai_confidence,
                       rec_signal
                  FROM trades
                 WHERE exit_time IS NOT NULL
                   AND exit_time BETWEEN ? AND ?
                 ORDER BY exit_time
                 LIMIT 10000
            """, (since, until)).fetchall()

        n         = len(rows)
        wins      = [r for r in rows if (r["pnl"] or 0) >  0]
        losses    = [r for r in rows if (r["pnl"] or 0) <  0]
        breakeven = [r for r in rows if (r["pnl"] or 0) == 0]

        total_pnl = sum((r["pnl"] or 0) for r in rows)
        avg_win   = (sum(r["pnl"] for r in wins)   / len(wins))   if wins   else 0.0
        avg_loss  = (sum(r["pnl"] for r in losses) / len(losses)) if losses else 0.0
        gross_win  = sum(r["pnl"]      for r in wins)
        gross_loss = sum(-r["pnl"]     for r in losses)
        profit_factor = (gross_win / gross_loss) if gross_loss > 0 else None

        # Closed-by Verteilung
        cb_counts = {}
        for r in rows:
            cb = r["closed_by"] or "unknown"
            cb_counts[cb] = cb_counts.get(cb, 0) + 1

        # KI-Trefferquote
        ai_correct = ai_total = 0
        for r in rows:
            sent = (r["ai_sentiment"] or "").lower()
            pnl  = r["pnl"] or 0
            direction = (r["direction"] or "").upper()
            if sent in ("bullish", "bearish") and pnl != 0:
                ai_total += 1
                # KI bullish + Long-Trade gewonnen, oder bearish + Short gewonnen
                ai_aligned = (
                    (sent == "bullish" and direction == "BUY"  and pnl > 0) or
                    (sent == "bearish" and direction == "SELL" and pnl > 0) or
                    (sent == "bullish" and direction == "SELL" and pnl < 0) or
                    (sent == "bearish" and direction == "BUY"  and pnl < 0)
                )
                if ai_aligned:
                    ai_correct += 1

        # Wochentag-Verteilung
        weekday_stats = {i: {"wins": 0, "losses": 0} for i in range(7)}
        for r in rows:
            wd = datetime.fromtimestamp(r["exit_time"]).weekday()
            if (r["pnl"] or 0) > 0:
                weekday_stats[wd]["wins"]   += 1
            elif (r["pnl"] or 0) < 0:
                weekday_stats[wd]["losses"] += 1

        # Stunden-Verteilung
        hour_stats = {h: {"wins": 0, "losses": 0} for h in range(24)}
        for r in rows:
            hr = datetime.fromtimestamp(r["exit_time"]).hour
            if (r["pnl"] or 0) > 0:
                hour_stats[hr]["wins"]   += 1
            elif (r["pnl"] or 0) < 0:
                hour_stats[hr]["losses"] += 1

        # Top-3 Stunden (mind. 2 Trades)
        top_hours = sorted(
            [(h, s["wins"], s["losses"]) for h, s in hour_stats.items()
             if (s["wins"] + s["losses"]) >= 2],
            key=lambda x: (x[1] / max(1, x[1]+x[2]), x[1]+x[2]),
            reverse=True)[:3]

        return {
            "period":         period,
            "since":          since,
            "until":          until,
            "n_trades":       n,
            "n_wins":         len(wins),
            "n_losses":       len(losses),
            "n_be":           len(breakeven),
            "winrate":        (len(wins) / n * 100) if n else 0.0,
            "total_pnl":      total_pnl,
            "avg_win":        avg_win,
            "avg_loss":       avg_loss,
            "gross_win":      gross_win,
            "gross_loss":     gross_loss,
            "profit_factor":  profit_factor,
            "closed_by":      cb_counts,
            "ai_correct":     ai_correct,
            "ai_total":       ai_total,
            "ai_winrate":     (ai_correct / ai_total * 100) if ai_total else None,
            "weekday_stats":  weekday_stats,
            "hour_stats":     hour_stats,
            "top_hours":      top_hours,
        }

    def setup_stats(self, period: str = "all") -> list[dict]:
        """
        Performance-Aufschlüsselung pro Setup-Typ.
        Liefert eine Liste {setup, n, wins, losses, winrate, total_pnl, avg_pnl, profit_factor}
        sortiert nach total_pnl absteigend.
        """
        since, until = self._range_to_ts(period)
        with self._connect() as conn:
            rows = conn.execute("""
                SELECT setup, pnl
                  FROM trades
                 WHERE exit_time IS NOT NULL
                   AND exit_time BETWEEN ? AND ?
                   AND setup IS NOT NULL
            """, (since, until)).fetchall()

        groups: dict[str, list[float]] = {}
        for r in rows:
            groups.setdefault(r["setup"] or "UNKNOWN", []).append(r["pnl"] or 0.0)

        out = []
        for setup, pnls in groups.items():
            wins   = [p for p in pnls if p > 0]
            losses = [p for p in pnls if p < 0]
            gross_w = sum(wins)
            gross_l = sum(-p for p in losses)
            out.append({
                "setup":         setup,
                "n":             len(pnls),
                "wins":          len(wins),
                "losses":        len(losses),
                "winrate":       (len(wins) / len(pnls) * 100) if pnls else 0.0,
                "total_pnl":     sum(pnls),
                "avg_pnl":       sum(pnls) / len(pnls) if pnls else 0.0,
                "gross_win":     gross_w,
                "gross_loss":    gross_l,
                "profit_factor": (gross_w / gross_l) if gross_l > 0 else None,
            })
        out.sort(key=lambda x: x["total_pnl"], reverse=True)
        return out

    def setup_multipliers(self, *, min_trades: int = 10,
                          lookback_days: int = 90) -> dict[str, dict]:
        """
        Lernt aus historischen Trades einen Confidence-Multiplikator pro Setup.

        Mapping:
          Profit-Factor 1.0 (break-even)       → 1.0  (kein Effekt)
          Profit-Factor ≥ 2.0 (sehr profitabel)→ 1.3  (Score +30 %)
          Profit-Factor ≤ 0.5 (klar verlierend)→ 0.5  (Score −50 %)
          Lineare Interpolation dazwischen.

        Setups unter `min_trades` bekommen Multiplikator 1.0 (Neutral, noch zu
        wenig Daten). Lookback begrenzt auf `lookback_days` Tage, damit
        Regime-Wechsel berücksichtigt werden.

        Rückgabe: {setup_name: {"multiplier": float, "n": int,
                                "profit_factor": float|None, "winrate": float}}
        """
        cutoff = int(time.time()) - lookback_days * 86400
        with self._connect() as conn:
            rows = conn.execute("""
                SELECT setup, pnl
                  FROM trades
                 WHERE exit_time IS NOT NULL
                   AND exit_time >= ?
                   AND setup IS NOT NULL
            """, (cutoff,)).fetchall()

        groups: dict[str, list[float]] = {}
        for r in rows:
            groups.setdefault(r["setup"], []).append(r["pnl"] or 0.0)

        out: dict[str, dict] = {}
        for setup, pnls in groups.items():
            n = len(pnls)
            wins   = [p for p in pnls if p > 0]
            losses = [p for p in pnls if p < 0]
            gross_w = sum(wins)
            gross_l = sum(-p for p in losses)
            pf = (gross_w / gross_l) if gross_l > 0 else (
                 float("inf") if gross_w > 0 else 1.0)
            wr = (len(wins) / n * 100) if n else 0.0

            if n < min_trades:
                mult = 1.0
            else:
                # Mapping pf → multiplier (geclampt)
                pf_capped = min(max(pf, 0.3), 3.0)
                if pf_capped >= 1.0:
                    # 1.0 → 1.0,  2.0 → 1.3,  3.0 → 1.3 (gecapped)
                    mult = 1.0 + min(0.30, (pf_capped - 1.0) * 0.30)
                else:
                    # 1.0 → 1.0,  0.5 → 0.5,  0.3 → 0.5 (gecapped)
                    mult = max(0.50, 1.0 - (1.0 - pf_capped) * 1.0)
            out[setup] = {
                "multiplier":    round(mult, 3),
                "n":             n,
                "profit_factor": pf if pf != float("inf") else None,
                "winrate":       wr,
            }
        return out

    # ══════════════════════════════════════════
    #  AUTO-TRADE DRY-RUN
    # ══════════════════════════════════════════
    def intended_open(self, *, direction: str, open_price: float,
                      sl: float, tp: float,
                      setup: str, regime: str | None,
                      rsi: float | None, score: float, conf_pct: int,
                      news_score: float | None,
                      ai_sentiment: str | None,
                      ai_confidence: int | None) -> int:
        """Loggt einen hypothetischen Auto-Trade. Liefert die neue Row-ID."""
        with self._lock, self._connect() as conn:
            cur = conn.execute("""
                INSERT INTO intended_trades
                  (open_ts, open_price, direction, sl, tp,
                   setup, regime, rsi, score, conf_pct,
                   news_score, ai_sentiment, ai_confidence)
                VALUES (?,?,?,?,?, ?,?,?,?,?, ?,?,?)
            """, (int(time.time()), open_price, direction, sl, tp,
                  setup, regime, rsi, score, conf_pct,
                  news_score, ai_sentiment, ai_confidence))
            conn.commit()
            return cur.lastrowid

    def intended_close(self, *, row_id: int, close_price: float,
                       close_reason: str):
        """Schließt einen hypothetischen Trade. Berechnet pnl_pct vorzeichenrichtig."""
        with self._lock, self._connect() as conn:
            row = conn.execute(
                "SELECT open_price, direction FROM intended_trades "
                "WHERE id = ? AND close_ts IS NULL", (row_id,)
            ).fetchone()
            if not row:
                return  # bereits geschlossen oder nicht vorhanden
            op = row["open_price"] or 0.0
            if op <= 0:
                pnl_pct = 0.0
            else:
                diff = (close_price - op) if row["direction"] == "BUY" \
                       else (op - close_price)
                pnl_pct = diff / op * 100.0
            conn.execute("""
                UPDATE intended_trades
                   SET close_ts = ?, close_price = ?,
                       close_reason = ?, pnl_pct = ?
                 WHERE id = ?
            """, (int(time.time()), close_price, close_reason, pnl_pct, row_id))
            conn.commit()

    def intended_open_list(self) -> list[dict]:
        """Liefert alle noch nicht geschlossenen intended trades."""
        with self._connect() as conn:
            rows = conn.execute("""
                SELECT * FROM intended_trades WHERE close_ts IS NULL
                 ORDER BY open_ts DESC
            """).fetchall()
        return [dict(r) for r in rows]

    def intended_today_count(self) -> int:
        """Anzahl heute geöffneter hypothetischer Trades (für Daily-Cap)."""
        from datetime import datetime as _dt
        today = _dt.now().replace(hour=0, minute=0, second=0, microsecond=0)
        since = int(today.timestamp())
        with self._connect() as conn:
            return conn.execute(
                "SELECT COUNT(*) FROM intended_trades WHERE open_ts >= ?",
                (since,)
            ).fetchone()[0]

    def intended_last_close_ts(self) -> int:
        """Timestamp des letzten geschlossenen intended trade (für Cooldown).
        Flip-Closes zählen nicht — nach einem Signal-Flip darf die neue
        Richtung sofort eröffnet werden (schnelle Breakout-Reaktion)."""
        with self._connect() as conn:
            row = conn.execute(
                "SELECT MAX(close_ts) FROM intended_trades "
                "WHERE close_ts IS NOT NULL "
                "  AND COALESCE(close_reason, '') != 'flip'"
            ).fetchone()
        return int(row[0] or 0)

    def intended_summary(self, period: str = "all") -> dict:
        """Performance-Zusammenfassung aller hypothetischen Trades."""
        since, until = self._range_to_ts(period)
        with self._connect() as conn:
            rows = conn.execute("""
                SELECT pnl_pct, close_reason, setup FROM intended_trades
                 WHERE close_ts IS NOT NULL
                   AND close_ts BETWEEN ? AND ?
            """, (since, until)).fetchall()
        n = len(rows)
        wins = sum(1 for r in rows if (r["pnl_pct"] or 0) > 0)
        losses = sum(1 for r in rows if (r["pnl_pct"] or 0) < 0)
        total_pnl_pct = sum(r["pnl_pct"] or 0 for r in rows)
        avg_pnl_pct = total_pnl_pct / n if n else 0.0
        by_reason: dict[str, int] = {}
        for r in rows:
            by_reason[r["close_reason"] or "?"] = by_reason.get(r["close_reason"] or "?", 0) + 1
        return {
            "n":             n,
            "wins":          wins,
            "losses":        losses,
            "winrate":       (wins / n * 100) if n else 0.0,
            "total_pnl_pct": total_pnl_pct,
            "avg_pnl_pct":   avg_pnl_pct,
            "by_reason":     by_reason,
        }

    def last_closed_trades(self, n: int = 5) -> list[dict]:
        """Liefert die letzten n geschlossenen Trades (neueste zuerst)."""
        with self._connect() as conn:
            rows = conn.execute("""
                SELECT direction, lots, entry_time, exit_time,
                       entry_price, exit_price, pnl, commission, closed_by, setup
                  FROM trades
                 WHERE exit_time IS NOT NULL
                 ORDER BY exit_time DESC
                 LIMIT ?
            """, (n,)).fetchall()
        return [dict(r) for r in rows]

    def open_trades(self) -> list[dict]:
        """Liefert alle Trades, für die noch kein exit_time eingetragen ist."""
        with self._connect() as conn:
            rows = conn.execute("""
                SELECT id, ticket, symbol, direction, entry_time, entry_price
                  FROM trades
                 WHERE exit_time IS NULL
                 ORDER BY entry_time
            """).fetchall()
        return [dict(r) for r in rows]

    def consistency_report(self) -> dict:
        """
        Diagnostiziert Inkonsistenzen in der History-DB.
        Liefert ein Dict mit erkannten Problemen:

          • n_open_trades_db       — Anzahl "offene" Trades in DB
          • n_orphan_old           — offene Trades älter als 14 Tage (verwaist)
          • n_negative_pnl_no_loss — geschlossene Trades mit pnl < 0 aber wins+losses=0
          • n_missing_setup        — geschlossene Trades ohne setup-Spalte
          • n_missing_exit_price   — geschlossene Trades ohne exit_price
          • duplicate_tickets      — Tickets, die mehrfach existieren (sollte 0 sein)
          • db_size_mb
          • oldest_trade_age_days
        """
        now = int(time.time())
        out = {
            "timestamp":             now,
            "n_open_trades_db":      0,
            "n_orphan_old":          0,
            "n_missing_setup":       0,
            "n_missing_exit_price":  0,
            "n_pnl_zero_closed":     0,
            "duplicate_tickets":     [],
            "db_size_mb":            self.db_size_mb(),
            "oldest_trade_age_days": None,
        }
        ORPHAN_AGE_S = 14 * 86400

        with self._connect() as conn:
            # Open trades
            opens = conn.execute(
                "SELECT ticket, entry_time FROM trades WHERE exit_time IS NULL"
            ).fetchall()
            out["n_open_trades_db"] = len(opens)
            out["n_orphan_old"] = sum(
                1 for r in opens if (r["entry_time"] or now) < now - ORPHAN_AGE_S
            )

            # Geschlossene Trades ohne Setup-Spalte (alte Daten vor Migration)
            out["n_missing_setup"] = conn.execute("""
                SELECT COUNT(*) FROM trades
                 WHERE exit_time IS NOT NULL AND (setup IS NULL OR setup = '')
            """).fetchone()[0]

            # Geschlossene Trades ohne exit_price
            out["n_missing_exit_price"] = conn.execute("""
                SELECT COUNT(*) FROM trades
                 WHERE exit_time IS NOT NULL
                   AND (exit_price IS NULL OR exit_price = 0)
            """).fetchone()[0]

            # Geschlossene Trades mit pnl = 0 — aber NUR die, die NICHT
            # bewusst als 'unknown' markiert sind (Phantom-Cleanup). Trades
            # mit closed_by='unknown' AND pnl=0 sind gewollt (uns fehlen
            # die echten Exit-Daten), das ist keine Anomalie.
            out["n_pnl_zero_closed"] = conn.execute("""
                SELECT COUNT(*) FROM trades
                 WHERE exit_time IS NOT NULL
                   AND (pnl IS NULL OR pnl = 0)
                   AND COALESCE(closed_by, '') != 'unknown'
            """).fetchone()[0]

            # Duplikat-Tickets (UNIQUE-Constraint sollte das verhindern,
            # aber wir prüfen trotzdem)
            dups = conn.execute("""
                SELECT ticket, COUNT(*) c FROM trades
                 GROUP BY ticket HAVING c > 1
            """).fetchall()
            out["duplicate_tickets"] = [{"ticket": r["ticket"], "count": r["c"]}
                                        for r in dups]

            # Ältester Trade
            row = conn.execute(
                "SELECT MIN(entry_time) FROM trades"
            ).fetchone()
            if row and row[0]:
                out["oldest_trade_age_days"] = (now - row[0]) / 86400.0

        return out

    def all_trades(self, period: str = "all") -> list[dict]:
        """Vollständige Trade-Liste für CSV-Export."""
        since, until = self._range_to_ts(period)
        with self._connect() as conn:
            rows = conn.execute("""
                SELECT * FROM trades
                 WHERE entry_time BETWEEN ? AND ?
                 ORDER BY entry_time DESC
            """, (since, until)).fetchall()
        return [dict(r) for r in rows]

    def export_csv(self, file_path: Path, period: str = "all") -> int:
        """Exportiert Trades als CSV. Liefert Anzahl exportierter Zeilen."""
        import csv
        rows = self.all_trades(period)
        if not rows:
            return 0
        with open(file_path, "w", encoding="utf-8", newline="") as f:
            writer = csv.DictWriter(f, fieldnames=rows[0].keys())
            writer.writeheader()
            writer.writerows(rows)
        return len(rows)

    def db_size_mb(self) -> float:
        try:
            return self.db_path.stat().st_size / (1024 * 1024)
        except Exception:
            return 0.0

    def n_total_trades(self) -> int:
        with self._connect() as conn:
            return conn.execute(
                "SELECT COUNT(*) FROM trades WHERE exit_time IS NOT NULL"
            ).fetchone()[0]