Axel HocksandClaude Opus 5
9887e3d5da
Im Projekt sind ueber Monate dutzende Szenario-Tests entstanden ("mit 10
Szenarien getestet", "mit 18 synthetischen Szenarien") — alle als
Wegwerf-Skripte, keiner lief je ein zweites Mal. Bei einem System, das man vor
jeder Aenderung neu starten und live verifizieren muss, ist das der teuerste
Teil der Schleife.
Erfasst sind die beiden juengsten Pfade:
· _check_pending_fill (6 Faelle) — u.a. "fremde Position NICHT taggen" und
"derselbe Fill zaehlt nur einmal"
· Selbst-Kalibrierung (7 Faelle) — Episoden-Dedup, WARTEN re-armt, Auswertung
in beide Richtungen inkl. Broker-Offset, fehlende Bar bleibt offen,
Aggregation
Drei harte Regeln in tests/conftest.py: keine Live-DB (frische Temp-Datei je
Test), kein MT5 / kein laufender Server (gestubbt), kein Netz. Das Schema legen
die ECHTEN Konstruktoren an — HistoryLogger UND CandleLogger, denn candles_m1
gehoert nicht zum History-Schema. Ein handgebautes Test-Schema wuerde
irgendwann vom Produktivstand abweichen.
Warum Temp-DB statt Kopie der Live-DB: am 06.08. haben echte candles_m1-Zeilen
in einem vermeintlich leeren Fenster einen Test verfaelscht — zweimal sah es
nach einem Code-Fehler aus, es waren Testfehler.
MUTATIONS-GEPRUEFT, weil eine Suite die immer gruen ist wertlos waere: eine
absichtlich eingebaute "fremde Position wird doch getaggt"-Regression wird
gefangen. Nebenbefund: der Fill-Dedup ist DOPPELT gesichert (_pending_tagged
und der pop aus _pending_tickets) — nur eines zu entfernen faellt nicht auf,
beides zusammen bricht sofort zwei Tests. Gewollte Redundanz, kein
ungetesteter Zweig.
tests_rec_outcomes.py (Wegwerf-Fassung von gestern) und die verwaiste
test_pbreak.db entfernt.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
134 lines
5.5 KiB
Python
134 lines
5.5 KiB
Python
"""Selbst-Kalibrierung: Episoden-Logging und Auswertung der Empfehlungen.
|
||
|
||
HINTERGRUND (2026-08-06): die Qualität der Gesamtempfehlung war unsichtbar und
|
||
musste auf Nachfrage rückwirkend gemessen werden (48–51 % Treffer über 1.071
|
||
Episoden). `rec_outcomes` schließt die Lücke — dasselbe Muster wie
|
||
`pbreak_predictions`, das den Ausfall des P(break)-Modells LIVE aufgedeckt hat.
|
||
|
||
⚠ Die zwei Fehlschläge beim ersten Schreiben dieser Tests waren TESTFEHLER, keine
|
||
Code-Fehler: die Test-Kerzen des Vorabschnitts lagen im Fenster des nächsten, und
|
||
die damals kopierte Live-DB enthielt ECHTE `candles_m1` im vermeintlich leeren
|
||
Fenster. Beides ist hier konstruktiv ausgeschlossen — jeder Test bekommt eine
|
||
frische, leere DB (s. `conftest.py`).
|
||
"""
|
||
import sqlite3
|
||
import time
|
||
|
||
OFF = 3 * 3600 # candles_m1.time ist ROHE Broker-Zeit (UTC+3)
|
||
|
||
|
||
def _episode(engine, sig):
|
||
"""Der Episoden-Teil aus `_run_analysis`, isoliert nachgestellt."""
|
||
if sig in ("LONG", "SHORT") and sig != engine._rec_epi_sig:
|
||
engine.history.log_rec_outcome(signal=sig, conf_pct=70, tf="M5",
|
||
price=75.0, atr=0.2)
|
||
if sig in ("LONG", "SHORT", "WARTEN"):
|
||
engine._rec_epi_sig = sig if sig != "WARTEN" else None
|
||
|
||
|
||
def _zeilen(db):
|
||
with sqlite3.connect(db) as c:
|
||
return c.execute("SELECT COUNT(*) FROM rec_outcomes").fetchone()[0]
|
||
|
||
|
||
# ── Episoden-Bildung ─────────────────────────────────────────────────────
|
||
def test_gleiche_richtung_erzeugt_eine_zeile(engine, db):
|
||
"""⚠ Der Kern: die Empfehlung wird ~1x/Minute geloggt. Wer die Zeilen zählt,
|
||
hält 20.000 Fälle für eine Stichprobe, obwohl es 1.000 sind."""
|
||
for _ in range(5):
|
||
_episode(engine, "LONG")
|
||
assert _zeilen(db) == 1
|
||
|
||
|
||
def test_flip_ist_eine_neue_episode(engine, db):
|
||
for _ in range(3):
|
||
_episode(engine, "LONG")
|
||
for _ in range(3):
|
||
_episode(engine, "SHORT")
|
||
assert _zeilen(db) == 2
|
||
|
||
|
||
def test_warten_re_armt(engine, db):
|
||
_episode(engine, "SHORT")
|
||
_episode(engine, "WARTEN")
|
||
_episode(engine, "WARTEN")
|
||
assert _zeilen(db) == 1 # WARTEN selbst erzeugt nichts
|
||
_episode(engine, "SHORT")
|
||
assert _zeilen(db) == 2 # danach ist SHORT wieder neu
|
||
|
||
|
||
# ── Auswertung gegen candles_m1 ──────────────────────────────────────────
|
||
def _setze(db, *, ts, signal, price, atr, kerzen):
|
||
with sqlite3.connect(db) as c:
|
||
c.execute("INSERT INTO rec_outcomes (ts,signal,conf_pct,tf,price,atr) "
|
||
"VALUES (?,?,?,?,?,?)", (ts, signal, 70, "M5", price, atr))
|
||
for minuten, kurs in kerzen:
|
||
c.execute("INSERT OR REPLACE INTO candles_m1 "
|
||
"(time,o,h,l,c,spread,tick_volume,symbol) "
|
||
"VALUES (?,?,?,?,?,?,?,?)",
|
||
(ts + minuten * 60 + OFF, kurs, kurs, kurs, kurs,
|
||
20, 1, "TESTSYM"))
|
||
c.commit()
|
||
|
||
|
||
def _ergebnis(db):
|
||
with sqlite3.connect(db) as c:
|
||
return c.execute("SELECT ret30,hit30,ret60,hit60,done "
|
||
"FROM rec_outcomes").fetchone()
|
||
|
||
|
||
def test_long_auswertung_beide_horizonte(engine, db_umleiten):
|
||
db = db_umleiten
|
||
t0 = int(time.time()) - 7200 # beide Horizonte fällig
|
||
_setze(db, ts=t0, signal="LONG", price=75.00, atr=0.20,
|
||
kerzen=((30, 75.10), (60, 74.90)))
|
||
engine._evaluate_rec_outcomes()
|
||
|
||
ret30, hit30, ret60, hit60, done = _ergebnis(db)
|
||
assert round(ret30, 3) == 0.5 and hit30 == 1 # +0,10 / 0,20 ATR
|
||
assert round(ret60, 3) == -0.5 and hit60 == 0
|
||
assert done == 1
|
||
|
||
|
||
def test_short_ist_spiegelbildlich(engine, db_umleiten):
|
||
db = db_umleiten
|
||
t0 = int(time.time()) - 7200
|
||
_setze(db, ts=t0, signal="SHORT", price=75.00, atr=0.20,
|
||
kerzen=((30, 74.90), (60, 74.90)))
|
||
engine._evaluate_rec_outcomes()
|
||
|
||
ret30, hit30, *_ = _ergebnis(db)
|
||
assert round(ret30, 3) == 0.5 and hit30 == 1 # fallender Kurs = Treffer
|
||
|
||
|
||
def test_fehlende_bar_bleibt_offen(engine, db_umleiten):
|
||
"""⚠ Lieber unbewertet als falsch bewertet: fehlt die Bar (Wochenende,
|
||
Broker-Ausfall), darf die Zeile NICHT mit einem Nachbarwert gefüllt werden."""
|
||
db = db_umleiten
|
||
t0 = int(time.time()) - 7200
|
||
_setze(db, ts=t0, signal="SHORT", price=75.00, atr=0.20,
|
||
kerzen=((30, 74.90),)) # für 60 min bewusst keine
|
||
engine._evaluate_rec_outcomes()
|
||
|
||
ret30, hit30, ret60, _, done = _ergebnis(db)
|
||
assert round(ret30, 3) == 0.5 and hit30 == 1
|
||
assert ret60 is None
|
||
assert not done
|
||
|
||
|
||
# ── Aggregation ──────────────────────────────────────────────────────────
|
||
def test_rec_accuracy_rechnet_richtig(engine, db, history):
|
||
with sqlite3.connect(db) as c:
|
||
for ret, hit in ((0.5, 1), (0.3, 1), (-0.4, 0), (-0.2, 0)):
|
||
c.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))
|
||
c.commit()
|
||
|
||
a = history.rec_accuracy(200)
|
||
assert a["n"] == 4
|
||
assert a["wr30"] == 50.0 # ⚠ gegen 50 % lesen, nicht gegen 0
|
||
assert a["avg30"] == 0.05
|