Telegram bei gruener M15-Karte (m15_alert) + Tests verschmutzen nicht mehr das Log
User: "wenn es eine Empfehlung gibt (gruen), schicke eine Telegram-Nachricht." engine._check_m15_alert, direkt hinter _log_m15_state -- dieselbe Quelle wie Anzeige und Selbstmessung, damit die drei nicht auseinanderlaufen koennen. DIE ENTPRELLUNG IST DER GANZE BAU. Roh springt die Karte 85x/Tag auf gruen (m15_states, 3 Tage nach der Hysterese vom 12.08.) -- das waere Spam. Alle Stufen an den echten Daten kalibriert, nicht geraten: nur Flanke 85,0/Tag + Level & Richtung als Schluessel 23,7/Tag + 30 min Mindestabstand 13,3/Tag + Fenster 8-20 Uhr (Default) 7,7/Tag (Vergleich: Close-Meldung ab 50 EUR 2,4-6,1/Tag) Nach 4 h darf dasselbe Level erneut melden, sonst verschweigt der Alert ein Setup, das den ganzen Tag wiederkommt. Der Text sagt ausdruecklich, dass es KEINE Prognose ist: die Karte automatisch zu handeln faellt gemessen durch (backtest_m15_auto.py: OeR -0,162/-0,094). Die Meldung ist ein Hinweis zum Hinsehen fuer den Menschen (+3,20 EUR/Lot ueber 146 Trades). Die P(break)-Formulierung ist wortgleich zu Karte und Chart (v1.35). 13 Tests, Schwerpunkt auf dem was NICHT gesendet werden darf. Mutationsprobe in vier Zweige, alle gefangen. DIE PROBE HAT EIN TESTLOCH AUFGEDECKT: das Entfernen der Flankenerkennung liess zunaechst ALLE Tests gruen, weil Cooldown und 4-h-Schluessel dasselbe abfangen. Zwei Schutzschichten, die sich maskieren -- dasselbe Muster wie beim Fill-Dedup. Geschlossen durch test_flankenerkennung_ISOLIERT. NEBENBEFUND, behoben: Tests schrieben ins Produktions-Log (ueber 150 M15-Alert-Zeilen + Close-Push #4711 aus test_close_buchung). Ich hielt die Entprellung deshalb zuerst fuer live defekt. conftest.py entfernt den FileHandler des oil-Loggers jetzt fuer die Sitzung -- vierte Grundregel der Suite. Verifiziert: Log-Groesse vor und nach einem Komplettlauf identisch. Live verifiziert: seit dem Serverstart genau 1 Alert trotz mehrfachem gruen/gelb-Wechsel. 85 Tests gruen. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 5
parent
c251652d33
commit
44191947ac
@@ -17,6 +17,7 @@ damit dokumentiert, statt still zu passieren.
|
||||
"""
|
||||
import sys
|
||||
import types
|
||||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
@@ -28,6 +29,33 @@ from core.engine import TradingEngine # noqa: E402
|
||||
from core.history import HistoryLogger # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True, scope="session")
|
||||
def _kein_produktions_log():
|
||||
"""VIERTE Grundregel der Suite: Tests schreiben NICHT ins Produktions-Log.
|
||||
|
||||
⚠⚠ ANLASS (2026-08-14, real passiert): beim Bau des M15-Alerts standen nach
|
||||
den Testläufen **über 150** `📨 M15-Alert`-Zeilen in `oil_widget.log` — dazu
|
||||
`Close-Push #4711` mit dem synthetischen Ticket aus `test_close_buchung.py`.
|
||||
Ich habe daraufhin zuerst geglaubt, die Entprellung sei live defekt, und
|
||||
musste die Herkunft mühsam an der Level-Folge (82.00, 82.01, …) nachweisen.
|
||||
In einem Projekt, dessen Forensik am Log hängt (der 18-h-Ausfall des
|
||||
Copiloten, der stille `positions_get`-Fehlbuchung), sind erfundene Zeilen
|
||||
im Produktions-Log ein echtes Risiko — nicht nur Unordnung.
|
||||
|
||||
`core/logger.py` haengt einen `RotatingFileHandler` an den Logger `oil`,
|
||||
sobald `core.engine` importiert wird. Hier wird er fuer die Dauer der
|
||||
Sitzung entfernt und danach zurueckgegeben — die Log-AUFRUFE laufen weiter
|
||||
(sie gehoeren zum getesteten Code), nur ihr Ziel entfaellt.
|
||||
"""
|
||||
lg = logging.getLogger("oil")
|
||||
weg = [h for h in lg.handlers if isinstance(h, logging.FileHandler)]
|
||||
for h in weg:
|
||||
lg.removeHandler(h)
|
||||
yield
|
||||
for h in weg:
|
||||
lg.addHandler(h)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def db(tmp_path):
|
||||
"""Frische, vollständig angelegte Test-Datenbank.
|
||||
|
||||
@@ -0,0 +1,181 @@
|
||||
"""M15-Alert: die ENTPRELLUNG ist der ganze Bau — also wird sie geprueft.
|
||||
|
||||
ANLASS (2026-08-14, User: „wenn es eine Empfehlung gibt (gruen), schicke eine
|
||||
Telegram-Nachricht").
|
||||
|
||||
⚠⚠ Roh gemessen springt die Karte **85x am Tag** auf gruen (`m15_states`, 3 Tage
|
||||
nach der Hysterese vom 12.08.). Ohne Entprellung waere die Meldung Spam und
|
||||
damit wertlos — genau die Sorte Funktion, die man nach zwei Tagen stummschaltet.
|
||||
Die Stufen und ihre gemessene Wirkung:
|
||||
Flanke allein 85,0/Tag
|
||||
+ Level+Richtung als Schluessel 23,7/Tag
|
||||
+ 30 min Mindestabstand 13,3/Tag
|
||||
+ Fenster 8-20 Uhr 7,7/Tag
|
||||
Deshalb liegt der Schwerpunkt der Tests auf dem, was NICHT gesendet werden darf.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
class _Fake:
|
||||
"""Minimal-Engine mit genau den Feldern, die `_check_m15_alert` anfasst."""
|
||||
|
||||
def __init__(self, **kw):
|
||||
from core.engine import TradingEngine
|
||||
self._check_m15_alert = TradingEngine._check_m15_alert.__get__(self)
|
||||
self._m15_alert = kw.get("an", True)
|
||||
self._m15_alert_cd = kw.get("cd", 1800.0)
|
||||
self._m15_alert_hours = kw.get("hours", (0, 24))
|
||||
self._m15_alert_war = False
|
||||
self._m15_alert_ts = 0.0
|
||||
self._m15_alert_key = None
|
||||
self.cfg = {"telegram": {"enabled": "true", "bot_token": "t", "chat_id": "c"}}
|
||||
|
||||
|
||||
def _karte(status="gruen", lesart="auf", level=82.069, pb=69):
|
||||
return {"status": status, "lesart": lesart,
|
||||
"nah": {"seite": "res", "level": level, "gap_atr15": 0.48, "p_break": pb},
|
||||
"bias": {"verboten": "SHORT"}, "kosten": {"ratio": 0.11}}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def gesendet(monkeypatch):
|
||||
"""Faengt jeden Telegram-Versand ab — kein Netz im Test (conftest-Regel)."""
|
||||
raus = []
|
||||
monkeypatch.setattr("core.engine.send_telegram",
|
||||
lambda text, *a, **k: raus.append(text))
|
||||
return raus
|
||||
|
||||
|
||||
def test_gruene_flanke_sendet(gesendet):
|
||||
e = _Fake()
|
||||
e._check_m15_alert(_karte())
|
||||
assert len(gesendet) == 1
|
||||
assert "Setup bereit" in gesendet[0]
|
||||
|
||||
|
||||
def test_zustand_sendet_nur_EINMAL(gesendet):
|
||||
"""⚠ Die wichtigste Zeile: gruen bleibt gruen ueber viele Ticks. Ohne
|
||||
Flankenerkennung kaeme im Sekundentakt eine Nachricht."""
|
||||
e = _Fake()
|
||||
for _ in range(50):
|
||||
e._check_m15_alert(_karte())
|
||||
assert len(gesendet) == 1
|
||||
|
||||
|
||||
def test_flankenerkennung_ISOLIERT(gesendet):
|
||||
"""⚠⚠ DIESER Test existiert wegen einer Mutationsprobe (2026-08-14): das
|
||||
Entfernen der Flankenerkennung liess ALLE anderen Tests gruen — weil der
|
||||
Cooldown und der 4-h-Schluessel dasselbe mit abfangen. Zwei Schutzschichten,
|
||||
die sich gegenseitig maskieren, sind genau das Muster, das im Projekt beim
|
||||
Fill-Dedup schon dokumentiert ist.
|
||||
|
||||
Isoliert wird sie so: Cooldown AUS und ein Level, das bei jedem Tick leicht
|
||||
wandert (real der Normalfall — der Kurs laeuft, der naechste Pivot rutscht).
|
||||
Damit greift WEDER Cooldown NOCH Schluessel; bleibt nur die Flanke.
|
||||
"""
|
||||
e = _Fake(cd=0.0)
|
||||
for i in range(30):
|
||||
e._check_m15_alert(_karte(level=82.0 + i * 0.01))
|
||||
assert len(gesendet) == 1, (
|
||||
"ohne Flankenerkennung kaeme hier eine Nachricht je Tick")
|
||||
|
||||
|
||||
def test_flackern_am_selben_level_sendet_nicht_erneut(gesendet):
|
||||
"""gruen -> gelb -> gruen am SELBEN Level ist dasselbe Setup, das flackert.
|
||||
Gemessen der Haupttreiber: 85 -> 23,7/Tag allein durch diesen Schluessel."""
|
||||
e = _Fake()
|
||||
for _ in range(10):
|
||||
e._check_m15_alert(_karte())
|
||||
e._check_m15_alert(_karte(status="gelb"))
|
||||
assert len(gesendet) == 1
|
||||
|
||||
|
||||
def test_neues_level_sendet_trotz_cooldown_nicht(gesendet):
|
||||
"""⚠ Der Mindestabstand hat VORRANG vor dem neuen Level — sonst genuegt ein
|
||||
um einen Cent verschobenes Level, um die Entprellung auszuhebeln."""
|
||||
e = _Fake()
|
||||
e._check_m15_alert(_karte(level=82.069))
|
||||
e._check_m15_alert(_karte(status="gelb"))
|
||||
e._check_m15_alert(_karte(level=83.500)) # anderes Level, aber sofort
|
||||
assert len(gesendet) == 1
|
||||
|
||||
|
||||
def test_neues_level_sendet_nach_cooldown(gesendet):
|
||||
e = _Fake(cd=0.0)
|
||||
e._check_m15_alert(_karte(level=82.069))
|
||||
e._check_m15_alert(_karte(status="gelb"))
|
||||
e._check_m15_alert(_karte(level=83.500))
|
||||
assert len(gesendet) == 2
|
||||
|
||||
|
||||
def test_gleiches_level_wiederholt_nicht_vor_vier_stunden(gesendet):
|
||||
"""Ohne Cooldown, aber gleicher Schluessel -> erst nach _ALERT_NEU_S wieder."""
|
||||
e = _Fake(cd=0.0)
|
||||
e._check_m15_alert(_karte(level=82.069))
|
||||
e._check_m15_alert(_karte(status="gelb"))
|
||||
e._check_m15_alert(_karte(level=82.069))
|
||||
assert len(gesendet) == 1
|
||||
|
||||
|
||||
def test_ausserhalb_des_stundenfensters_still(gesendet):
|
||||
e = _Fake(hours=(23, 24)) # jetzt ist praktisch nie drin
|
||||
if time.localtime().tm_hour == 23:
|
||||
pytest.skip("Testzeit liegt zufaellig im Fenster")
|
||||
e._check_m15_alert(_karte())
|
||||
assert gesendet == []
|
||||
|
||||
|
||||
def test_unklare_lesart_sendet_nicht(gesendet):
|
||||
"""45-55 % ist ein Muenzwurf — daraus eine Meldung zu machen waere genau der
|
||||
Fehler, den die Karte an drei Stellen schon gemacht hat."""
|
||||
e = _Fake()
|
||||
e._check_m15_alert(_karte(lesart="unklar"))
|
||||
assert gesendet == []
|
||||
|
||||
|
||||
def test_gelb_und_rot_senden_nicht(gesendet):
|
||||
e = _Fake()
|
||||
e._check_m15_alert(_karte(status="gelb"))
|
||||
e._check_m15_alert(_karte(status="rot"))
|
||||
assert gesendet == []
|
||||
|
||||
|
||||
def test_abgeschaltet_sendet_nicht(gesendet):
|
||||
e = _Fake(an=False)
|
||||
e._check_m15_alert(_karte())
|
||||
assert gesendet == []
|
||||
|
||||
|
||||
def test_telegram_deaktiviert_sendet_nicht(gesendet):
|
||||
e = _Fake()
|
||||
e.cfg["telegram"]["enabled"] = "false"
|
||||
e._check_m15_alert(_karte())
|
||||
assert gesendet == []
|
||||
|
||||
|
||||
def test_text_nennt_die_gemessene_lesart_und_warnt(gesendet):
|
||||
"""Die Zahl muss WORTGLEICH zur Karte und zum Chart sein (v1.35-Fix), und der
|
||||
Text darf keine Prognose behaupten."""
|
||||
e = _Fake()
|
||||
e._check_m15_alert(_karte(pb=33))
|
||||
t = gesendet[0]
|
||||
assert "67 % Abprall" in t, "P(Abprall) wie in Karte und Chart"
|
||||
assert "keine Richtungsprognose" in t
|
||||
assert "erlaubt wäre LONG" in t # Veto SHORT -> erlaubt LONG
|
||||
|
||||
|
||||
def test_kein_absturz_ohne_daten(gesendet):
|
||||
e = _Fake()
|
||||
e._check_m15_alert(None)
|
||||
e._check_m15_alert({})
|
||||
e._check_m15_alert({"status": "gruen", "lesart": "auf"}) # `nah` fehlt
|
||||
assert isinstance(gesendet, list)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(pytest.main([__file__, "-v"]))
|
||||
Reference in New Issue
Block a user