"""Szenario-Test der Selbst-Kalibrierung (rec_outcomes) — ohne MT5, auf DB-Kopie. Geprueft werden die Faelle, die live schiefgehen koennen: 1) Episode wird EINMAL geloggt, nicht je Minute 2) WARTEN re-armt → die naechste Richtung ist eine neue Episode 3) Auswertung rechnet den Forward-Return korrekt (inkl. Broker-Offset) 4) fehlende M1-Bars → Zeile bleibt offen statt falsch bewertet 5) rec_accuracy aggregiert Trefferquote und Ø-Return richtig """ import os import shutil import sqlite3 import sys import tempfile import time import types sys.path.insert(0, r"c:\Users\ah\Downloads\tradingrobot") from core.engine import TradingEngine from core.history import HistoryLogger QUELL = r"c:\Users\ah\Downloads\tradingrobot\oil_widget_history.db" TMP = os.path.join(tempfile.mkdtemp(), "t.db") shutil.copy(QUELL, TMP) con = sqlite3.connect(TMP) con.execute("DELETE FROM rec_outcomes") con.commit() h = HistoryLogger(TMP) fehler = [] def pruef(lbl, ist, soll): ok = ist == soll if not ok: fehler.append(lbl) print(f" {'OK ' if ok else 'FEHLER'} {lbl:<50} ist={ist!r:<14} soll={soll!r}") print("=" * 86) print(" 1) Episoden-Logging: eine Zeile je Richtungswechsel, nicht je Minute") print("=" * 86) class E: """Nur der Episoden-Teil aus `_run_analysis`, isoliert nachgestellt.""" def __init__(self): self._rec_epi_sig = None self.history = h def tick(self, sig): if sig in ("LONG", "SHORT") and sig != self._rec_epi_sig: self.history.log_rec_outcome(signal=sig, conf_pct=70, tf="M5", price=75.0, atr=0.2) if sig in ("LONG", "SHORT", "WARTEN"): self._rec_epi_sig = sig if sig != "WARTEN" else None e = E() for s in ["LONG"] * 5: # dieselbe Episode, 5 Minuten lang e.tick(s) pruef("5x LONG hintereinander → 1 Zeile", con.execute("SELECT COUNT(*) FROM rec_outcomes").fetchone()[0], 1) for s in ["SHORT"] * 3: # Flip = neue Episode e.tick(s) pruef("Flip auf SHORT → 2 Zeilen", con.execute("SELECT COUNT(*) FROM rec_outcomes").fetchone()[0], 2) e.tick("WARTEN"); e.tick("WARTEN") pruef("WARTEN erzeugt keine Zeile", con.execute("SELECT COUNT(*) FROM rec_outcomes").fetchone()[0], 2) e.tick("SHORT") pruef("nach WARTEN ist SHORT eine NEUE Episode", con.execute("SELECT COUNT(*) FROM rec_outcomes").fetchone()[0], 3) print() print("=" * 86) print(" 2) Auswertung gegen candles_m1 (mit Broker-Offset)") print("=" * 86) con.execute("DELETE FROM rec_outcomes") con.execute("DELETE FROM candles_m1 WHERE symbol='TESTSYM'") OFF = 3 * 3600 t0 = int(time.time()) - 7200 # vor 2 h → beide Horizonte faellig # LONG @ 75.00, ATR 0.20; nach 30 min steht 75.10 (+0,5 ATR), nach 60 min 74.90 con.execute("INSERT INTO rec_outcomes (ts,signal,conf_pct,tf,price,atr) " "VALUES (?,?,?,?,?,?)", (t0, "LONG", 70, "M5", 75.00, 0.20)) for minuten, kurs in ((30, 75.10), (60, 74.90)): con.execute("INSERT OR REPLACE INTO candles_m1 (time,o,h,l,c,spread,tick_volume,symbol)" " VALUES (?,?,?,?,?,?,?,?)", (t0 + minuten * 60 + OFF, kurs, kurs, kurs, kurs, 20, 1, "TESTSYM")) con.commit() eng = TradingEngine.__new__(TradingEngine) eng._rec_eval_ts = 0.0 eng.trader = types.SimpleNamespace(_broker_offset_s=lambda: OFF) import core.engine as CE _alt = CE.HISTORY_DB_FILE CE.HISTORY_DB_FILE = TMP eng._evaluate_rec_outcomes() CE.HISTORY_DB_FILE = _alt r = con.execute("SELECT ret30,hit30,ret60,hit60,done FROM rec_outcomes").fetchone() pruef("ret30 = (75.10-75.00)/0.20 = +0.5", round(r[0], 3), 0.5) pruef("hit30 = 1 (lief in Signalrichtung)", r[1], 1) pruef("ret60 = (74.90-75.00)/0.20 = -0.5", round(r[2], 3), -0.5) pruef("hit60 = 0", r[3], 0) pruef("done = 1", r[4], 1) print() print("=" * 86) print(" 3) SHORT-Richtung spiegelbildlich · fehlende Bars → bleibt offen") print("=" * 86) con.execute("DELETE FROM rec_outcomes") # ⚠ Test-Kerzen aus Abschnitt 2 MUESSEN weg — sonst faellt die dortige # 60-Minuten-Bar in das Fenster dieses Abschnitts (beide Anker liegen 2 h # zurueck) und der Fall "keine Bar" waere gar nicht getestet. con.execute("DELETE FROM candles_m1 WHERE symbol='TESTSYM'") t1 = int(time.time()) - 7200 con.execute("INSERT INTO rec_outcomes (ts,signal,conf_pct,tf,price,atr) " "VALUES (?,?,?,?,?,?)", (t1, "SHORT", 70, "M5", 75.00, 0.20)) con.execute("INSERT OR REPLACE INTO candles_m1 (time,o,h,l,c,spread,tick_volume,symbol)" " VALUES (?,?,?,?,?,?,?,?)", (t1 + 1800 + OFF, 74.9, 74.9, 74.9, 74.9, 20, 1, "TESTSYM")) # fuer 60 min BEWUSST keine Bar → Zeile darf nicht faelschlich bewertet werden. # ⚠ Dafuer muss das 60-Minuten-Fenster WIRKLICH leer sein: die DB-Kopie enthaelt # ECHTE candles_m1 aus dem Live-Betrieb, und die liegen genau hier. Ohne dieses # DELETE testet man nicht "keine Bar", sondern "echte Bar" (erster Versuch lieferte # ret60 = -14,985 = der reale Kurs ~78). con.execute("DELETE FROM candles_m1 WHERE time > ? AND time <= ?", (t1 + 3600 + OFF - 900, t1 + 3600 + OFF)) con.commit() eng._rec_eval_ts = 0.0 CE.HISTORY_DB_FILE = TMP eng._evaluate_rec_outcomes() CE.HISTORY_DB_FILE = _alt r = con.execute("SELECT ret30,hit30,ret60,done FROM rec_outcomes").fetchone() pruef("SHORT: Kurs faellt → ret30 = +0.5", round(r[0], 3), 0.5) pruef("SHORT: hit30 = 1", r[1], 1) pruef("ret60 ohne Bar bleibt NULL", r[2], None) pruef("done bleibt offen (0/None)", bool(r[3]), False) print() print("=" * 86) print(" 4) rec_accuracy aggregiert") print("=" * 86) con.execute("DELETE FROM rec_outcomes") for ret, hit in ((0.5, 1), (0.3, 1), (-0.4, 0), (-0.2, 0)): con.execute("INSERT INTO rec_outcomes (ts,signal,conf_pct,tf,price,atr," "ret30,hit30,ret60,hit60,done) VALUES (?,?,?,?,?,?,?,?,?,?,1)", (int(time.time()), "LONG", 70, "M5", 75.0, 0.2, ret, hit, ret, hit)) con.commit() a = h.rec_accuracy(200) pruef("n = 4", a["n"], 4) pruef("Trefferquote 30' = 50 %", a["wr30"], 50.0) pruef("Ø-Return 30' = +0.05", a["avg30"], 0.05) con.close() print() print("=" * 86) print(" ERGEBNIS: " + ("alle Szenarien korrekt" if not fehler else f"{len(fehler)} FEHLER: {fehler}")) print("=" * 86) sys.exit(1 if fehler else 0)