Systemzustand in den Tagesreport - der fehlende letzte Meter

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>

CLAUDE.md

@@ -6161,6 +6161,61 @@ Neumessung. Sie bleiben unverändert und reproduzierbar; wo ihre Schlussfolgerun
 relevant wird, gehört sie live-treu **neu gerechnet** (Muster:
 `backtest_auto_signal_norev.py`).
 
+## ✅✅ SYSTEMZUSTAND IM TAGESREPORT (2026-08-11) — der fehlende letzte Meter
+
+**Selbständige Empfehlung nach der Gesamtprüfung, vom User beauftragt.**
+
+⚠⚠ **DER BEFUND: die Erkennung funktioniert, die ZUSTELLUNG nicht.** Das Projekt
+hat sieben Wächter, jeden nach einem eigenen Vorfall gebaut — und jeder tut, was
+er soll. Trotzdem blieben allein in dieser Woche **sechs** Zustände tagelang
+unbemerkt, **vier davon waren bereits erkannt**:
+| Vorfall | unbemerkt | erkannt? |
+|---|---|---|
+| `auto_sr_close` aus | **5 Tage** | ✅ 1× gemeldet, dann Marker → Stille |
+| `auto_squeeze` aus | seit 11.08. | ✅ — Marker von **02.08.** → **gar keine** Meldung |
+| `rec_outcomes` tot | **5 Tage** | ❌ (Puls-Wächter heute gebaut) |
+| Telegram still | **4 Tage** | ❌ |
+| Divergenz-Wächter Dauerwarnung | **6 Tage** | er *war* der Wächter |
+| `squeeze_entry_gap` 8/12 | seit 05.08. | ✅ „sammelt noch" |
+
+✅ **BELEGT AN DEN MARKER-DATEIEN:** `cfg_auto_sr_close` trägt **06.08. 18:10**,
+`cfg_auto_squeeze` trägt **02.08. 18:00**. Der Config-Wächter meldet je Schlüssel
+**genau einmal** und schweigt danach für immer. `auto_sr_close` war damit fünf
+Tage aus — und wurde am 11.08. mit **+350,81 € über 43 Trades** gemessen
+(grob 150–300 € entgangener Wert, je nach Vergleichbarkeit der Bedingungen).
+⚠⚠ **Die Ursache ist ein Kategorienfehler:** die Marker-Logik behandelt einen
+**Dauerzustand** wie ein **Einmal-Ereignis**.
+
+✅ **GEBAUT: `engine._system_status()` → Block im Tagesreport (07:30, Telegram +
+E-Mail).** Der Report ist der einzige Kanal, der nachweislich **jeden Tag**
+ankommt (Telegram-Log lückenlos `ok`); der Popup verstummt nach dem ersten Mal.
+**Saubere Trennung: Popup = „etwas Neues ist passiert", Tagesreport = „so steht
+das System gerade".**
+Realer erster Block:
+```
+⚙ Systemzustand:
+  Automatik: BRK aus · SIG aus · S/R-Close AN · Flip aus
+  ⚠ Puls: rec_outcomes: 0 Zeilen — Logger hat NIE geschrieben
+  ⚠ Config: auto_squeeze steht auf 'false' (validiert: 'true')
+  Messungen: pbreak_accuracy_v2 143/350 · squeeze_b5 11/20 · squeeze_entry_gap 8/12
+```
+⚠ **Reine Anzeige** — es wird nichts geschaltet und nichts entschieden.
+⚠ **`measurement_reminder` wird LAZY und GEKAPSELT importiert** (1×/Tag, nicht im
+Live-Pfad): ein Auswertungs-Skript gehört NICHT in den Live-Abhängigkeitsgraphen
+— derselbe Fehler wie der `backtest_breakout_squeeze`-Import. Ein Fehlschlag wird
+**sichtbar** gemeldet, nicht still verschluckt (ein fehlender Block sähe aus wie
+„alles gut" — exakt die Klasse, die `rec_outcomes` fünf Tage verbarg).
+✅ **`telemetrie_puls()` liegt in `core/history.py`** (DB-Schicht), **beide**
+Nutzer rufen dieselbe Funktion — Skripte importieren `core`, nie umgekehrt.
+✅ **7 Tests** (`tests/test_system_status.py`), Schwerpunkt auf den **Ausfallpfaden**:
+ein Report, der wegen der Statuszeile gar nicht ankommt, wäre schlimmer als keine
+Statuszeile. Geprüft: kein Befund wird ausdrücklich bestätigt · ein Puls-Fehler
+wird gemeldet statt verschluckt · ohne `history` kein Absturz · die Methode wirft
+**nie** und liefert immer mindestens eine Zeile · die Puls-Tabellenliste ist
+vollständig.
+⚠ **Verifikation der Zustellung steht aus** — sie erfolgt mit dem Report am
+nächsten Morgen um 07:30. Die Bausteine sind einzeln live gegengeprüft.
+
 ## Gesamtprüfung 2026-08-11 (nach `docs/review-prompt.md`)
 
 Zweiter vollständiger Durchlauf. **Drei echte Defekte**, alle behoben; keine

core/engine.py

@@ -2287,14 +2287,29 @@ class TradingEngine:
         else:
             fc = f"nicht verfügbar ({dl.get('error') or '—'})"
 
+        # ── Systemzustand (2026-08-11) ──────────────────────────────────────
+        # ⚠⚠ Der Tagesreport ist der einzige Kanal, der nachweislich JEDEN Tag
+        #    ankommt (Telegram-Log lückenlos `ok`). Die Wächter kannten den
+        #    Zustand längst — er erreichte nur niemanden. Details und Beleg in
+        #    `_system_status()`.
+        try:
+            zust = self._system_status()
+        except Exception as e:
+            zust = [f"⚠ Systemzustand nicht ermittelbar ({e})"]
+        zust_txt = ("\n\n<b>⚙ Systemzustand:</b>\n" + "\n".join(zust)) if zust else ""
+        zust_html = ("<h3>⚙ Systemzustand</h3><p>"
+                     + "<br>".join(zust) + "</p>") if zust else ""
+
         txt = (f"📊 <b>Oil Tagesreport</b>\n\n"
                f"<b>Heute — Prognose (Web):</b>\n{fc}\n\n"
                f"<b>Gestern ({yday}):</b>\n"
-               f"{n} Trades · WR {wr:.0f}% · PnL {pnl:+.2f} {cur} · PF {pf_s}")
+               f"{n} Trades · WR {wr:.0f}% · PnL {pnl:+.2f} {cur} · PF {pf_s}"
+               f"{zust_txt}")
         html = (f"<h2>Oil Tagesreport</h2>"
                 f"<h3>Heute — Prognose (Web)</h3><p>{fc.replace(chr(10), '<br>')}</p>"
                 f"<h3>Gestern ({yday})</h3><p>{n} Trades · Trefferquote {wr:.0f} % · "
                 f"PnL <b>{pnl:+.2f} {cur}</b> · Profit-Faktor {pf_s}</p>"
+                f"{zust_html}"
                 f"<p style='color:#888;font-size:12px'>Prognose ist Dritt-Kontext "
                 f"(Web-Recherche), kein Handelssignal.</p>")
 
