Selbstaendige Empfehlung nach der Gesamtpruefung, vom User beauftragt. Befund: die Erkennung funktioniert, die Zustellung nicht. Sieben Waechter, jeder nach einem eigenen Vorfall gebaut, jeder funktioniert - und trotzdem blieben diese Woche sechs Zustaende tagelang unbemerkt, vier davon waren bereits erkannt. Belegt an den Marker-Dateien: cfg_auto_sr_close traegt 06.08. 18:10, cfg_auto_squeeze sogar 02.08. Der Config-Waechter meldet je Schluessel genau einmal und schweigt danach fuer immer. auto_sr_close war damit fuenf Tage aus - gemessen +350,81 EUR ueber 43 Trades. Ursache ist ein Kategorienfehler: die Marker-Logik behandelt einen Dauerzustand wie ein Einmal-Ereignis. Gebaut: engine._system_status() als Block im Tagesreport (07:30, Telegram + E-Mail) - der einzige Kanal, der nachweislich jeden Tag ankommt. Saubere Trennung: Popup = "etwas Neues", Report = "so steht es gerade". Reine Anzeige. - telemetrie_puls() liegt in core/history.py, beide Nutzer rufen dieselbe Funktion (Skripte importieren core, nie umgekehrt) - measurement_reminder wird lazy und gekapselt importiert; ein Fehlschlag wird sichtbar gemeldet statt still verschluckt - tests/test_system_status.py: 7 Tests, Schwerpunkt Ausfallpfade - der Report darf an dieser Zeile nie scheitern Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
101 lines
3.7 KiB
Python
101 lines
3.7 KiB
Python
"""Der Systemzustand im Tagesreport — sagt er die Wahrheit, und faellt er nie aus?
|
|
|
|
ANLASS (2026-08-11). Das Projekt hat sieben Waechter, und sie funktionieren. Was
|
|
fehlte, war der letzte Meter: die Marker-Datei `cfg_auto_sr_close` traegt den
|
|
Zeitstempel **06.08. 18:10** und wurde danach NIE wieder gemeldet — fuenf Tage
|
|
lang stand ein Exit aus, der am 11.08. mit **+350,81 EUR ueber 43 Trades**
|
|
gemessen wurde. Ursache ist ein Kategorienfehler: die Marker-Logik behandelt
|
|
einen DAUERZUSTAND wie ein EINMAL-EREIGNIS.
|
|
|
|
Deshalb die Trennung: Popup = „etwas Neues", Tagesreport = „so steht es gerade".
|
|
|
|
⚠ Die wichtigste Eigenschaft dieses Blocks ist, dass er NIE den Report sprengt.
|
|
Ein Report, der wegen der Statuszeile gar nicht ankommt, waere schlimmer als
|
|
gar keine Statuszeile — deshalb prueft der Test vor allem die Ausfallpfade.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from core.engine import TradingEngine
|
|
from core.history import HistoryLogger
|
|
|
|
|
|
class _Hist:
|
|
def __init__(self, puls):
|
|
self._puls = puls
|
|
|
|
def telemetrie_puls(self):
|
|
if isinstance(self._puls, Exception):
|
|
raise self._puls
|
|
return self._puls
|
|
|
|
|
|
class _Stub:
|
|
_system_status = TradingEngine._system_status
|
|
|
|
def __init__(self, *, brk=False, sig=False, src=True, flip=False, puls=None):
|
|
self._auto_squeeze = brk
|
|
self._auto_signal = sig
|
|
self._auto_sr_close = src
|
|
self._flip_close = flip
|
|
self.history = _Hist(puls if puls is not None else [])
|
|
|
|
|
|
def _text(z):
|
|
return "\n".join(z)
|
|
|
|
|
|
def test_automatik_zeile_nennt_alle_vier_schalter():
|
|
z = _text(_Stub(brk=True, sig=False, src=True, flip=False)._system_status())
|
|
assert "BRK AN" in z and "SIG aus" in z and "S/R-Close AN" in z and "Flip aus" in z
|
|
|
|
|
|
def test_toter_logger_wird_gemeldet():
|
|
"""Der reale Fund: rec_outcomes stand 5 Tage bei 0 Zeilen."""
|
|
z = _text(_Stub(puls=["rec_outcomes: 0 Zeilen — Logger hat NIE geschrieben"])
|
|
._system_status())
|
|
assert "⚠ Puls:" in z and "rec_outcomes" in z
|
|
|
|
|
|
def test_stiller_puls_wird_ausdruecklich_bestaetigt():
|
|
"""⚠ Kein Befund muss SICHTBAR sein — eine fehlende Zeile liest sich sonst
|
|
wie „nicht geprueft"."""
|
|
z = _text(_Stub(puls=[])._system_status())
|
|
assert "alle Telemetrie-Tabellen schreiben" in z
|
|
|
|
|
|
def test_puls_fehler_wird_gemeldet_nicht_verschluckt():
|
|
"""⚠ Genau die Fehlerklasse, die rec_outcomes 5 Tage verbarg: ein stiller
|
|
Fallback saehe aus wie „alles gut"."""
|
|
z = _text(_Stub(puls=RuntimeError("DB weg"))._system_status())
|
|
assert "⚠ Puls: Prüfung fehlgeschlagen" in z and "DB weg" in z
|
|
|
|
|
|
def test_ohne_history_kein_absturz():
|
|
s = _Stub()
|
|
s.history = None
|
|
z = _text(s._system_status())
|
|
assert "Automatik:" in z # der Rest laeuft weiter
|
|
|
|
|
|
def test_liefert_immer_zeilen_und_wirft_nie():
|
|
"""Die Kern-Invariante: der Tagesreport darf an dieser Zeile nicht scheitern."""
|
|
for kw in ({}, {"brk": True}, {"puls": RuntimeError("x")}):
|
|
z = _Stub(**kw)._system_status()
|
|
assert isinstance(z, list) and z, "muss immer mindestens eine Zeile liefern"
|
|
assert all(isinstance(x, str) for x in z)
|
|
|
|
|
|
def test_puls_tabellenliste_ist_vollstaendig():
|
|
"""Jede Telemetrie-Tabelle, die das Projekt schreibt, muss ueberwacht sein —
|
|
sonst entsteht genau die Luecke wieder, die rec_outcomes hatte."""
|
|
ueberwacht = {t for t, *_ in HistoryLogger.PULS}
|
|
for t in ("recommendations", "candles_m1", "verdict_votes", "m15_states",
|
|
"cone_checks", "pbreak_predictions", "rec_outcomes"):
|
|
assert t in ueberwacht, f"{t} wird nicht ueberwacht"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(pytest.main([__file__, "-v"]))
|