ah-admin/AH-Oil-Trader
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d0f3b78796ea67912c0f2afaa1077c491160e967
AH-Oil-Trader/core/history.py
T
Axel HocksandClaude Opus 5 97617cd81d Deployment-Drift: Stufe 1 (Trail-mult) + Stufe 2 (Entscheidungs-Telemetrie) 
Behebt die strukturelle Schwachstelle, die heute dreimal zugeschlagen hat:
Komponenten werden unter anderen Bedingungen BETRIEBEN als VALIDIERT.
Track B schuetzt gegen Overfitting, nicht gegen Deployment-Drift.

STUFE 1 (core/trailing.py): _MULT_BY_TF durchgehend 1,5 (vorher M15 2,0 /
M30 2,5 / H1 3,0) und _MULT_BREAKOUT_ADD 0,5 -> 0,0. Die Staffelung war nie
gemessen; backtest_trailing.py validiert nur 1,5 und die breiteren Werte
sind dort messbar schlechter (80k Bars, 2 Halbjahre):
    mult 1,5  H1 SigmaR  -249 · H2 +1252 · Worst -1,50
    mult 2,0  H1 SigmaR  -412 · H2 +1187 · Worst -2,00
    mult 2,5  H1 SigmaR -1367 · H2 +1450 · Worst -2,50
Auf M30 fuhr der Bot also 5,5x schlechteres H1 und einen 67 % groesseren
Einzelverlust; bei Breakouts waren es 3,0 - nie getestet. Relevant, weil die
TF-Heuristik oft auf M15/M30 landet (31.07.: 19 von 33 Phasen).

STUFE 2a (wave_rec/history/engine): jede Empfehlung loggt einen
maschinenlesbaren block_reason - deadband, dead_hour, eia, htf_counter,
stretch, min_conf, breakout_pending, entry_room, no_data, stale. DB migriert
(Backup angelegt). Ohne den war nicht feststellbar, WELCHES der Gates die
93 % WARTEN erzeugt (Backtest-Erwartung ~43 %).

STUFE 2b (analyze_divergence.py, NEU): haelt vier Dinge gegen die
Backtest-Erwartung - A Signal-Verteilung, B Gate-Anteile inkl.
"im Backtest modelliert?", C TF-Wechsel-Rate, D0 Modell-Kalibrierung,
D Merkmals-Drift gegen _PB_MU/_SD.

LEHRE AUS DEM BAU SELBST: die Merkmals-Drift (D) haette den realen Fehler
NICHT gefangen - mom3 lag nur 0,50 Sigma unter mu, bei Gewicht 1,43 wurden
daraus aber -0,71 im Logit (23 % statt 41 % vorhergesagt). Deshalb ist D0
der schaerfere Test und die D-Schwelle auf 0,5 Sigma gesetzt.

Erster Lauf nach dem Fix bestaetigt das Nachtraining live:
D0 = 27,0 % vorhergesagt vs 28,0 % real (vorher 23 vs 41).

Stufe 3 (geteilter Exit-Kern) bleibt offen, spaeter und einzeln.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-07-31 12:54:34 +02:00

962 lines
43 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 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):
"""Loggt das Öffnen eines Trades. Idempotent über UNIQUE(ticket)."""
# 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())
with self._lock, self._connect() as conn:
conn.execute("""
INSERT OR IGNORE INTO trades
(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)
VALUES (?,?,?,?, ?,?,?,?, ?,?,?,?, ?,?,?,?)
""", (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))
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.
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 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 alignment_stats(self, last_n: int = 40) -> dict:
"""Ausrichtungs-Split der letzten N geschlossenen Trades: liefen sie MIT,
GEGEN oder OHNE die Empfehlung (rec_signal beim Entry)? Für die ehrliche
Verhaltens-Anzeige im Order-Bestätigungsdialog (B0 als Software — die
gemessene Kern-Leckage sind diskretionäre Gegen-Signal-Trades)."""
with self._connect() as conn:
rows = conn.execute("""
SELECT direction, rec_signal, pnl FROM trades
WHERE exit_time IS NOT NULL AND pnl IS NOT NULL
ORDER BY exit_time DESC LIMIT ?
""", (last_n,)).fetchall()
out = {}
for key, match in (("with", True), ("against", False)):
grp = []
for r in rows:
rec = (r["rec_signal"] or "").upper()
if rec not in ("LONG", "SHORT"):
continue
aligned = (rec == "LONG") == (r["direction"] == "BUY")
if aligned == match:
grp.append(r["pnl"])
n = len(grp)
out[key] = {"n": n,
"wr": round(100 * sum(1 for p in grp if p > 0) / n) if n else None,
"pnl": round(sum(grp), 2)}
none_grp = [r["pnl"] for r in rows
if (r["rec_signal"] or "").upper() not in ("LONG", "SHORT")]
n = len(none_grp)
out["none"] = {"n": n,
"wr": round(100 * sum(1 for p in none_grp if p > 0) / n) if n else None,
"pnl": round(sum(none_grp), 2)}
return out
# ══════════════════════════════════════════
# 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 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