@@ -4334,6 +4349,66 @@ class TradingEngine:
         except Exception:
             return None
 
+    def _system_status(self) -> list[str]:
+        """Klartext-Zeilen zum Systemzustand für den Tagesreport.
+
+        ⚠⚠ WARUM ES DAS GIBT (2026-08-11). Das Projekt hat sieben Wächter, und
+        sie funktionieren — was fehlte, war der letzte Meter. Belegt an den
+        Marker-Dateien: `cfg_auto_sr_close` trägt den Zeitstempel **06.08. 18:10**
+        und wurde danach NIE wieder gemeldet. Fünf Tage lang stand ein Exit aus,
+        der am 11.08. mit **+350,81 € über 43 Trades** gemessen wurde.
+        `cfg_auto_squeeze` trägt sogar 02.08. — als der Schalter am 11.08. umgelegt
+        wurde, kam deshalb GAR KEINE Meldung.
+        ⚠ Die Ursache ist ein Kategorienfehler: die Marker-Logik behandelt einen
+        **Dauerzustand** wie ein **Einmal-Ereignis**. Deshalb die saubere Trennung
+        — **Popup = „etwas Neues ist passiert", Tagesreport = „so steht das System
+        gerade"**. Der Report kommt nachweislich jeden Tag an (Telegram-Log
+        lückenlos `ok`), der Popup verstummt nach dem ersten Mal.
+        ⚠ Reine Anzeige: nichts wird hier geschaltet oder entschieden.
+        """
+        z: list[str] = []
+        try:
+            an = lambda b: "AN" if b else "aus"      # noqa: E731
+            z.append(f"Automatik: BRK {an(self._auto_squeeze)} · "
+                     f"SIG {an(self._auto_signal)} · "
+                     f"S/R-Close {an(self._auto_sr_close)} · "
+                     f"Flip {an(getattr(self, '_flip_close', False))}")
+        except Exception as e:
+            z.append(f"Automatik: nicht lesbar ({e})")
+        try:
+            puls = self.history.telemetrie_puls() if self.history else []
+            z.extend(f"⚠ Puls: {p}" for p in puls)
+            if not puls:
+                z.append("Puls: alle Telemetrie-Tabellen schreiben")
+        except Exception as e:
+            z.append(f"⚠ Puls: Prüfung fehlgeschlagen ({e})")
+        # ── Config-Drift + fällige Messungen aus dem Reminder ────────────────
+        # ⚠ LAZY und GEKAPSELT: `measurement_reminder` ist ein Auswertungs-Skript
+        #   und gehört NICHT in den Live-Abhängigkeitsgraphen (derselbe Fehler wie
+        #   der `backtest_breakout_squeeze`-Import). Der Import passiert deshalb
+        #   erst hier, 1×/Tag, und ein Fehlschlag wird SICHTBAR gemeldet statt
+        #   still verschluckt — ein fehlender Block sähe sonst aus wie „alles gut".
+        try:
+            import measurement_reminder as _mr
+            for d, cur, val in _mr.config_drift():
+                z.append(f"⚠ Config: {d['key']} steht auf '{cur}' "
+                         f"(validiert: '{val}')")
+            faellig = []
+            for c in _mr.CHECKS:
+                try:
+                    have = c["have"]()
+                    if have >= c["need"]:
+                        faellig.append(f"{c['key']} FÄLLIG")
+                    elif have > 0:
+                        faellig.append(f"{c['key']} {have:.0f}/{c['need']}")
+                except Exception:
+                    continue
+            if faellig:
+                z.append("Messungen: " + " · ".join(faellig[:6]))
+        except Exception as e:
+            z.append(f"⚠ Config-/Messungs-Prüfung nicht verfügbar ({e})")
+        return z
+
     def _check_close_notify(self):
         """Telegram bei jedem geschlossenen Trade ab `close_notify_min_eur`.
 

core/history.py

@@ -407,6 +407,58 @@ class HistoryLogger:
             # alert „🔔 CLOSE-Signal") sowie bei Notfall-/Gewinn-Auto-Close. Die terse
             # „Sym +X"-Bestätigung bei jedem Close ist entfernt.
 
+    # ── TELEMETRIE-PULS ──────────────────────────────────────────────────
+    # ⚠⚠ ANLASS (2026-08-11): `rec_outcomes` stand seit dem Einbau am 06.08. bei
+    #    0 Zeilen — ein falsches dict liess die Schreibbedingung nie wahr werden.
+    #    KEIN NameError (der Linter sieht nichts), und das umgebende `except`
+    #    loggt auf DEBUG. Fuenf Tage still tot; aufgefallen nur, weil die Tabelle
+    #    neben ihren Schwestern mit 0 Zeilen stand.
+    # ⚠ Warum hier und nicht im Reminder-Skript: der Puls ist eine DB-Frage, und
+    #   BEIDE Nutzer (Reminder-Popup + Tagesreport) brauchen ihn. Zwei Kopien
+    #   waeren genau die Divergenz, die im Projekt dreimal zugeschlagen hat.
+    #   Richtungsregel: Skripte duerfen `core` importieren, nicht umgekehrt.
+    PULS = (
+        # (Tabelle, Zeitspalte, erwartete Kadenz in Minuten, wozu)
+        ("recommendations",    "timestamp", 1,   "Empfehlungs-Log"),
+        ("candles_m1",         "time",      1,   "M1-Kerzen-Sammlung"),
+        ("verdict_votes",      "ts",        1,   "Modul-Stimmen"),
+        ("m15_states",         "ts",        60,  "M15-Karte Selbstmessung"),
+        ("cone_checks",        "ts",        60,  "Kegel-Kalibrierung (ACI)"),
+        ("pbreak_predictions", "ts",        60,  "P(break)-Prognose-Tracking"),
+        ("rec_outcomes",       "ts",        180, "Selbst-Kalibrierung Empfehlung"),
+    )
+
+    def telemetrie_puls(self) -> list[str]:
+        """Welche Telemetrie-Tabelle schreibt nicht mehr? → Klartext-Zeilen.
+
+        ⚠ Toleranz bewusst grosszuegig (Faktor 20 der Kadenz): ein ruhiger Markt
+        oder ein Wochenende darf keinen Fehlalarm ausloesen — ein Waechter, der
+        oft warnt, wird ignoriert (Lehre vom 02.08.).
+        """
+        out: list[str] = []
+        try:
+            jetzt = time.time()
+            with self._lock, self._connect() as conn:
+                for tab, sp, kadenz, wozu in self.PULS:
+                    try:
+                        n = conn.execute(f"SELECT COUNT(*) FROM {tab}").fetchone()[0]
+                        letzt = conn.execute(f"SELECT MAX({sp}) FROM {tab}").fetchone()[0]
+                    except Exception:
+                        continue                 # Tabelle gibt es (noch) nicht
+                    if not n:
+                        out.append(f"{tab}: 0 Zeilen — Logger hat NIE geschrieben ({wozu})")
+                        continue
+                    # ⚠ `candles_m1.time` ist ROHE Brokerzeit (UTC+3) — ohne den
+                    #   Abzug meldet der Puls dort dauerhaft 3 h Rueckstand.
+                    ts = letzt - 10800 if tab == "candles_m1" else letzt
+                    alter_min = (jetzt - ts) / 60.0
+                    if alter_min > kadenz * 20:
+                        out.append(f"{tab}: letzte Zeile vor {alter_min/60:.1f} h "
+                                   f"(erwartet ~alle {kadenz} min) — {wozu}")
+        except Exception:
+            return out                            # fail-open wie der Rest der Schicht
+        return out
+
     def trade_by_ticket(self, ticket: int) -> dict | None:
         """Einen Trade komplett lesen — für die Close-Benachrichtigung (2026-08-11).
 

