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