AH-Oil-Trader/tests_rec_outcomes.py

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