"""Abendkontrolle eines Paper-Handelstags: liest nur, urteilt im Code, meldet nur bei Befund. Koordinator-Werkzeug, nicht fuers Repo. Laeuft in WSL im Venv des Projekts; gestartet von abendkontrolle.py (Windows-Aufgabe "Trading Plattform - Abendkontrolle", werktags 22:45 Berlin und bei jeder Anmeldung). Auftrag Pascal 2026-09-24: taegliche Abendkontrolle, dauerhaft, uebersteht Neustarts und blockiert keine Updates von Claude Code oder Desktop -- deshalb ohne Claude-Prozess. Entscheidungsliste (Frage -> Herkunft der Antwort): 1. Ist ein Handelstag? Quelle: GET /v2/calendar?start=&end=, gemessen am 2026-09-24: eine Liste von Objekten mit 'date' (JJJJ-MM-TT) und 'close' (HH:MM New York); Feiertage fehlen, das Wochenende ist leer, verkuerzte Tage tragen ein frueheres 'close'. Antwort nicht lesbar -> Befund, und die Fragen 2 bis 7 bleiben offen. 2. Lief ein scharfer Runner? Quelle: scripts/paper.py:554, Protokoll artifacts/paper/-paper.jsonl. Der Start um 09:10 New York traegt das Datum des Handelstags (gilt, solange max_wartezeit_h unter etwa 13 h bleibt; heute 8). Keines -> Befund. 3. Wie endete jeder Lauf? Quelle: scripts/paper.py:517, Zeile "Ende: , Position , Orders , Killswitch , Exit " im Journal, und die Zeile 'ende' jedes Protokolls. Projektvorgabe: jeder Lauf endet mit boersenschluss und Exit 0, der letzte mit Position 0. Ausgenommen ist ein Lauf ausserhalb der Handelszeit (Ende wartezeit, im Protokoll ohne Takt; gemessen am 2026-09-20 nach WSL-Neustarts): nur ein Hinweis. Ein Lauf, der mittags mit abbruch endet, ist ein Befund, auch wenn der naechste sauber schliesst. Keine Endezeile im Journal -> Befund; "Failed with result" -> Befund. 4. Was sagt das Protokoll? Quelle: handelstag_auswertung.py (dieses Verzeichnis); jede Zeile, die mit "BEFUND" beginnt, ist ein Befund; Exit ungleich 0 ebenfalls. 5. Ist das Konto glatt? Quelle: dieselbe Auswertung mit --broker, Zeile "offene positionen beim broker jetzt: ". n != 0 -> Befund; fehlt die Zeile -> Befund (die Frage blieb unbeantwortet). 6. Hat der Runner gehandelt? Projektvorgabe (Pascal 2026-09-24, Paket AF): an einem Handelstag steht der Killswitch in mindestens einem Takt auf aktiv. Kein solcher Takt -> Befund "Tag ohne Handel". Takte, die nach dem ersten aktiven nicht mehr aktiv sind (Watchdog, Pascals Anhalten), sind ein Hinweis: Der Runner hat dafuer schon selbst gemeldet. Ereignisse auto_fortsetzen werden berichtet, nicht bewertet. 7. Liegt der Tagesbericht? Quelle: scripts/tagesauswertung.py:58 und :192, data/paper/auswertungen/.md. Fehlt -> Befund. 8. Laufen Dashboard und Zeitgeber? Projektvorgabe (Pakete F und Z): trading-dashboard.service, trading-paper.timer, trading-auswertung.timer "active"; ein anderer Zustand wird nach DIENSTE_NOCHMAL_S einmal neu gelesen (eine eben geweckte WSL startet ihre Dienste noch). 9. Wie steht der Killswitch? Quelle: Tabelle killswitch_zustand in data/registry.db (trading/registry/registry.py:281). gesperrt -> Befund: ohne "Freigeben" handelt der naechste Tag nicht. Ein Zustand ausserhalb aktiv/angehalten/gesperrt -> Befund. Jede Frage laeuft fuer sich: Scheitert eine (Sperre im Register, Journal nicht lesbar), ist das ein Befund, und die anderen werden trotzdem beantwortet. Meldung: bei mindestens einem Befund eine Meldung mit hoher Prioritaet ueber den Melder des Projekts. Ob sie ankam, sagt der Rueckgabewert von Melder.melde (None oder der Grund, trading/paper/melder.py); er steht in fakten.meldung. Kein Alarm entsteht, wenn WSL selbst nicht laeuft -- dann steht nur die Zeile in der Tagesnotiz (benannte Luecke; der Totmann in Home Assistant, Paket Z2-7, deckt sie ab). Nicht gesehen: die Guete der Kurse (steht im Bericht), Teilausfuehrungen einzeln, alles vor dem Handelstag. Betraege gibt das Werkzeug nie aus; fremde Zeilen laufen durch maskiere(). Seit Paket IB2 prueft --teil ibkr statt des Alpaca-Runners die IBKR-Instanz in ihrem Betriebsordner (--ibkr-ordner, Vorgabe IBKR_ORDNER), mit eigener Ausgabe und eigener Meldung (Titel "IBKR ..."): I1. Handelstag? wie Frage 1, derselbe Kalender von Alpaca. I2. Lief die Instanz? Protokolle T*-paper.jsonl unter artifacts/paper des Betriebsordners. Keines und trading-paper-ibkr.timer nicht active -> nur der Hinweis "IBKR-Instanz heute nicht gestartet", kein Befund, der Rest entfaellt (No-Go-Tag). Keines bei aktivem oder unlesbarem Zeitgeber -> Befund. Der Zeitgeber wird zum Zeitpunkt der Kontrolle gelesen, nicht fuer den geprueften Tag -- beim Nachholen eine benannte Luecke. I3. Ist der Ordner die Instanz? data/paper/broker.txt sagt ibkr, sonst Befund. I4. Killswitch der Instanz? killswitch_zustand in data/registry.db des Betriebsordners (mode=ro); gesperrt -> Befund mit Stufe und Ausloeser des juengsten Ereignisses. I5. Ende je Lauf, Handel? wie Fragen 3 und 6 (pruefe_laeufe); dazu die Schlussposition jedes Laufs aus seiner Zeile ende (pruefe_schlusspositionen), unabhaengig vom Journal: Position != 0 -> Befund. I6. Journal? wie Frage 3 fuer trading-paper-ibkr.service; jede Zeile mit einem Muster aus IBKR_MUSTER -> Befund. I7. Konto am Gateway glatt? scripts/ibkr_pruefen.py lesen im Betriebsordner mit IBKR_PY (Client-ID 12, lesend), nur werktags ausserhalb 09:00 bis 16:15 New Yorker Zeit und am Wochenende, sonst ein Hinweis. Weiter gehen nur die Zahlen aus "Positionen: n" und "Offene Orders: n" und die BEFUND-Zeilen, maskiert, Kontonummern als [konto]. n != 0, eine fehlende Zeile oder Exit != 0 -> Befund. Nicht gesehen im Teil ibkr: ein Tagesbericht (die Instanz hat keine Tagesauswertung), handelstag_auswertung.py und der Dienst ibkr-gateway.service. Seit Paket E9 prueft --sicherung fuer beide Teile die Sicherung des vorigen Handelstags auf dem Docker-Host. Die Datei schreibt abendkontrolle-host.sh vor jedem Lauf (absoluter Pfad, im Container /abend/sicherung.json); die Ablage selbst sieht das Werkzeug nicht. Die Pruefung laeuft nach dem Teil in einem eigenen Schritt, auch an einem Tag ohne Handel und nach einem Abbruch. E1. Ist der Stand lesbar? Projektvorgabe (Plan E9, Task 726): Datei fehlt, kein JSON, kein Objekt, ohne erster_tag oder eintraege -> Befund. Ein Schluessel fehler (Ablage fehlt, nicht lesbar, erster-tag kaputt) -> Befund. E2. Ab wann wird geprueft? Projektvorgabe (S6, S8): erster_tag aus erster-tag der Ablage; null, kein Datum oder ein Tag nach --tag -> Befund, sonst schwiege die Pruefung auf Dauer; nur ein voriger Handelstag vor erster_tag -> nicht geprueft, still. E3. Welcher Tag? Quelle: derselbe Kalender wie Frage 1, GET /v2/calendar fuer die SICHERUNG_FENSTER_TAGE Kalendertage vor --tag, bis zum Vortag; der juengste Eintrag darin, unabhaengig von der Reihenfolge der Liste. Die Spanne setzt das Werkzeug selbst durch, fuer die Antwort wie fuer eine Kalenderdatei. Form wie Frage 1, dazu date als JJJJ-MM-TT, sonst Befund; keiner -> Befund. E4. Liegt die Sicherung? Projektvorgabe (Schnittstelle E9): eintraege["-"], instanz nach --teil; fehlt der Eintrag -> Befund; summe nicht 64 Hexzeichen in Kleinbuchstaben -> Befund; archiv nicht true (die Datei .tar fehlt auf dem Host) -> Befund. E5. Ist sie nach D: geholt? Projektvorgabe (S4, S5): bestaetigt hat die Form von summe und ist ihr gleich, sonst Befund. Die Windows-Seite bestaetigt erst nach geprueftem Archiv, und abendkontrolle-lesen.sh nimmt nur die Summe des Hosts an. Nicht gesehen: ob die Kopie auf D: noch liegt und ob sie sich zuruecksichern laesst. Ein Werktag ohne Handel hat eine Sicherung, die niemand prueft. Ein Fangnetz, kein Nachweis ueber den Handelstag. Aufruf (WSL, Venv): python abendkontrolle_wsl.py --tag JJJJ-MM-TT [--ohne-meldung] [--teil ibkr] [--sicherung ] Nur eine Meldung schicken: python abendkontrolle_wsl.py --meldung "" "" Fuer Angriffsfaelle ohne Netz und ohne systemd: --projekt, --kalender, --journal, --dienste, --ohne-broker, --ohne-kerzen; im Teil ibkr dazu --ibkr-ordner, --ibkr-konto (JSON mit exit und text statt des Aufrufs von ibkr_pruefen.py) und --jetzt (ISO-Zeitpunkt mit Versatz). Seit dem Nachtrag zu Paket DH (IB2): --ibkr-python nennt den Interpreter fuer ibkr_pruefen.py lesen (Vorgabe IBKR_PY, das Venv in WSL; auf dem Docker-Host python im Abbild, abendkontrolle-host.sh). Ausgabe: genau eine JSON-Zeile auf stdout. Exit 0 gelaufen (Befunde stehen in der Ausgabe), 2 abgebrochen (die bis dahin gesammelten Befunde stehen trotzdem in der Ausgabe). """ import argparse import json import os import re import sqlite3 import subprocess import sys import time from datetime import date from datetime import datetime from datetime import timedelta from pathlib import Path from zoneinfo import ZoneInfo PROJEKT = "/mnt/d/Claude/Code Projekte/active - Trading Plattform" GEHEIMNISSE = "/home/claude/.config/trading-plattform/secrets.toml" AUSWERTUNG = Path(__file__).resolve().parent / "handelstag_auswertung.py" # Nur fuer die Angriffsfaelle: ein Ersatz, der vorgegebene Zeilen druckt, statt ans Netz zu gehen. AUSWERTUNG_ERSATZ = os.environ.get("ABENDKONTROLLE_AUSWERTUNG") HANDEL_URL = "https://paper-api.alpaca.markets" DIENSTE = ("trading-dashboard.service", "trading-paper.timer", "trading-auswertung.timer") DIENSTE_NOCHMAL_S = 30 KILLSWITCH_ZUSTAENDE = ("aktiv", "angehalten", "gesperrt") AUSWERTUNG_TIMEOUT_S = 900 # Seit Paket IB2: die IBKR-Instanz im eigenen Betriebsordner (Unit trading-paper-ibkr.service). IBKR_ORDNER = "/mnt/d/Claude/.arbeitsbaeume/trading-ibkr-betrieb" IBKR_PY = "/home/claude/venvs/trading-ibkr/bin/python" IBKR_DIENST = "trading-paper-ibkr.service" IBKR_ZEITGEBER = "trading-paper-ibkr.timer" IBKR_MARKE = Path("data") / "paper" / "broker.txt" IBKR_TITEL = "IBKR " IBKR_LESEN_TIMEOUT_S = 180 # Journalzeilen, die ohne Ausnahme ein Befund sind: ein Traceback, und der Satz, mit dem # scripts/paper.py abbricht, wenn Instanzmarke und Konfiguration nicht zusammenpassen. IBKR_MUSTER = ("Traceback (most recent call last)", "Nichts wird gebaut") # Das Konto wird am Gateway nur ausserhalb dieses Fensters gelesen (New Yorker Zeit, werktags): # Waehrend der Handelszeit gehoerten Positionen und Orders schon zum naechsten Handelstag. NEW_YORK = ZoneInfo("America/New_York") IBKR_SPERRE_VON = (9, 0) IBKR_SPERRE_BIS = (16, 15) IBKR_POSITIONEN = re.compile(r"^Positionen: (\d+)[ \t\r]*$", re.MULTILINE) IBKR_ORDERS = re.compile(r"^Offene Orders: (\d+)[ \t\r]*$", re.MULTILINE) # IBKR-Kontonummern: Buchstaben vorn, dann Ziffern. Das Paper-Konto hat mehr als "DU" vorn; # die erste Fassung (nur DU/U) traf die echte Nummer nicht (Pruefung IB2, 2026-09-25). Dieselbe # Form wie im Projekt (trading/broker/ibkr.py). Nie in Notiz oder Meldung. KONTONUMMER = re.compile(r"\b[A-Z]{1,4}[0-9]{5,}\b") NICHT_GESTARTET = "IBKR-Instanz heute nicht gestartet" ENDE = re.compile( r"Ende: (?P[^,]+), Position (?P[^,]+), Orders (?P[^,]+), " r"Killswitch (?P[^,]+), Exit (?P-?\d+)" ) BROKER_OFFEN = re.compile(r"offene positionen beim broker jetzt: (\d+)") # Maske fuer alles, was aus fremder Ausgabe in Notiz und Meldung geht, in dieser Reihenfolge: # - jede Zahl hinter einem Geldwort, # - jede Zahl vor einer Waehrung, # - jede Zahl mit mehr als zwei Nachkommastellen, # - jede Zahl mit mindestens vier Stellen vor dem Komma und Nachkommastellen. MASKEN = ( (re.compile(r"(?i)\b(kontowert|konto|startkapital|equity|cash|preis|kurs|usd|pnl)([^0-9-]{0,3})-?[0-9][0-9.,]*"), lambda m: f"{m.group(1)}{m.group(2)}[x]"), (re.compile(r"-?\d[\d.,]*\s?(?:USD|EUR|\$|€)"), lambda m: "[x]"), (re.compile(r"-?\d[\d,]*\.\d{3,}"), lambda m: "[x]"), (re.compile(r"-?\d{4,}[.,]\d{1,2}\b"), lambda m: "[x]"), ) def maskiere(text: str) -> str: for muster, ersatz in MASKEN: text = muster.sub(ersatz, text) return text class Kontrolle: def __init__(self) -> None: self.befunde: list[str] = [] self.hinweise: list[str] = [] self.fakten: dict = {} def befund(self, text: str) -> None: self.befunde.append(maskiere(text)) def hinweis(self, text: str) -> None: self.hinweise.append(maskiere(text)) def schritt(self, name: str, aufruf, *argumente): """Eine Frage fuer sich: Scheitert sie, ist das ein Befund, und der Rest laeuft weiter.""" try: return aufruf(*argumente) except Exception as ausnahme: self.befund(f"{name} nicht geprueft ({type(ausnahme).__name__}: {str(ausnahme)[:160]})") return None # --- 1 Handelstag --------------------------------------------------------------------------------- def lies_kalender(tag: str, kalender_datei: str | None, projekt: Path): """Die Liste aus /v2/calendar oder aus einer Datei (Angriffsfaelle).""" if kalender_datei is not None: return json.loads(Path(kalender_datei).read_text(encoding="utf-8")) import requests # erst hier: Die Angriffsfaelle laufen ohne Netz sys.path.insert(0, str(projekt)) sys.path.insert(0, str(AUSWERTUNG.parent)) from handelstag_auswertung import kopfzeilen # liest die Schluessel, gibt sie nie aus kopf = kopfzeilen() if kopf is None: raise RuntimeError("Schluessel fuer Alpaca nicht lesbar") antwort = requests.get( f"{HANDEL_URL}/v2/calendar", headers=kopf, params={"start": tag, "end": tag}, timeout=30 ) antwort.raise_for_status() return antwort.json() def ist_handelstag(tag: str, liste) -> bool | None: """True, False, oder None, wenn die Antwort nicht die gemessene Form hat.""" if not isinstance(liste, list): return None for eintrag in liste: if not isinstance(eintrag, dict) or not isinstance(eintrag.get("date"), str): return None return any(eintrag["date"] == tag for eintrag in liste) # --- 2 Protokolle, 3 Ende je Lauf, 6 Handel -------------------------------------------------------- def protokolle(projekt: Path, tag: str) -> list[Path]: return sorted((projekt / "artifacts" / "paper").glob(f"{tag}T*-paper.jsonl")) def lies_protokoll(pfad: Path, k: Kontrolle) -> dict: takte = aktiv = pausiert_danach = unlesbar = 0 auto = [] ende = None # Dateiiteration statt splitlines: U+2028 in einem Text zerlegte sonst eine JSON-Zeile. with pfad.open(encoding="utf-8", errors="replace") as datei: for zeile in datei: if not zeile.strip(): continue try: satz = json.loads(zeile) except json.JSONDecodeError: unlesbar += 1 continue if not isinstance(satz, dict): unlesbar += 1 continue art = satz.get("art") if art == "takt": takte += 1 if satz.get("killswitch") == "aktiv": aktiv += 1 elif aktiv: pausiert_danach += 1 elif art == "ereignis" and satz.get("was") == "auto_fortsetzen": auto.append(str(satz.get("text"))[:200]) elif art == "ende": ende = { schluessel: satz.get(schluessel) for schluessel in ("grund", "fehler", "position", "positionen", "orders", "killswitch") } if unlesbar: k.befund(f"{pfad.name}: {unlesbar} Zeile(n) nicht lesbar") return {"datei": pfad.name, "takte": takte, "takte_aktiv": aktiv, "takte_pausiert_danach": pausiert_danach, "auto_fortsetzen": auto, "ende": ende} def ausserhalb_der_handelszeit(lauf: dict) -> bool: """Ein Lauf, der ohne einen einzigen Takt mit wartezeit endete (Start nach Boersenschluss).""" return bool(lauf["ende"]) and lauf["ende"].get("grund") == "wartezeit" and lauf["takte"] == 0 def pruefe_laeufe(laeufe: list[dict], k: Kontrolle) -> None: for lauf in laeufe: ende = lauf["ende"] if ende is None: continue # die Auswertung meldet "keine Zeile 'ende'" if ausserhalb_der_handelszeit(lauf): k.hinweis(f"{lauf['datei']}: Lauf ausserhalb der Handelszeit (Ende wartezeit), nicht bewertet") continue if ende.get("grund") != "boersenschluss": k.befund(f"{lauf['datei']}: Lauf endete mit {ende.get('grund')} nach {lauf['takte']} Takten") if ende.get("fehler"): k.befund(f"{lauf['datei']}: Lauf endete mit Fehler {str(ende.get('fehler'))[:120]}") if lauf["takte_pausiert_danach"]: k.hinweis(f"{lauf['datei']}: {lauf['takte_pausiert_danach']} Takte nach dem Fortsetzen nicht aktiv") for text in lauf["auto_fortsetzen"]: k.hinweis(f"auto_fortsetzen: {text}") bewertet = [lauf for lauf in laeufe if not ausserhalb_der_handelszeit(lauf)] if laeufe and not bewertet: k.befund("kein Lauf des scharfen Runners in der Handelszeit, nur Laeufe nach Boersenschluss") elif bewertet and sum(lauf["takte_aktiv"] for lauf in bewertet) == 0: k.befund("Tag ohne Handel: in keinem Takt stand der Killswitch auf aktiv") # --- 3 Journal ------------------------------------------------------------------------------------- def lies_journal(tag: str, journal_datei: str | None, dienst: str = "trading-paper.service") -> str: if journal_datei is not None: return Path(journal_datei).read_text(encoding="utf-8", errors="replace") morgen = (date.fromisoformat(tag) + timedelta(days=1)).isoformat() lauf = subprocess.run( ["journalctl", "-u", dienst, "--since", f"{tag} 00:00", "--until", f"{morgen} 06:00", "-o", "cat", "--no-pager"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=60, ) if lauf.returncode != 0: raise RuntimeError(f"journalctl Exit {lauf.returncode}") return lauf.stdout def pruefe_journal(text: str, k: Kontrolle, dienst: str = "trading-paper.service") -> dict | None: alle = [{schluessel: wert.strip() for schluessel, wert in m.groupdict().items()} for m in ENDE.finditer(text)] # Ein Lauf ausserhalb der Handelszeit (Neustart nach Boersenschluss) endet mit wartezeit und Exit 3. enden = [satz for satz in alle if satz["grund"] != "wartezeit"] k.fakten["enden_im_journal"] = len(enden) if len(alle) > len(enden): k.hinweis(f"{len(alle) - len(enden)} Lauf/Laeufe ausserhalb der Handelszeit im Journal (Ende wartezeit)") if "Failed with result" in text: k.befund(f"Journal: {dienst} endete mit 'Failed with result'") if not enden: k.befund(f"Journal: keine Zeile 'Ende: …, Exit n' eines Handelslaufs von {dienst} an diesem Tag") return None # Jeder Lauf des Tages zaehlt: Ein Runner, der mittags mit einer Stoerung endete und neu gestartet # wurde, ist auch dann ein Befund, wenn der zweite sauber schliesst. for nummer, satz in enumerate(enden, start=1): if satz["exit"] != "0": k.befund(f"Runner (Lauf {nummer} von {len(enden)}) endete mit Exit {satz['exit']} (Grund {satz['grund']})") if satz["grund"] != "boersenschluss": k.befund(f"Runner (Lauf {nummer} von {len(enden)}) endete nicht mit boersenschluss, sondern mit {satz['grund']}") letzte = enden[-1] if letzte["position"] != "0": k.befund(f"Runner endete mit Position {letzte['position']}") return letzte # --- 4 und 5 Auswertung ---------------------------------------------------------------------------- def fahre_auswertung(pfad: Path, projekt: Path, broker: bool, kerzen: bool, k: Kontrolle) -> None: befehl = [sys.executable, AUSWERTUNG_ERSATZ or str(AUSWERTUNG), str(pfad)] if broker: befehl.append("--broker") if kerzen: befehl.append("--kerzen") lauf = subprocess.run( befehl, cwd=projekt, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=AUSWERTUNG_TIMEOUT_S, env={**os.environ, "PYTHONIOENCODING": "utf-8"}, ) zeilen = lauf.stdout.splitlines() + lauf.stderr.splitlines() for zeile in zeilen: if zeile.strip().startswith("BEFUND") or zeile.startswith("VORAUSSETZUNG FEHLT"): k.befund(f"{pfad.name}: {zeile.strip()}") if lauf.returncode != 0: k.befund(f"{pfad.name}: handelstag_auswertung.py endete mit Exit {lauf.returncode}") if broker: offen = [int(m.group(1)) for m in map(BROKER_OFFEN.search, zeilen) if m] if not offen: k.befund(f"{pfad.name}: keine Zeile zu offenen Positionen beim Broker -- Konto nicht geprueft") else: k.fakten["broker_offen"] = offen[-1] if offen[-1] != 0: k.befund(f"Beim Broker stehen {offen[-1]} offene Position(en)") # --- 8 Dienste und 9 Killswitch ------------------------------------------------------------------- def lies_dienste(dienste_datei: str | None) -> dict[str, str]: if dienste_datei is not None: return json.loads(Path(dienste_datei).read_text(encoding="utf-8")) def lesen(): ergebnis = {} for dienst in DIENSTE: lauf = subprocess.run(["systemctl", "is-active", dienst], capture_output=True, text=True, timeout=30) ergebnis[dienst] = lauf.stdout.strip() or f"exit {lauf.returncode}" return ergebnis zustaende = lesen() if any(wert != "active" for wert in zustaende.values()): time.sleep(DIENSTE_NOCHMAL_S) zustaende = lesen() return zustaende def pruefe_dienste(zustaende, k: Kontrolle) -> None: if not isinstance(zustaende, dict): k.befund("Dienste: Zustand nicht lesbar") return k.fakten["dienste"] = zustaende for dienst in DIENSTE: if zustaende.get(dienst) != "active": k.befund(f"{dienst} ist nicht aktiv ({zustaende.get(dienst)})") def pruefe_killswitch(projekt: Path, k: Kontrolle) -> None: register = projekt / "data" / "registry.db" if not register.is_file(): k.befund("Register data/registry.db fehlt") return with sqlite3.connect(f"file:{register}?mode=ro", uri=True, timeout=10) as verbindung: zeile = verbindung.execute( "SELECT zustand, seit, ausloeser FROM killswitch_zustand WHERE id = 1" ).fetchone() if zeile is None: k.befund("Register: kein Killswitch-Zustand") return k.fakten["killswitch"] = {"zustand": zeile[0], "seit": zeile[1], "ausloeser": zeile[2]} if zeile[0] == "gesperrt": k.befund("Killswitch steht auf gesperrt: Ohne 'Freigeben' handelt der naechste Tag nicht") elif zeile[0] not in KILLSWITCH_ZUSTAENDE: k.befund(f"Killswitch in unbekanntem Zustand {zeile[0]!r}") # --- IBKR-Instanz (seit Paket IB2) ----------------------------------------------------------------- def pruefe_marke_ibkr(ordner: Path, k: Kontrolle) -> None: datei = ordner / IBKR_MARKE marke = datei.read_text(encoding="utf-8").strip() if datei.is_file() else None k.fakten["marke"] = marke if marke != "ibkr": k.befund( f"Instanzmarke {IBKR_MARKE.as_posix()} sagt {marke!r} statt 'ibkr': " "Das ist nicht der Betriebsordner der IBKR-Instanz" ) def pruefe_register_ibkr(ordner: Path, k: Kontrolle) -> None: """Zustand und juengstes Ereignis aus dem Register der Instanz, nur lesend (mode=ro).""" register = ordner / "data" / "registry.db" if not register.is_file(): k.befund("IBKR-Register data/registry.db fehlt im Betriebsordner") return with sqlite3.connect(f"file:{register}?mode=ro", uri=True, timeout=10) as verbindung: stand = verbindung.execute( "SELECT zustand, seit, ausloeser FROM killswitch_zustand WHERE id = 1" ).fetchone() letztes = verbindung.execute( "SELECT stufe, ausloeser FROM killswitch_ereignisse ORDER BY id DESC LIMIT 1" ).fetchone() if stand is None: k.befund("IBKR-Register: kein Killswitch-Zustand") return zustand = stand[0] k.fakten["killswitch"] = {"zustand": zustand, "seit": stand[1], "ausloeser": stand[2]} if zustand == "gesperrt": grund = "ohne Ereignis im Register" if letztes is None else f"letztes Ereignis {letztes[0]} von {letztes[1]}" k.befund(f"IBKR-Killswitch steht auf gesperrt ({grund}): Ohne Freigeben an der Konsole handelt die Instanz nicht") elif zustand not in KILLSWITCH_ZUSTAENDE: k.befund(f"IBKR-Killswitch in unbekanntem Zustand {zustand!r}") def pruefe_schlusspositionen(laeufe: list[dict], k: Kontrolle) -> None: """Die Schlussposition jedes Laufs aus seiner Zeile ende, unabhaengig vom Journal. pruefe_laeufe bewertet die Position nicht, und pruefe_journal nur die des letzten Laufs; der Alpaca-Teil faengt sie ueber handelstag_auswertung.py, das der Teil ibkr nicht faehrt. Ebenso einen Lauf ohne Zeile ende: pruefe_laeufe ueberspringt ihn, weil im Alpaca-Teil die Auswertung ihn meldet (Nachkritik IB2). """ for lauf in laeufe: ende = lauf["ende"] if ende is None: # Im Alpaca-Teil meldet das handelstag_auswertung.py; hier sonst niemand. k.befund(f"{lauf['datei']}: Lauf ohne Zeile ende im Protokoll") elif ende.get("position") != 0: k.befund(f"{lauf['datei']}: Lauf endete laut Protokoll mit Position {ende.get('position')}") def lies_zeitgeber_ibkr(dienste_datei: str | None) -> str: """Der Zustand von trading-paper-ibkr.timer (active, inactive, ...); aus einer Datei in den Angriffsfaellen. Gelesen wird der Zustand jetzt, nicht der am geprueften Tag: Beim Nachholen kann er sich inzwischen geaendert haben (benannte Luecke). """ if dienste_datei is not None: zustaende = json.loads(Path(dienste_datei).read_text(encoding="utf-8")) if not isinstance(zustaende, dict): raise ValueError("Dienste-Datei ist kein Objekt") return str(zustaende.get(IBKR_ZEITGEBER, "fehlt")) lauf = subprocess.run(["systemctl", "is-active", IBKR_ZEITGEBER], capture_output=True, text=True, timeout=30) return lauf.stdout.strip() or f"exit {lauf.returncode}" def lesen_erlaubt(jetzt: datetime) -> bool: """Ob der Stand am Gateway jetzt fuer den Tagesschluss spricht: Wochenende, oder werktags ausserhalb IBKR_SPERRE_VON bis IBKR_SPERRE_BIS New Yorker Zeit.""" ny = jetzt.astimezone(NEW_YORK) if ny.weekday() >= 5: return True return not IBKR_SPERRE_VON <= (ny.hour, ny.minute) < IBKR_SPERRE_BIS def lies_konto_ibkr(ordner: Path, konto_datei: str | None, python: str = IBKR_PY) -> tuple[int, str]: """Exit und Ausgabe (stdout und stderr) von scripts/ibkr_pruefen.py lesen, oder aus einer Datei. python ist der Interpreter dafuer: in WSL das Venv der Instanz (IBKR_PY), auf dem Docker-Host das python des Abbilds (--ibkr-python, seit dem Nachtrag zu Paket DH). """ if konto_datei is not None: daten = json.loads(Path(konto_datei).read_text(encoding="utf-8")) return int(daten["exit"]), str(daten["text"]) lauf = subprocess.run( [python, "scripts/ibkr_pruefen.py", "lesen"], cwd=ordner, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=IBKR_LESEN_TIMEOUT_S, env={**os.environ, "PYTHONIOENCODING": "utf-8"}, ) return lauf.returncode, lauf.stdout + "\n" + lauf.stderr def ohne_kontonummer(text: str) -> str: # Erst die Kontonummer, dann die Maske: Sonst machte die Maske aus "Konto DU..." ein # "Konto DU[x]", und die Kontonummer-Regel faende nichts mehr. return maskiere(KONTONUMMER.sub("[konto]", text)) def pruefe_konto_ibkr(code: int, text: str, k: Kontrolle) -> None: """Positionen und offene Orders am Gateway; aus der Ausgabe gehen nur Zahlen und Befunde weiter.""" treffer_positionen = IBKR_POSITIONEN.search(text) treffer_orders = IBKR_ORDERS.search(text) positionen = int(treffer_positionen.group(1)) if treffer_positionen else None orders = int(treffer_orders.group(1)) if treffer_orders else None k.fakten["konto"] = {"positionen": positionen, "orders": orders, "exit": code} for zeile in text.splitlines(): if zeile.strip().startswith(("BEFUND", "Voraussetzung fehlt")): k.befund(f"ibkr_pruefen.py lesen: {ohne_kontonummer(zeile.strip())[:200]}") if code != 0: k.befund(f"ibkr_pruefen.py lesen endete mit Exit {code}") if positionen is None or orders is None: k.befund("Konto am Gateway nicht gelesen: keine Zeile 'Positionen: n' oder 'Offene Orders: n'") return if positionen != 0: k.befund(f"Beim IBKR-Gateway stehen {positionen} Position(en)") if orders != 0: k.befund(f"Beim IBKR-Gateway stehen {orders} offene Order(s)") def kontrolliere_ibkr(argumente: argparse.Namespace, k: Kontrolle) -> None: ordner = Path(argumente.ibkr_ordner) tag = argumente.tag date.fromisoformat(tag) # ein falsches Datum ist ein Bedienfehler und bricht ab k.fakten["ordner"] = str(ordner) if not ordner.is_dir(): k.befund(f"Betriebsordner der IBKR-Instanz fehlt: {ordner}") return liste = k.schritt("Kalender, und damit Lauf, Ende und Konto,", lies_kalender, tag, argumente.kalender, Path(argumente.projekt)) handelstag = None if liste is None else ist_handelstag(tag, liste) k.fakten["handelstag"] = handelstag if handelstag is None: if liste is not None: k.befund("Kalender in unerwarteter Form; Lauf, Ende und Konto der IBKR-Instanz ungeprueft") return dateien = protokolle(ordner, tag) if not handelstag: if dateien: k.hinweis(f"kein Handelstag, aber {len(dateien)} Protokoll(e) der IBKR-Instanz") return zeitgeber = k.schritt("Zeitgeber der IBKR-Instanz", lies_zeitgeber_ibkr, argumente.dienste) k.fakten["zeitgeber"] = zeitgeber # No-Go-Tag: Pascal hat den Zeitgeber abgeschaltet, und die Instanz lief nicht. Leise. if not dateien and zeitgeber is not None and zeitgeber != "active": k.fakten["nicht_gestartet"] = True k.hinweis(f"{NICHT_GESTARTET} ({IBKR_ZEITGEBER}: {zeitgeber})") return k.schritt("Instanzmarke", pruefe_marke_ibkr, ordner, k) k.schritt("IBKR-Killswitch", pruefe_register_ibkr, ordner, k) if not dateien: k.befund(f"kein Protokoll der IBKR-Instanz an einem Handelstag, obwohl {IBKR_ZEITGEBER} " f"{zeitgeber} ist -- die Instanz lief nicht") laeufe = [lauf for lauf in (k.schritt(f"Protokoll {pfad.name}", lies_protokoll, pfad, k) for pfad in dateien) if lauf] k.fakten["laeufe"] = laeufe pruefe_laeufe(laeufe, k) pruefe_schlusspositionen(laeufe, k) text = k.schritt("Journal", lies_journal, tag, argumente.journal, IBKR_DIENST) if text is not None: k.fakten["ende"] = pruefe_journal(text, k, IBKR_DIENST) for muster in IBKR_MUSTER: if muster in text: k.befund(f"Journal: {IBKR_DIENST} zeigt '{muster}'") jetzt = datetime.fromisoformat(argumente.jetzt) if argumente.jetzt else datetime.now(NEW_YORK) if not lesen_erlaubt(jetzt): k.hinweis(f"Konto am IBKR-Gateway nicht gelesen: {jetzt.astimezone(NEW_YORK):%H:%M} New Yorker Zeit " "liegt im Handelsfenster") return ergebnis = k.schritt("Konto am IBKR-Gateway", lies_konto_ibkr, ordner, argumente.ibkr_konto, argumente.ibkr_python) if ergebnis is not None: pruefe_konto_ibkr(ergebnis[0], ergebnis[1], k) # --- Sicherung des vorigen Handelstags (seit Paket E9) --------------------------------------------- # Wie weit der Kalender zurueck gefragt wird. Projektvorgabe (Plan E9, Task 726), nicht am Kalender # gemessen: Zehn Kalendertage decken ein langes Wochenende mit Feiertag und eine kurze # ausserplanmaessige Schliessung. SICHERUNG_FENSTER_TAGE = 10 # Summe und Bestaetigung auf dem Host (Schnittstelle Paket E9) und ein Datum, nur ASCII-Ziffern. SUMME = re.compile(r"[0-9a-f]{64}") DATUM = re.compile(r"[0-9]{4}-[0-9]{2}-[0-9]{2}") def in_spanne(liste, start: str, ende: str): """Aus einer Liste des Kalenders nur die Eintraege, deren date von start bis ende reicht. Es faellt nur ein Eintrag mit einem date JJJJ-MM-TT ausserhalb der Spanne weg; jeder andere bleibt, und eine Antwort, die keine Liste ist, kommt unveraendert zurueck. Die Form prueft juengster_handelstag. """ if not isinstance(liste, list): return liste return [ eintrag for eintrag in liste if not isinstance(eintrag, dict) or not isinstance(eintrag.get("date"), str) or not DATUM.fullmatch(eintrag["date"]) or start <= eintrag["date"] <= ende ] def lies_kalender_spanne(start: str, ende: str, kalender_datei: str | None, projekt: Path): """Die Liste aus /v2/calendar oder aus einer Datei (Angriffsfaelle), je nur von start bis ende. Die Spanne setzt das Werkzeug fuer beide Quellen selbst durch (Plan E9, Plankritik Runde 1): So pinnen die Angriffsfaelle mit Kalenderdatei die Grenzen, mit denen es Alpaca fragt. """ if kalender_datei is not None: return in_spanne(json.loads(Path(kalender_datei).read_text(encoding="utf-8")), start, ende) import requests # erst hier: Die Angriffsfaelle laufen ohne Netz sys.path.insert(0, str(projekt)) sys.path.insert(0, str(AUSWERTUNG.parent)) from handelstag_auswertung import kopfzeilen # liest die Schluessel, gibt sie nie aus kopf = kopfzeilen() if kopf is None: raise RuntimeError("Schluessel fuer Alpaca nicht lesbar") antwort = requests.get( f"{HANDEL_URL}/v2/calendar", headers=kopf, params={"start": start, "end": ende}, timeout=30 ) antwort.raise_for_status() return in_spanne(antwort.json(), start, ende) def juengster_handelstag(liste) -> str | None: """Der juengste Handelstag in der Liste, None, wenn keiner darin steht. Die Liste reicht schon nur bis zum Tag vor dem geprueften (lies_kalender_spanne). Die Grenze steht allein dort, damit ein Angriffsfall sie pinnen kann. Die Reihenfolge der Liste zaehlt nicht. Hat die Liste nicht die gemessene Form (Frage 1) oder ein date nicht die Form JJJJ-MM-TT, ist das ein ValueError. """ if not isinstance(liste, list): raise ValueError("Kalender ist keine Liste") tage = [] for eintrag in liste: datum = eintrag.get("date") if isinstance(eintrag, dict) else None if not isinstance(datum, str) or not DATUM.fullmatch(datum): raise ValueError("Kalender mit einem Eintrag ohne date JJJJ-MM-TT") tage.append(datum) return max(tage) if tage else None def pruefe_sicherung(argumente: argparse.Namespace, k: Kontrolle) -> None: """Die Sicherung des vorigen Handelstags fuer den geprueften Teil (Paket E9, S6). Liest nur die Datei aus --sicherung, nie die Ablage. Jede Abweichung ist ein Befund, auch ein fehlender erster Sicherungstag, einer nach dem geprueften Tag und ein fehlendes Archiv. Still bleibt nur ein voriger Handelstag vor dem ersten Sicherungstag. """ geprueft = date.fromisoformat(argumente.tag) # ein falsches Datum ist ein Bedienfehler tag = geprueft.isoformat() instanz = "ibkr" if argumente.teil == "ibkr" else "alpaca" stand = json.loads(Path(argumente.sicherung).read_text(encoding="utf-8")) if not isinstance(stand, dict): k.befund("Sicherung: Stand ist kein Objekt") return if "fehler" in stand: k.befund(f"Sicherung nicht geprueft: {str(stand['fehler'])[:160]}") return if "erster_tag" not in stand or not isinstance(stand.get("eintraege"), dict): k.befund("Sicherung: Stand in unerwarteter Form (erster_tag oder eintraege fehlt)") return erster = stand["erster_tag"] if erster is None: # Seit der Plankritik Runde 1 ein Befund: Ohne erster-tag schwiege die Pruefung auf Dauer. k.befund("Sicherung nicht geprueft: auf dem Docker-Host fehlt erster-tag der Sicherung") return if not isinstance(erster, str) or not DATUM.fullmatch(erster): k.befund("Sicherung: erster_tag ist kein Datum JJJJ-MM-TT") return # Ein erster Sicherungstag nach dem geprueften Tag (etwa ein Tippfehler im Jahr) schaltete die # Pruefung sonst still ab, bis der Tag kommt. DATUM hat die Form gesichert, also vergleicht der # Text wie das Datum. if erster > tag: k.befund(f"Sicherung nicht geprueft: erster_tag {erster} liegt nach dem geprueften Tag {tag}") return start = (geprueft - timedelta(days=SICHERUNG_FENSTER_TAGE)).isoformat() ende = (geprueft - timedelta(days=1)).isoformat() vorher = juengster_handelstag(lies_kalender_spanne(start, ende, argumente.kalender, Path(argumente.projekt))) if vorher is None: k.befund(f"Sicherung: kein Handelstag in den {SICHERUNG_FENSTER_TAGE} Tagen vor {tag} im Kalender") return name = f"{vorher}-{instanz}" k.fakten["sicherung"] = {"name": name, "erster_tag": erster, "geprueft": False} # Still bleibt nur das: ein voriger Handelstag vor dem ersten Sicherungstag (die ersten Tage nach # dem Einspielen). if vorher < erster: return k.fakten["sicherung"]["geprueft"] = True eintrag = stand["eintraege"].get(name) if not isinstance(eintrag, dict): k.befund(f"Sicherung {name} fehlt auf dem Docker-Host") return summe = eintrag.get("summe") bestaetigt = eintrag.get("bestaetigt") if not isinstance(summe, str) or not SUMME.fullmatch(summe): k.befund(f"Sicherung {name}: keine gueltige Summe auf dem Docker-Host") elif eintrag.get("archiv") is not True: k.befund(f"Sicherung {name}: Archiv fehlt auf dem Docker-Host") elif not isinstance(bestaetigt, str) or not SUMME.fullmatch(bestaetigt): k.befund(f"Sicherung {name}: nicht bestaetigt, also nicht nach D: geholt") elif bestaetigt != summe: k.befund(f"Sicherung {name}: die Bestaetigung passt nicht zur Summe auf dem Docker-Host") # --- Meldung --------------------------------------------------------------------------------------- def melde(titel: str, text: str, projekt: Path) -> str: """Eine Meldung mit hoher Prioritaet ueber den Melder des Projekts. Scheitert nie nach aussen. Melder.melde wirft nicht: Es gibt None bei Erfolg und sonst den Grund zurueck (Typ der Ausnahme oder "HTTP "). Genau das steht im Ergebnis. """ try: sys.path.insert(0, str(projekt)) os.chdir(projekt) from trading.paper.melder import baue_melder from trading.paper.meldungen import HOCH from trading.paper.meldungen import Meldung melder = baue_melder(config_pfad="config/paper.toml") if not melder.aktiv: return "nicht gesendet: Melder nicht aktiv" fehler = melder.melde(Meldung(titel=titel, text=maskiere(text), prioritaet=HOCH, tags="warning")) return "gesendet" if fehler is None else f"nicht gesendet: {fehler}" except Exception as ausnahme: # die Kontrolle haengt nicht an der Meldung return f"nicht gesendet: {type(ausnahme).__name__}" # --- Ablauf ---------------------------------------------------------------------------------------- def kontrolliere(argumente: argparse.Namespace, k: Kontrolle) -> None: projekt = Path(argumente.projekt) tag = argumente.tag date.fromisoformat(tag) # ein falsches Datum ist ein Bedienfehler und bricht ab # handelstag_auswertung.py und der Melder lesen aus dem Arbeitsverzeichnis (Projektordner). os.chdir(projekt) zustaende = k.schritt("Dienste", lies_dienste, argumente.dienste) if zustaende is not None: pruefe_dienste(zustaende, k) k.schritt("Killswitch", pruefe_killswitch, projekt, k) liste = k.schritt("Kalender, und damit Lauf, Ende und Bericht,", lies_kalender, tag, argumente.kalender, projekt) handelstag = None if liste is None else ist_handelstag(tag, liste) k.fakten["handelstag"] = handelstag if handelstag is None: if liste is not None: k.befund("Kalender in unerwarteter Form; Lauf, Ende und Bericht ungeprueft") return dateien = protokolle(projekt, tag) if not handelstag: if dateien: k.hinweis(f"kein Handelstag, aber {len(dateien)} Protokoll(e) des scharfen Runners") return if not dateien: k.befund("kein Protokoll des scharfen Runners an einem Handelstag -- der Runner lief nicht") laeufe = [lauf for lauf in (k.schritt(f"Protokoll {pfad.name}", lies_protokoll, pfad, k) for pfad in dateien) if lauf] k.fakten["laeufe"] = laeufe pruefe_laeufe(laeufe, k) text = k.schritt("Journal", lies_journal, tag, argumente.journal) if text is not None: k.fakten["ende"] = pruefe_journal(text, k) ausserhalb = {lauf["datei"] for lauf in laeufe if ausserhalb_der_handelszeit(lauf)} for pfad in dateien: if pfad.name in ausserhalb: continue k.schritt(f"Auswertung {pfad.name}", fahre_auswertung, pfad, projekt, not argumente.ohne_broker, not argumente.ohne_kerzen, k) bericht = projekt / "data" / "paper" / "auswertungen" / f"{tag}.md" k.fakten["bericht"] = bericht.is_file() if not bericht.is_file(): k.befund(f"Tagesbericht data/paper/auswertungen/{tag}.md fehlt") def parse_args(argv=None) -> argparse.Namespace: zerleger = argparse.ArgumentParser(description=__doc__.splitlines()[0], allow_abbrev=False) zerleger.add_argument("--tag") zerleger.add_argument("--meldung", nargs=2, metavar=("TITEL", "TEXT")) zerleger.add_argument("--projekt", default=PROJEKT) zerleger.add_argument("--kalender") zerleger.add_argument("--journal") zerleger.add_argument("--dienste") zerleger.add_argument("--ohne-broker", action="store_true") zerleger.add_argument("--ohne-kerzen", action="store_true") zerleger.add_argument("--ohne-meldung", action="store_true") # Seit Paket IB2: der Teil ibkr prueft die IBKR-Instanz statt des Alpaca-Runners. zerleger.add_argument("--teil", choices=("alpaca", "ibkr"), default="alpaca") zerleger.add_argument("--ibkr-ordner", default=IBKR_ORDNER) # Seit dem Nachtrag zu Paket DH (IB2): der Interpreter fuer ibkr_pruefen.py lesen. zerleger.add_argument("--ibkr-python", default=IBKR_PY) # Nur fuer die Angriffsfaelle: Exit und Ausgabe von ibkr_pruefen.py als JSON-Datei, und die Uhr. zerleger.add_argument("--ibkr-konto") zerleger.add_argument("--jetzt", help="ISO-Zeitpunkt mit Versatz statt der Uhr") # Seit Paket E9 (S6): der Stand der Sicherungsablage des Hosts als JSON-Datei (abendkontrolle-host.sh). zerleger.add_argument("--sicherung") argumente = zerleger.parse_args(argv) if bool(argumente.tag) == bool(argumente.meldung): zerleger.error("genau eines von --tag und --meldung") return argumente def main(argv=None) -> int: os.environ.setdefault("TRADING_GEHEIMNISSE", GEHEIMNISSE) argumente = parse_args(argv) if argumente.meldung: print(json.dumps({"meldung": melde(*argumente.meldung, Path(argumente.projekt))}, ensure_ascii=False)) return 0 k = Kontrolle() code = 0 ibkr = argumente.teil == "ibkr" try: if ibkr: kontrolliere_ibkr(argumente, k) else: kontrolliere(argumente, k) except Exception as ausnahme: k.befund(f"Abendkontrolle abgebrochen: {type(ausnahme).__name__}: {ausnahme}") code = 2 # Seit Paket E9 (S6): die Sicherung des vorigen Handelstags, fuer sich und vor der Meldung. if argumente.sicherung: k.schritt("Sicherung", pruefe_sicherung, argumente, k) if k.befunde and not argumente.ohne_meldung: text = "\n".join(f"- {befund}" for befund in k.befunde[:12]) titel = f"Abendkontrolle {argumente.tag}: {len(k.befunde)} Befund(e)" k.fakten["meldung"] = melde(IBKR_TITEL + titel if ibkr else titel, text, Path(argumente.projekt)) print(json.dumps( {"tag": argumente.tag, "teil": argumente.teil, "befunde": k.befunde, "hinweise": k.hinweise, "fakten": k.fakten, "exit": code}, ensure_ascii=False, default=str, )) return code if __name__ == "__main__": sys.exit(main())