#!/usr/bin/env python3 """Zeigen die LIVE ausgegebenen Empfehlungen in die richtige Richtung? ANLASS (2026-08-06, User): „Die Empfehlungen heute waren fast immer gegenläufig — sollten wir das Konzept überdenken?" Bevor man ein Konzept verwirft, misst man es. Das ist bewusst KEIN Backtest: gemessen werden die **tatsächlich ausgegebenen** Zeilen aus `recommendations` (also nach ALLEN Live-Gates: Totband, M30-Filter, Anti-Überdehnung, min_conf, Breakout-Bestätigung, Entry-Raum, EIA …) gegen den echten Kursverlauf danach. Was hier steht, ist das, was der User gesehen hat. DREI METHODIK-PUNKTE, ohne die das Ergebnis wertlos wäre: 1) **Gegen die DRIFT lesen, nicht gegen null.** Fällt der Markt im Messzeitraum, sieht JEDES Short gut aus. Ausgewiesen wird deshalb der Überschuss gegenüber der mittleren Kursbewegung derselben Periode (Projekt-Standard seit `backtest_candles.py`). 2) **Episoden statt Zeilen.** Die Empfehlung wird ~1×/Minute geloggt; eine Marktlage erzeugt dutzende identische Zeilen. Ungruppiert zählt man dieselbe Situation vielfach und täuscht sich über die Stichprobengröße. 3) **Zwei Hälften.** Eine Zahl über den Gesamtzeitraum verwischt genau den Regimewechsel, um den es bei einem Trendfolger geht. Aufruf: python analyze_signal_live.py [horizont_min] """ import bisect import sqlite3 import sys from datetime import datetime, timedelta, timezone from zoneinfo import ZoneInfo import MetaTrader5 as mt5 _BERLIN = ZoneInfo("Europe/Berlin") _BROKER = timezone(timedelta(hours=3)) _UTC = timezone.utc _LUECKE_S = 300 # >5 min Abstand = neue Episode _HORIZONTE = (30, 60, 120) def _berlin(ts): return (datetime.fromtimestamp(int(ts), _UTC) .replace(tzinfo=_BROKER).astimezone(_BERLIN)) def _atr(H, L, C, n=14): tr = [0.0] + [max(H[i] - L[i], abs(H[i] - C[i - 1]), abs(L[i] - C[i - 1])) for i in range(1, len(C))] out, s = [0.0] * len(C), 0.0 for i in range(1, len(C)): s += tr[i] if i > n: s -= tr[i - n] out[i] = s / min(i, n) return out def main(): hz = [int(sys.argv[1])] if len(sys.argv) > 1 else list(_HORIZONTE) if not mt5.initialize(): print("MT5 nicht erreichbar"); return 1 sym = next((c for c in ("SpotCrude", "USOIL", "WTI", "XTIUSD") if mt5.symbol_info(c)), None) b = mt5.copy_rates_from_pos(sym, mt5.TIMEFRAME_M5, 0, 30000) mt5.shutdown() H = [float(x["high"]) for x in b]; L = [float(x["low"]) for x in b] C = [float(x["close"]) for x in b] EP = [_berlin(x["time"]).timestamp() for x in b] A = _atr(H, L, C) # ── Episoden bilden ────────────────────────────────────────────────── db = sqlite3.connect("oil_widget_history.db") db.row_factory = sqlite3.Row rows = db.execute( "SELECT timestamp ts, signal, conf_pct, setup FROM recommendations " "WHERE signal IN ('LONG','SHORT') ORDER BY timestamp").fetchall() epi, letzte = [], None for r in rows: if (letzte and r["signal"] == letzte["signal"] and r["ts"] - letzte["ts"] <= _LUECKE_S): letzte = r continue epi.append(r) letzte = r print("=" * 90) print(" LIVE-EMPFEHLUNGEN gegen den echten Kursverlauf") print("=" * 90) print(f" {len(rows)} geloggte Zeilen → {len(epi)} Episoden " f"({_berlin(EP[0] if False else rows[0]['ts']):%d.%m.} – " f"{datetime.fromtimestamp(rows[-1]['ts'], _BERLIN):%d.%m.%Y})") print(" ⚠ Ohne Episoden-Bildung wären es 20k 'Fälle' — dieselbe Marktlage,") print(" vielfach gezählt. Das ist die Zahl, die zählt.") mitte = epi[len(epi) // 2]["ts"] def bar(ts): j = bisect.bisect_right(EP, ts) - 1 return j if 0 <= j < len(C) else None for minuten in hz: h = minuten // 5 print() print("=" * 90) print(f" HORIZONT {minuten} MINUTEN") print("=" * 90) # Drift-Null der jeweiligen Hälfte: mittlere Kursbewegung ueber h Bars for lbl, lo, hi in (("H1 (erste Hälfte)", 0, mitte), ("H2 (zweite Hälfte)", mitte, 1 << 62)): grp = [e for e in epi if lo <= e["ts"] < hi] js = [bar(e["ts"]) for e in grp] paare = [(e, j) for e, j in zip(grp, js) if j is not None and j + h < len(C) and A[j] > 0] if len(paare) < 20: print(f" {lbl:<20} zu wenige Episoden ({len(paare)})") continue # Drift ueber genau dieselben Bar-Indizes (nicht ueber den ganzen Chart) drift = sum((C[j + h] - C[j]) / A[j] for _, j in paare) / len(paare) sig = [((C[j + h] - C[j]) * (1 if e["signal"] == "LONG" else -1)) / A[j] for e, j in paare] m = sum(sig) / len(sig) nL = sum(1 for e, _ in paare if e["signal"] == "LONG") # Erwartung bei NULL Information = Drift × mittlere Richtung d_mittel = (nL - (len(paare) - nL)) / len(paare) null = drift * d_mittel wr = 100.0 * sum(1 for x in sig if x > 0) / len(sig) print(f" {lbl:<20} n={len(paare):<5} L/S {nL}/{len(paare)-nL} " f"Ø {m:+.3f}×ATR Treffer {wr:.1f} %") print(f" {'':<20} Drift-Null {null:+.3f} → " f"**Überschuss {m - null:+.3f}×ATR**") # ── Der Tag, um den es ging ────────────────────────────────────────── print() print("=" * 90) print(" HEUTE (der Anlass der Frage)") print("=" * 90) heute = datetime.now(_BERLIN).strftime("%Y-%m-%d") grp = [e for e in epi if datetime.fromtimestamp(e["ts"], _BERLIN).strftime("%Y-%m-%d") == heute] if not grp: print(" keine gerichteten Episoden heute") for e in grp: j = bar(e["ts"]) if j is None or A[j] <= 0: continue t = datetime.fromtimestamp(e["ts"], _BERLIN) aus = [] for minuten in (30, 60, 120): k = j + minuten // 5 if k < len(C): r = (C[k] - C[j]) * (1 if e["signal"] == "LONG" else -1) / A[j] aus.append(f"{minuten}′ {r:+.2f}") print(f" {t:%H:%M} {e['signal']:<6} conf {e['conf_pct'] or 0:<3} " f"@ {C[j]:.3f} " + " · ".join(aus)) return 0 if __name__ == "__main__": sys.exit(main())