measurement_reminder.py

@@ -427,44 +427,20 @@ CONFIG_DEPS = [
 # ⚠ Die Toleranz ist grosszügig (Faktor 20 der erwarteten Kadenz), damit ein
 #   ruhiger Markt keinen Fehlalarm auslöst — ein Wächter, der oft warnt, wird
 #   ignoriert (Lehre vom 02.08.).
-_PULS = [
+# ⚠ Die Implementierung liegt in `core/history.py` (DB-Schicht) — BEIDE Nutzer
-    # (Tabelle, Zeitspalte, erwartete Kadenz in Minuten, wozu)
+#   (dieses Popup und der Tagesreport in `engine._send_daily_report`) rufen
-    ("recommendations",    "timestamp", 1,    "Empfehlungs-Log"),
+#   dieselbe Funktion. Zwei Kopien waeren die Divergenz-Falle, die im Projekt
-    ("candles_m1",         "time",      1,    "M1-Kerzen-Sammlung"),
+#   dreimal zugeschlagen hat. Richtungsregel: Skripte importieren `core`.
-    ("verdict_votes",      "ts",        1,    "Modul-Stimmen"),
-    ("m15_states",         "ts",        60,   "M15-Karte Selbstmessung"),
-    ("cone_checks",        "ts",        60,   "Kegel-Kalibrierung (ACI)"),
-    ("pbreak_predictions", "ts",        60,   "P(break)-Prognose-Tracking"),
-    ("rec_outcomes",       "ts",        180,  "Selbst-Kalibrierung Empfehlung"),
-]
-
-
 def telemetrie_puls() -> list:
-    """Welche Telemetrie-Tabelle schreibt nicht mehr? → Liste von Klartext-Zeilen."""
+    """Welche Telemetrie-Tabelle schreibt nicht mehr? → Klartext-Zeilen."""
-    import sqlite3
-    import time as _t
-    out = []
     try:
-        db = sqlite3.connect(DB)
+        from core.history import HistoryLogger
-    except Exception:
+        return HistoryLogger(DB).telemetrie_puls()
-        return out
+    except Exception as e:
-    jetzt = _t.time()
+        return [f"Puls-Pruefung nicht moeglich: {e}"]
-    for tab, sp, kadenz, wozu in _PULS:
+
-        try:
+
-            n = db.execute(f"SELECT COUNT(*) FROM {tab}").fetchone()[0]
+_PULS = ()   # Rueckwaerts-Kompatibilitaet fuer die Zaehl-Ausgabe unten
-            letzt = db.execute(f"SELECT MAX({sp}) FROM {tab}").fetchone()[0]
-        except Exception:
-            continue                      # Tabelle gibt es (noch) nicht
-        if not n:
-            out.append(f"{tab}: 0 Zeilen — der Logger hat NIE geschrieben ({wozu})")
-            continue
-        # ⚠ candles_m1.time ist ROHE Brokerzeit (UTC+3) — sonst 3 h Fehlalarm.
-        alter_min = (jetzt - (letzt - 10800 if tab == "candles_m1" else letzt)) / 60.0
-        if alter_min > kadenz * 20:
-            out.append(f"{tab}: letzte Zeile vor {alter_min/60:.1f} h "
-                       f"(erwartet ~alle {kadenz} min) — {wozu}")
-    db.close()
-    return out
 
 
 def _cfg_now() -> dict:
@@ -591,7 +567,8 @@ def main() -> None:
         print("=" * 78)
         puls = telemetrie_puls()
         if not puls:
-            print(f"\n  ✅ alle {len(_PULS)} Telemetrie-Tabellen schreiben")
+            from core.history import HistoryLogger as _HL
+            print(f"\n  ✅ alle {len(_HL.PULS)} Telemetrie-Tabellen schreiben")
         for x in puls:
             print(f"\n  ⚠ {x}")
         return

tests/test_system_status.py

@@ -0,0 +1,100 @@
+"""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"]))