ah-admin/AH-Oil-Trader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b2a1e35f5e5c8506ead7dc4babdccfba0f383ddd
AH-Oil-Trader/core/history.py
T
Axel HocksandClaude Opus 5 b18b594639 Systemzustand in den Tagesreport - der fehlende letzte Meter 
Selbstaendige Empfehlung nach der Gesamtpruefung, vom User beauftragt.

Befund: die Erkennung funktioniert, die Zustellung nicht. Sieben
Waechter, jeder nach einem eigenen Vorfall gebaut, jeder funktioniert -
und trotzdem blieben diese Woche sechs Zustaende tagelang unbemerkt,
vier davon waren bereits erkannt.

Belegt an den Marker-Dateien: cfg_auto_sr_close traegt 06.08. 18:10,
cfg_auto_squeeze sogar 02.08. Der Config-Waechter meldet je Schluessel
genau einmal und schweigt danach fuer immer. auto_sr_close war damit
fuenf Tage aus - gemessen +350,81 EUR ueber 43 Trades. Ursache ist ein
Kategorienfehler: die Marker-Logik behandelt einen Dauerzustand wie ein
Einmal-Ereignis.

Gebaut: engine._system_status() als Block im Tagesreport (07:30,
Telegram + E-Mail) - der einzige Kanal, der nachweislich jeden Tag
ankommt. Saubere Trennung: Popup = "etwas Neues", Report = "so steht
es gerade". Reine Anzeige.

- telemetrie_puls() liegt in core/history.py, beide Nutzer rufen
  dieselbe Funktion (Skripte importieren core, nie umgekehrt)
- measurement_reminder wird lazy und gekapselt importiert; ein
  Fehlschlag wird sichtbar gemeldet statt still verschluckt
- tests/test_system_status.py: 7 Tests, Schwerpunkt Ausfallpfade -
  der Report darf an dieser Zeile nie scheitern

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-11 22:43:24 +02:00

1341 lines
63 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
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."""
try:
with self._lock, self._connect() as conn:
rows = conn.execute("""
SELECT hit30, hit60, ret30 FROM m15_states
WHERE done=1 AND lesart IN ('auf','ab')
ORDER BY id DESC LIMIT ?
""", (int(last_n),)).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}
n = len(rows)
if not n:
return {"n": 0, "offen": offen}
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,
"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]
Reference in New Issue View Git Blame Copy Permalink