claude-regelwerk/.vault-tools/trading-plattform-pruefung/abendkontrolle.py

"""Startet die Abendkontrolle des Paper-Betriebs und schreibt ihr Ergebnis in die Tagesnotiz.

Koordinator-Werkzeug, nicht fuers Repo. Laeuft unter Windows mit pythonw.exe als geplante Aufgabe
"Trading Plattform - Abendkontrolle" (abendkontrolle_einrichten.ps1): werktags 22:45 (kommt ein Teil
vom Docker-Host, 23:15; seit Paket DH) und fuenf Minuten nach jeder Anmeldung. Auftrag Pascal
2026-09-24: dauerhaft, uebersteht Neustarts, blockiert keine
Updates -- deshalb eine Windows-Aufgabe (bleibt ueber Neustarts, holt verpasste Laeufe nach) und kein
Claude-Prozess (haelt keine Datei von Claude Code oder Desktop offen).

Ablauf: Jeder Werktag seit ERSTER_TAG, hoechstens NACHHOLEN_TAGE zurueck, ohne Ergebnisdatei in
ABLAGE wird geprueft; der heutige erst ab FRUEHESTENS (der Bericht der Tagesauswertung kommt 22:20).
Je Tag: abendkontrolle_wsl.py in WSL (liest nur, meldet bei Befund selbst ueber ntfy), dann zuerst das
Ergebnis als JSON nach ABLAGE und danach eine Zeile in Inbox/<tag>.md unter der ersten Ueberschrift,
die mit "## Agentenprotokoll" beginnt.
- Liefert der WSL-Teil kein Ergebnis oder endet er mit Exit 2 (abgebrochen), bleibt die Ergebnisdatei
  aus, die Notiz bekommt einen Befund mit #needs-review, und der naechste Lauf versucht den Tag erneut.
- Scheitert nur die Notiz (etwa eine Datei, die kein UTF-8 ist), bleibt es bei der Ergebnisdatei und
  einer Zeile in lauf.log: Die Kontrolle ist gelaufen und hat ihre Meldung schon geschickt; ein zweiter
  Lauf schickte sie nur noch einmal.
- Ein Tag, der laenger als NACHHOLEN_TAGE scheitert, faellt aus dem Fenster (lauf.log behaelt ihn).
Seit Paket IB2 zwei Teile je Tag, nacheinander: alpaca wie oben, dann ibkr (abendkontrolle_wsl.py
--teil ibkr) mit eigener Ergebnisdatei <tag>-ibkr.json, eigener Zeile "Abendkontrolle IBKR" in
derselben Notiz und eigenem Nachholen ab IBKR_ERSTER_TAG. --teil nimmt nur einen Teil; mit
--ohne-wsl gilt ohne --teil nur alpaca (die Angriffsfaelle von vor IB2 bleiben gleich). Ein
Ergebnis, das zu einem anderen Teil gehoert, gilt als gescheitert.
Die Notiz wird ueber eine Nachbardatei und os.replace geschrieben: Ein Abbruch mitten im Schreiben
hinterlaesst die alte Datei, nicht eine halbe. Gegen eine Claude-Sitzung, die dieselbe Notiz im selben
Augenblick schreibt, schuetzt das nicht (Zeitfenster Lesen bis Ersetzen, Millisekunden) -- benannt.

Annahme, benannt: Die Uhr dieses Rechners geht nach Berliner Zeit (tzutil: W. Europe Standard Time,
gelesen am 2026-09-24); ohne das Paket tzdata kennt Python unter Windows keine Zeitzonen-Datenbank.
Seit Paket E9 holt die Aufgabe fuer jeden Teil mit Quelle dockerhost auch die Sicherung des Hosts ab
(hole_sicherungen): je Kalendertag der letzten SICHERUNG_NACHHOLEN_TAGE die Summe des Hosts fragen; wo
die Summendatei auf D: fehlt oder abweicht, <tag>-<instanz>.tar ueber denselben Schluessel holen, Summe
pruefen, nach SICHERUNG_ZIEL/<tag>/<instanz>.tar umbenennen, beim Host bestaetigen, zuletzt
<instanz>.tar.sha256 schreiben; eine Zeile in der Notiz je Tag mit Neuem. Solange SICHERUNG_ERSTER_TAG
None ist, holt sie nichts. Den Abend-Ausloeser legt abendkontrolle_einrichten.ps1 in Ortszeit an (S10).
Seit Paket NT (Quorum 4): Endet die Frage nach einer Summe oder die Bestaetigung mit ssh-Exit 255 oder
im Zeitlimit, ist der Docker-Host nicht erreichbar, und die Sicherung hoert sofort auf: eine Zeile in der
Notiz von heute, Exit 1. Bricht nur das Archiv so ab, scheitert allein diese Instanz. --tag ausser der
Form JJJJ-MM-TT oder ohne gueltiges Datum endet gleich nach dem Zerlegen mit Exit 2, ohne Notiz und ssh.
Aufruf von Hand: python abendkontrolle.py [--tag JJJJ-MM-TT] [--vault D:\\Claude] [--ohne-wsl <json-datei>]
                 [--quelle wsl|dockerhost] [--quelle-ibkr wsl|dockerhost] [--sicherung-ziel <ordner>]
"""

import argparse
import hashlib
import json
import os
import re
import subprocess
import sys
import traceback
from datetime import date
from datetime import datetime
from datetime import time
from datetime import timedelta
from pathlib import Path
from time import monotonic

VAULT = Path(r"D:\Claude")
PROJEKT_TEIL = Path("Code Projekte") / "active - Trading Plattform"
ABLAGE_TEIL = PROJEKT_TEIL / "data" / "paper" / "abendkontrollen"
WSL = r"C:\Windows\System32\wsl.exe"
WSL_PY = "/home/claude/venvs/trading/bin/python"
WSL_SKRIPT = "/mnt/d/Claude/.vault-tools/trading-plattform-pruefung/abendkontrolle_wsl.py"
ERSTER_TAG = date(2026, 9, 24)
NACHHOLEN_TAGE = 7
FRUEHESTENS = time(22, 30)
WSL_TIMEOUT_S = 1200
KEIN_FENSTER = 0x08000000  # CREATE_NO_WINDOW: pythonw hat keine Konsole, wsl.exe soll keine oeffnen
UEBERSCHRIFT = "## Agentenprotokoll"
# Seit Paket DH (Trading Plattform 0.38.0): Nach dem Umzug auf den Docker-Host prueft der Host
# selbst (abendkontrolle-host.sh, werktags 16:30 New Yorker Zeit, beide Teile); diese Aufgabe
# liest nur noch sein Ergebnis ueber SSH. Der Schluessel darf dort genau einen Befehl
# (abendkontrolle-lesen.sh mit JJJJ-MM-TT oder JJJJ-MM-TT-ibkr), der Host ist ueber seinen
# Fingerabdruck festgelegt. Umgeschaltet wird je Teil: --quelle fuer alpaca, --quelle-ibkr fuer
# ibkr (ohne Angabe wie --quelle), weil die IBKR-Instanz an einem eigenen Abend umzieht.
QUELLEN = ("wsl", "dockerhost")
SSH = r"C:\Windows\System32\OpenSSH\ssh.exe"
SSH_ZIEL = "[email protected]"
SSH_PORT = "2498"
SSH_SCHLUESSEL = Path.home() / ".ssh" / "trading-dh-kontrolle"
# Ohne Leerzeichen im Wert: ssh trennt UserKnownHostsFile an Leerraum; ~ setzt ssh selbst ein.
SSH_BEKANNTE = "~/.ssh/known_hosts_trading_dh"
SSH_TIMEOUT_S = 120
# Seit Paket IB2: die IBKR-Instanz als zweiter Teil mit eigener Ergebnisdatei und eigener Zeile.
TEILE = ("alpaca", "ibkr")
# Der erste Tag, den der Teil ibkr prueft; gesetzt beim Einspielen von IB2 (Planuebersicht IB2).
IBKR_ERSTER_TAG = date(2026, 9, 25)
IBKR_ENDUNG = "-ibkr"
KOEPFE = {
    "alpaca": "Abendkontrolle Trading Plattform (automatisch)",
    "ibkr": "Abendkontrolle IBKR (automatisch)",
}


# Seit Paket E9 (Trading Plattform 0.39.0): Der Host sichert werktags 16:45 New Yorker Zeit je
# Instanz Register, data/paper und artifacts/paper (sicherung-host.sh). Diese Aufgabe holt fuer
# jeden Teil mit Quelle dockerhost die Sicherung ueber denselben Schluessel ab (abendkontrolle-lesen.sh
# mit sicherung-summe, sicherung und sicherung-bestaetigt), prueft die Summe und legt sie unter
# SICHERUNG_ZIEL/<tag>/<instanz>.tar samt <instanz>.tar.sha256 ab. Massstab ist die aktuelle Summe des
# Hosts: Fehlt die Summendatei auf D: oder weicht sie ab, holt der Lauf neu und ersetzt die alte Fassung
# (Plankritik Runde 1, Befunde 5 und 24). Sonst loescht er auf D: nichts.
SICHERUNG_ZIEL = Path(r"D:\Trading-Sicherung")
# Der erste Tag, den die Aufgabe abholt; None holt nichts. Gesetzt beim Einspielen von E9 (S8): Diese
# Datei laeuft schon vorher jeden Abend, der Host kennt die Lesebefehle erst danach.
SICHERUNG_ERSTER_TAG: date | None = None
# So viele Kalendertage vor heute fragt jeder Lauf, dazu heute. Das Fenster liegt in der Aufbewahrung
# des Hosts: Er haelt die neuesten zehn Sicherungstage, nur werktags gesichert, also gut zwei Kalenderwochen.
# Die Sicherung entsteht 16:45 New Yorker Zeit, in Berlin meist 22:45, vom 25.10. bis 31.10.2026
# und vom 14.3. bis 27.3.2027 schon 21:45. Die Aufgabe startet 23:15 Ortszeit (S10); fehlt die
# Sicherung von heute trotzdem noch, ist das kein Fehlschlag.
SICHERUNG_NACHHOLEN_TAGE = 10
# Je Archiv; Summe und Bestaetigung nehmen SSH_TIMEOUT_S. Die Aufgabe selbst endet nach 45 Minuten.
SICHERUNG_TIMEOUT_S = 600
# Zeitbudget je Lauf fuer die Sicherung, geprueft vor jeder Instanz: Ist es um, faengt der Lauf nichts
# mehr an, und der naechste holt den Rest (Plankritik Runde 1, Befund 12).
SICHERUNG_BUDGET_S = 1500
SICHERUNG_KOPF = "Sicherung Trading Plattform (automatisch)"
# Eine Zeile im Format von sha256sum: 64 Hexzeichen in Kleinbuchstaben, zwei Leerzeichen, Name, LF.
SUMMENZEILE = re.compile(r"([0-9a-f]{64})  (\S+)\n")
TAGFORM = re.compile(r"[0-9]{4}-[0-9]{2}-[0-9]{2}")


def tag_gueltig(tag: str) -> bool:
    """Ist tag ein Datum JJJJ-MM-TT, das es gibt, nur mit ASCII-Ziffern? Seit Paket NT fuer --tag in main.

    TAGFORM allein liesse 2026-13-05 durch, date.fromisoformat allein seit Python 3.11 auch 20261005.
    """
    if not TAGFORM.fullmatch(tag):
        return False
    try:
        date.fromisoformat(tag)
    except ValueError:
        return False
    return True


def faellige_tage(jetzt: datetime, ablage: Path, erster_tag: date = ERSTER_TAG, endung: str = "") -> list[str]:
    tage = []
    for zurueck in range(NACHHOLEN_TAGE, -1, -1):
        tag = jetzt.date() - timedelta(days=zurueck)
        if tag < erster_tag or tag.weekday() >= 5:
            continue
        if tag == jetzt.date() and jetzt.time() < FRUEHESTENS:
            continue
        if (ablage / f"{tag.isoformat()}{endung}.json").exists():
            continue
        tage.append(tag.isoformat())
    return tage


def fahre_wsl(tag: str, teil: str = "alpaca") -> dict:
    lauf = subprocess.run(
        [WSL, "-d", "Debian", "-u", "claude", "--", WSL_PY, WSL_SKRIPT, "--tag", tag, "--teil", teil],
        capture_output=True, timeout=WSL_TIMEOUT_S, creationflags=KEIN_FENSTER,
    )
    ausgabe = lauf.stdout.decode("utf-8", errors="replace")
    for zeile in reversed(ausgabe.splitlines()):
        zeile = zeile.strip()
        if zeile.startswith("{"):
            try:
                return json.loads(zeile)
            except json.JSONDecodeError:
                break
    fehler = lauf.stderr.decode("utf-8", errors="replace").strip().splitlines()[-3:]
    raise RuntimeError(f"keine JSON-Zeile vom WSL-Teil (Exit {lauf.returncode}; {' | '.join(fehler)[:300]})")


def fahre_dockerhost(tag: str, teil: str = "alpaca") -> dict:
    """Liest das Ergebnis, das die Abendkontrolle auf dem Docker-Host fuer <tag> und <teil> abgelegt hat.

    Der Host legt je Tag <tag>.json (alpaca) und <tag>-ibkr.json (ibkr) ab; angefragt wird mit
    genau dem Namen ohne .json. Exit 3 von abendkontrolle-lesen.sh heisst: Dafuer liegt dort noch
    nichts. Das ist ein Fehlschlag wie jeder andere -- die Notiz bekommt #needs-review, und der
    naechste Lauf versucht es erneut. BatchMode: nie eine Rueckfrage. StrictHostKeyChecking=yes:
    nur der bekannte Host.
    """
    ziel = tag + (IBKR_ENDUNG if teil == "ibkr" else "")
    lauf = subprocess.run(
        [
            SSH,
            "-p", SSH_PORT,
            "-i", str(SSH_SCHLUESSEL),
            "-o", "BatchMode=yes",
            "-o", "IdentitiesOnly=yes",
            "-o", "StrictHostKeyChecking=yes",
            "-o", f"UserKnownHostsFile={SSH_BEKANNTE}",
            "-o", "ConnectTimeout=20",
            SSH_ZIEL,
            ziel,
        ],
        capture_output=True, timeout=SSH_TIMEOUT_S, creationflags=KEIN_FENSTER,
    )
    if lauf.returncode == 3:
        raise RuntimeError(
            f"auf dem Docker-Host liegt fuer {ziel} noch kein Ergebnis "
            "(dort laeuft die Abendkontrolle werktags 16:30 New Yorker Zeit)"
        )
    ausgabe = lauf.stdout.decode("utf-8", errors="replace")
    for zeile in reversed(ausgabe.splitlines()):
        zeile = zeile.strip()
        if zeile.startswith("{"):
            try:
                return json.loads(zeile)
            except json.JSONDecodeError:
                break
    fehler = lauf.stderr.decode("utf-8", errors="replace").strip().splitlines()[-3:]
    raise RuntimeError(f"keine JSON-Zeile vom Docker-Host (Exit {lauf.returncode}; {' | '.join(fehler)[:300]})")


def ssh_aufruf(befehl: str) -> list[str]:
    """Die Befehlszeile fuer einen Befehl der Sicherung, mit denselben Optionen wie fahre_dockerhost."""
    return [
        SSH,
        "-p", SSH_PORT,
        "-i", str(SSH_SCHLUESSEL),
        "-o", "BatchMode=yes",
        "-o", "IdentitiesOnly=yes",
        "-o", "StrictHostKeyChecking=yes",
        "-o", f"UserKnownHostsFile={SSH_BEKANNTE}",
        "-o", "ConnectTimeout=20",
        SSH_ZIEL,
        befehl,
    ]


def fehlerzeile(lauf: subprocess.CompletedProcess) -> str:
    """Exit und die letzten Zeilen von stderr, gekuerzt."""
    fehler = (lauf.stderr or b"").decode("utf-8", errors="replace").strip().splitlines()[-3:]
    return f"Exit {lauf.returncode}; {' | '.join(fehler)[:300]}"


def summe_von(datei: Path) -> str:
    """SHA-256 einer Datei in Kleinbuchstaben, stueckweise gelesen."""
    summe = hashlib.sha256()
    with datei.open("rb") as strom:
        for stueck in iter(lambda: strom.read(1 << 20), b""):
            summe.update(stueck)
    return summe.hexdigest()


class HostNichtErreichbar(Exception):
    """ssh selbst scheiterte (Exit 255) oder ein Befehl der Sicherung lief in sein Zeitlimit (seit Paket NT).

    Beim Archiv macht hole_sicherung daraus den Fehlschlag dieser einen Instanz.
    """


def ssh_lauf(befehl: str, timeout, stdout=None) -> subprocess.CompletedProcess:
    """Ein Befehl der Sicherung beim Host. Mit stdout (offene Datei) geht die Ausgabe dorthin.

    Eigene Funktion, damit die Angriffsfaelle Zeitlimit und Dauer jedes Befehls sehen. Seit Paket NT
    (Quorum 4, G1): abendkontrolle-lesen.sh endet mit 0 bis 4. Exit 255 meldet ssh selbst: keine
    Verbindung oder Anmeldung, aber auch ein Befehl, den auf dem Host ein Signal beendet. Er und ein
    abgelaufenes Zeitlimit heissen: Der Docker-Host ist nicht erreichbar (HostNichtErreichbar).
    """
    try:
        if stdout is None:
            lauf = subprocess.run(
                ssh_aufruf(befehl), capture_output=True, timeout=timeout, creationflags=KEIN_FENSTER,
            )
        else:
            lauf = subprocess.run(
                ssh_aufruf(befehl), stdout=stdout, stderr=subprocess.PIPE, timeout=timeout,
                creationflags=KEIN_FENSTER,
            )
    except subprocess.TimeoutExpired:
        raise HostNichtErreichbar(f"{befehl}: Zeitlimit {timeout} s abgelaufen") from None
    if lauf.returncode == 255:
        raise HostNichtErreichbar(f"{befehl}: {fehlerzeile(lauf)}")
    return lauf


def summe_auf_d(ordner: Path, instanz: str) -> str | None:
    """Die Summe aus <instanz>.tar.sha256 in ordner, wenn sie dort richtig steht und das Archiv daneben liegt.

    Die Summendatei schreibt nur hole_sicherung, nach Pruefung und Bestaetigung: Sie steht fuer ein
    geprueftes und bestaetigtes Archiv. Das Archiv selbst hasht dieser Lauf nicht nach (Plankritik
    Runde 1, Befund 21); sonst laese jeder Lauf alle Archive des Fensters.
    """
    archiv = ordner / f"{instanz}.tar"
    try:
        treffer = SUMMENZEILE.fullmatch((ordner / f"{instanz}.tar.sha256").read_bytes().decode("ascii"))
    except (OSError, UnicodeDecodeError):
        return None
    if treffer is None or treffer.group(2) != archiv.name or not archiv.is_file():
        return None
    return treffer.group(1)


def sicherungstage(jetzt: datetime, tag: str | None = None) -> list[str]:
    """Die Tage, fuer die der Lauf den Host fragt, der aelteste zuerst.

    Ohne SICHERUNG_ERSTER_TAG keiner. Mit tag (--tag) nur dieser. Sonst jeder Kalendertag der letzten
    SICHERUNG_NACHHOLEN_TAGE bis heute ab SICHERUNG_ERSTER_TAG, auch Samstag und Sonntag (ein
    nachgeholter Lauf des Hosts traegt das Datum des Nachholens) und auch heute (Plankritik Runde 1,
    Befunde 8, 16, 18 und 19). Ob das Ergebnis der Abendkontrolle fuer den Tag schon liegt, spielt
    keine Rolle.
    """
    erster = SICHERUNG_ERSTER_TAG
    if erster is None:
        return []
    if tag is not None:
        # Der Tag wird Teil eines Pfads und eines Befehls auf dem Host. Seine Form (JJJJ-MM-TT, ein Datum,
        # das es gibt) prueft seit Paket NT main gleich nach dem Zerlegen der Argumente (tag_gueltig).
        return [tag]
    tage = []
    for zurueck in range(SICHERUNG_NACHHOLEN_TAGE, -1, -1):
        kandidat = jetzt.date() - timedelta(days=zurueck)
        if kandidat >= erster:
            tage.append(kandidat.isoformat())
    return tage


def summe_vom_host(name: str) -> str | None:
    """Die Summe, die der Host fuer <name> fuehrt; None bei Exit 3 (dort liegt keine Sicherung)."""
    lauf = ssh_lauf(f"sicherung-summe {name}", SSH_TIMEOUT_S)
    if lauf.returncode == 3:
        return None
    if lauf.returncode != 0:
        raise RuntimeError(f"Summe nicht gelesen ({fehlerzeile(lauf)})")
    treffer = SUMMENZEILE.fullmatch(lauf.stdout.decode("ascii", errors="replace"))
    if treffer is None or treffer.group(2) != f"{name}.tar":
        raise RuntimeError("Summendatei des Hosts in unerwarteter Form")
    return treffer.group(1)


def ohne_sicherung_still(tag: str, jetzt: datetime) -> bool:
    """Ist Exit 3 fuer diesen Tag kein Fehlschlag? Ja fuer heute (und spaeter) und fuer Samstag und Sonntag.

    Heute kann die Sicherung des Hosts noch ausstehen; am Wochenende sichert er nur, wenn er einen
    verpassten Lauf nachholt. Fehlt sie an einem vergangenen Werktag, ist das ein Fehlschlag.
    """
    kandidat = date.fromisoformat(tag)
    return kandidat >= jetzt.date() or kandidat.weekday() >= 5


def sichere(tag: str, instanz: str, ziel: Path, jetzt: datetime) -> tuple[str, bool]:
    """Eine Instanz eines Tags. Liefert ihren Teil der Zeile und ob geholt wurde; Fehlschlag: Ausnahme.

    Massstab ist die aktuelle Summe des Hosts (Plankritik Runde 1, Befunde 5 und 24). Steht dieselbe in
    der Summendatei auf D:, ist der Tag geholt und bestaetigt. Sonst holt der Lauf neu und ersetzt die
    alte Fassung, etwa nachdem der Host den Tag ein zweites Mal gesichert hat.
    """
    name = f"{tag}-{instanz}"
    summe = summe_vom_host(name)
    if summe is None:
        if ohne_sicherung_still(tag, jetzt):
            return f"{instanz} auf dem Docker-Host noch nicht gesichert", False
        raise RuntimeError(f"auf dem Docker-Host liegt fuer {name} noch keine Sicherung")
    if summe_auf_d(ziel / tag, instanz) == summe:
        return f"{instanz} lag schon vollständig", False
    return hole_sicherung(tag, instanz, ziel, summe), True


def hole_sicherung(tag: str, instanz: str, ziel: Path, summe: str) -> str:
    """Holt die Sicherung <tag>-<instanz> mit der Summe des Hosts, prueft, legt ab und bestaetigt.

    Reihenfolge: Archiv in die Nachbardatei <instanz>.tar.teil holen, Summe pruefen, eine alte
    <instanz>.tar.sha256 entfernen, auf <instanz>.tar umbenennen (ersetzt eine alte Fassung), beim
    Host bestaetigen, zuletzt <instanz>.tar.sha256 schreiben. Erst die Summendatei macht die Ablage
    vollstaendig: Scheitert die Bestaetigung, fehlt sie, und der naechste Lauf holt und bestaetigt
    erneut. Bei falscher Summe wird nichts umbenannt und nichts bestaetigt. Jeder Fehlschlag ist
    eine Ausnahme mit dem Grund. Liefert den Teil der Zeile.
    """
    name = f"{tag}-{instanz}"
    ordner = ziel / tag
    ordner.mkdir(parents=True, exist_ok=True)
    teil = ordner / f"{instanz}.tar.teil"
    archiv = ordner / f"{instanz}.tar"
    summendatei = ordner / f"{instanz}.tar.sha256"
    try:
        with teil.open("wb") as strom:
            try:
                lauf = ssh_lauf(f"sicherung {name}", SICHERUNG_TIMEOUT_S, strom)
            except HostNichtErreichbar as abbruch:
                # Seit Paket NT (Plankritik, Befund 1): Mitten im Archiv heissen Zeitlimit und Exit 255 nicht,
                # dass der Host fehlt; er hat die Summe gerade beantwortet. Nur diese Instanz scheitert, und
                # ein einzelnes zu langsames Archiv haelt nicht jeden Lauf an derselben Stelle an. Ist der
                # Host wirklich weg, merkt es die naechste Frage nach einer Summe.
                raise RuntimeError(f"Archiv nicht geholt ({abbruch})") from None
        if lauf.returncode != 0:
            raise RuntimeError(f"Archiv nicht geholt ({fehlerzeile(lauf)})")
        if summe_von(teil) != summe:
            raise RuntimeError("Summe des geholten Archivs ist nicht die des Hosts; nicht abgelegt, nicht bestaetigt")
        summendatei.unlink(missing_ok=True)
        os.replace(teil, archiv)
    finally:
        try:
            teil.unlink(missing_ok=True)
        except OSError:
            pass
    lauf = ssh_lauf(f"sicherung-bestaetigt {name} {summe}", SSH_TIMEOUT_S)
    if lauf.returncode != 0:
        raise RuntimeError(f"abgelegt, aber nicht bestaetigt ({fehlerzeile(lauf)})")
    neu = ordner / f"{instanz}.tar.sha256.neu"
    neu.write_bytes(f"{summe}  {archiv.name}\n".encode("ascii"))
    os.replace(neu, summendatei)
    return f"{instanz} abgelegt und bestätigt ({archiv.stat().st_size} Bytes)"


def hole_sicherungen(jetzt: datetime, vault: Path, ablage: Path, ziel: Path, instanzen: list[str],
                     frist: float, tag: str | None = None) -> int:
    """Je Tag des Fensters und Instanz: Summe des Hosts fragen und holen, wo D: sie nicht hat.

    Eine Zeile in der Notiz je Tag, an dem etwas geholt wurde, etwas scheiterte oder das Zeitbudget
    endete; ein Fehlschlag traegt #needs-review. Ein Tag ohne Neues bekommt keine Zeile. frist ist der
    Stand von monotonic(), an dem SICHERUNG_BUDGET_S ab dem Start des Laufs um ist (main); danach
    faengt der Lauf keine Instanz mehr an, und der naechste holt den Rest.
    """
    if not instanzen:
        return 0
    code = 0
    erschoepft = False
    for kandidat in sicherungstage(jetzt, tag):
        if erschoepft:
            break
        uhrzeit = jetzt.strftime("%H:%M") if kandidat == jetzt.date().isoformat() else f"{jetzt:%Y-%m-%d %H:%M}"
        stuecke = []
        gescheitert = False
        getan = False
        for instanz in instanzen:
            if monotonic() >= frist:
                erschoepft = getan = True
                stuecke.append(
                    f"{instanz} nicht angefangen: Zeitbudget des Laufs ({SICHERUNG_BUDGET_S} s) erschöpft, "
                    "der nächste Lauf holt den Rest"
                )
                logge(ablage, f"sicherung {kandidat}-{instanz}: Zeitbudget erschoepft")
                continue
            try:
                stueck, geholt = sichere(kandidat, instanz, ziel, jetzt)
                stuecke.append(stueck)
                if geholt:
                    getan = True
                    logge(ablage, f"sicherung {kandidat}-{instanz}: abgelegt und bestaetigt")
            except HostNichtErreichbar as ausnahme:
                # Seit Paket NT (Quorum 4, G1): Ohne Host scheiterte sonst jeder Tag des Fensters mit eigener
                # Zeile, auch in neu angelegten Notizen von Wochenenden. Die Sicherung hoert hier auf: eine
                # Zeile in der Notiz von heute, Exit 1. Fertige Tage behalten ihre Zeilen; was der
                # angefangene Tag schon hat, nennt diese Zeile. Auf D: entsteht danach nichts mehr.
                logge(ablage, f"sicherung {kandidat}-{instanz}: Docker-Host nicht erreichbar ({ausnahme}), Sicherung abgebrochen")
                zeile = (
                    f"- {jetzt:%H:%M} Uhr — {SICHERUNG_KOPF}: **gescheitert** #needs-review. Docker-Host nicht "
                    f"erreichbar ({str(ausnahme)[:200]}), Sicherung abgebrochen."
                )
                if stuecke:
                    zeile += f" Bis dahin am {kandidat}: {'; '.join(stuecke)}."
                zeile += " Der nächste Lauf versucht es erneut."
                try:
                    schreibe_notiz(vault, jetzt.date().isoformat(), zeile)
                except Exception as zweite:
                    logge(ablage, f"sicherung: Notiz nicht geschrieben {zweite!r}")
                return 1
            except Exception as ausnahme:
                gescheitert = True
                stuecke.append(f"{instanz} gescheitert ({type(ausnahme).__name__}: {str(ausnahme)[:200]})")
                logge(ablage, f"sicherung {kandidat}-{instanz}: gescheitert {ausnahme!r}")
        if not (gescheitert or getan):
            continue
        kopf = f"- {uhrzeit} Uhr — {SICHERUNG_KOPF}"
        if gescheitert:
            code = 1
            zeile = f"{kopf}: **gescheitert** #needs-review. {'; '.join(stuecke)}. Der nächste Lauf versucht es erneut."
        else:
            zeile = f"{kopf}: {'; '.join(stuecke)}."
        try:
            schreibe_notiz(vault, kandidat, zeile)
        except Exception as ausnahme:
            code = 1
            logge(ablage, f"sicherung {kandidat}: Notiz nicht geschrieben {ausnahme!r}")
    return code


def zeile_fuer(ergebnis: dict, uhrzeit: str) -> str:
    kopf = f"- {uhrzeit} Uhr — Abendkontrolle Trading Plattform (automatisch)"
    befunde = ergebnis.get("befunde") or []
    fakten = ergebnis.get("fakten") or {}
    hinweise = ergebnis.get("hinweise") or []
    if fakten.get("handelstag") is False and not befunde:
        return f"{kopf}: kein Handelstag laut Alpaca-Kalender."
    teile = []
    ende = fakten.get("ende") or {}
    if ende:
        teile.append(
            f"Ende {ende.get('grund')}, Exit {ende.get('exit')}, Orders {ende.get('orders')}, "
            f"Position {ende.get('position')}"
        )
    laeufe = fakten.get("laeufe") or []
    if laeufe:
        teile.append(f"{len(laeufe)} Lauf/Läufe, {sum(l.get('takte_aktiv', 0) for l in laeufe)} Takte aktiv")
    if "broker_offen" in fakten:
        teile.append(f"offene Positionen beim Broker {fakten['broker_offen']}")
    if "bericht" in fakten:
        teile.append("Tagesbericht da" if fakten["bericht"] else "Tagesbericht fehlt")
    killswitch = fakten.get("killswitch") or {}
    if killswitch:
        teile.append(f"Killswitch {killswitch.get('zustand')} ({killswitch.get('ausloeser')})")
    return _zeile(kopf, befunde, "; ".join(teile), fakten, hinweise)


def _zeile(kopf: str, befunde: list, zusammen: str, fakten: dict, hinweise: list) -> str:
    """Kopf, Urteil, Zusammenfassung, Befunde und Hinweise -- fuer beide Teile dieselbe Form."""
    if not befunde:
        text = f"{kopf}: ohne Befund. {zusammen}."
    else:
        text = f"{kopf}: **{len(befunde)} Befund(e)** #needs-review. {zusammen}."
        text += "".join(f"\n  - {befund}" for befund in befunde)
        if fakten.get("meldung"):
            text += f"\n  - Meldung ans Telefon: {fakten['meldung']}"
    text += "".join(f"\n  - Hinweis: {hinweis}" for hinweis in hinweise[:5])
    return text


def zeile_fuer_ibkr(ergebnis: dict, uhrzeit: str) -> str:
    """Die eigene Zeile der IBKR-Instanz (seit Paket IB2). Kein Betrag, keine Kontonummer."""
    kopf = f"- {uhrzeit} Uhr — {KOEPFE['ibkr']}"
    befunde = ergebnis.get("befunde") or []
    fakten = ergebnis.get("fakten") or {}
    hinweise = ergebnis.get("hinweise") or []
    if fakten.get("handelstag") is False and not befunde:
        return f"{kopf}: kein Handelstag laut Alpaca-Kalender."
    if fakten.get("nicht_gestartet") and not befunde:
        return f"{kopf}: IBKR-Instanz heute nicht gestartet (Zeitgeber aus)."
    teile = []
    ende = fakten.get("ende") or {}
    if ende:
        teile.append(
            f"Ende {ende.get('grund')}, Exit {ende.get('exit')}, Orders {ende.get('orders')}, "
            f"Position {ende.get('position')}"
        )
    laeufe = fakten.get("laeufe") or []
    if laeufe:
        teile.append(f"{len(laeufe)} Lauf/Läufe, {sum(l.get('takte_aktiv', 0) for l in laeufe)} Takte aktiv")
    konto = fakten.get("konto") or {}
    if konto:
        teile.append(
            f"Konto beim Gateway: {konto.get('positionen')} Position(en), "
            f"{konto.get('orders')} offene Order(s)"
        )
    killswitch = fakten.get("killswitch") or {}
    if killswitch:
        teile.append(f"Killswitch {killswitch.get('zustand')} ({killswitch.get('ausloeser')})")
    return _zeile(kopf, befunde, "; ".join(teile), fakten, hinweise)


def schreibe_notiz(vault: Path, tag: str, zeile: str) -> None:
    notiz = vault / "Inbox" / f"{tag}.md"
    if notiz.exists():
        text = notiz.read_bytes().decode("utf-8")  # kein UTF-8: Ausnahme, die Notiz bleibt unberuehrt
    else:
        text = (
            f"---\ntype: daily\ntitle: \"{tag}\"\nstatus: open\nsummary: \"Tagesnotiz vom {tag}.\"\n"
            f"tags: [daily]\ncreated: {tag}\nupdated: {tag}\n---\n\n# {tag}\n"
        )
    # Zeilenenden der Datei bleiben, wie sie sind; gesucht wird ohne \r.
    zeilen = text.split("\n")
    bereinigt = [z.rstrip("\r").rstrip() for z in zeilen]
    cr = "\r" if any(z.endswith("\r") for z in zeilen) else ""
    # Die erste Ueberschrift, die mit "## Agentenprotokoll" beginnt ("## Agentenprotokoll — …" kommt vor).
    start = next((i for i, z in enumerate(bereinigt) if z == UEBERSCHRIFT or z.startswith(UEBERSCHRIFT + " ")), None)
    if start is None:
        nl = cr + "\n"
        text = text.rstrip("\r\n") + nl + nl + UEBERSCHRIFT + nl + nl + nl.join(zeile.split("\n")) + nl
    else:
        ende = next((i for i in range(start + 1, len(zeilen)) if bereinigt[i].startswith("## ")), len(zeilen))
        # Hinter den letzten nicht leeren Eintrag des Abschnitts; die Leerzeilen danach ersetzt eine.
        einfuegen = ende
        while einfuegen > start + 1 and not bereinigt[einfuegen - 1]:
            einfuegen -= 1
        block = [z + cr for z in zeile.split("\n")]
        if einfuegen == start + 1:
            block = [cr] + block
        zeilen[einfuegen:ende] = block + ([cr] if ende < len(zeilen) else [""])
        text = "\n".join(zeilen)
        if not text.endswith("\n"):
            text += "\n"
    neu = notiz.with_name(notiz.name + ".abendkontrolle.tmp")
    neu.write_bytes(text.encode("utf-8"))
    os.replace(neu, notiz)


def logge(ablage: Path, text: str) -> None:
    ablage.mkdir(parents=True, exist_ok=True)
    with (ablage / "lauf.log").open("a", encoding="utf-8", newline="\n") as datei:
        datei.write(f"{datetime.now().isoformat(timespec='seconds')} {text}\n")


def main(argv=None) -> int:
    zerleger = argparse.ArgumentParser(allow_abbrev=False)
    zerleger.add_argument("--tag")
    zerleger.add_argument("--vault", default=str(VAULT))
    zerleger.add_argument("--ohne-wsl", help="Ergebnis aus einer JSON-Datei statt aus WSL (Angriffsfaelle)")
    zerleger.add_argument("--jetzt", help="ISO-Zeitpunkt statt der Uhr (Angriffsfaelle)")
    zerleger.add_argument("--teil", choices=TEILE, help="nur diesen Teil; ohne Angabe beide, mit --ohne-wsl nur alpaca")
    zerleger.add_argument(
        "--quelle", choices=QUELLEN, default="wsl",
        help="wsl (Vorgabe): pruefen in WSL; dockerhost: das Ergebnis des Docker-Hosts ueber SSH lesen",
    )
    zerleger.add_argument(
        "--quelle-ibkr", choices=QUELLEN,
        help="die Quelle fuer den Teil ibkr; ohne Angabe dieselbe wie --quelle",
    )
    zerleger.add_argument(
        "--sicherung-ziel",
        help="Ordner fuer die Sicherungen statt SICHERUNG_ZIEL (Angriffsfaelle)",
    )
    argumente = zerleger.parse_args(argv)
    # Seit Paket NT (Quorum 4): --tag gleich hier pruefen. Der Tag wird Name der Notiz, Pfad unter
    # SICHERUNG_ZIEL und Teil eines Befehls auf dem Host; vorher legte ein Tag ausser der Form
    # Inbox/<tag>.md an, mit ../ auch ausserhalb von Inbox/. Leer zaehlt als ausser der Form. Die
    # Ausgabe ist ASCII (!a), damit kein Zeichen an der Kodierung von stderr scheitert.
    if argumente.tag is not None and not tag_gueltig(argumente.tag):
        print(f"abendkontrolle: --tag {argumente.tag!a} ist kein Datum JJJJ-MM-TT", file=sys.stderr)
        return 2
    # Seit Paket E9: Das Zeitbudget der Sicherung zaehlt ab hier, also mit der Zeit fuer die
    # Ergebnisse (bis zu zwoelf Abfragen mit je SSH_TIMEOUT_S; Plankritik Runde 2, Befund 1).
    frist = monotonic() + SICHERUNG_BUDGET_S
    vault = Path(argumente.vault)
    ablage = vault / ABLAGE_TEIL
    jetzt = datetime.fromisoformat(argumente.jetzt) if argumente.jetzt else datetime.now()
    # Ohne --teil beide Teile, erst alpaca, dann ibkr. Mit --ohne-wsl ohne --teil nur alpaca: Eine
    # Ergebnisdatei ist immer die eines Teils, und die Angriffsfaelle von vor IB2 bleiben gleich.
    teile = [argumente.teil] if argumente.teil else (["alpaca"] if argumente.ohne_wsl else list(TEILE))
    code = 0
    for teil in teile:
        if argumente.tag:
            tage = [argumente.tag]
        elif teil == "ibkr":
            tage = faellige_tage(jetzt, ablage, erster_tag=IBKR_ERSTER_TAG, endung=IBKR_ENDUNG)
        else:
            tage = faellige_tage(jetzt, ablage)
        vorsatz = "" if teil == "alpaca" else f"{teil} "
        # Scheitert ein Teil ausserhalb seiner eigenen Fehlerbehandlung (Ergebnisdatei nicht
        # schreibbar, lauf.log gesperrt), laeuft der naechste trotzdem (Pruefung IB2).
        try:
            logge(ablage, f"{vorsatz}start, faellig: {tage}")
            # Seit Paket DH: die Quelle je Teil; die IBKR-Instanz zieht an einem eigenen Abend um.
            quelle = (argumente.quelle_ibkr or argumente.quelle) if teil == "ibkr" else argumente.quelle
            code = max(code, fahre_teil(teil, tage, jetzt, vault, ablage, argumente.ohne_wsl, quelle))
        except Exception as ausnahme:
            code = 1
            try:
                logge(ablage, f"{vorsatz}abgebrochen: {type(ausnahme).__name__}")
            except Exception:
                pass
    # Seit Paket E9: die Sicherung jedes Teils mit Quelle dockerhost, nach den Ergebnissen beider
    # Teile und unabhaengig davon, ob ein Ergebnis gescheitert ist.
    instanzen = [
        teil for teil in teile
        if ((argumente.quelle_ibkr or argumente.quelle) if teil == "ibkr" else argumente.quelle) == "dockerhost"
    ]
    ziel = Path(argumente.sicherung_ziel) if argumente.sicherung_ziel else SICHERUNG_ZIEL
    try:
        code = max(code, hole_sicherungen(jetzt, vault, ablage, ziel, instanzen, frist, argumente.tag))
    except Exception as ausnahme:
        code = 1
        try:
            logge(ablage, f"sicherung abgebrochen: {type(ausnahme).__name__}: {ausnahme}")
        except Exception:
            pass
    return code


def fahre_teil(teil: str, tage: list[str], jetzt: datetime, vault: Path, ablage: Path, ohne_wsl: str | None,
               quelle: str = "wsl") -> int:
    """Ein Teil fuer alle faelligen Tage: Ergebnisdatei, Zeile in der Notiz, lauf.log.

    quelle (seit Paket DH): wsl prueft in WSL, dockerhost liest das Ergebnis des Docker-Hosts.
    """
    endung = IBKR_ENDUNG if teil == "ibkr" else ""
    zeile_von = zeile_fuer_ibkr if teil == "ibkr" else zeile_fuer
    vorsatz = "" if teil == "alpaca" else f"{teil} "
    code = 0
    for tag in tage:
        uhrzeit = jetzt.strftime("%H:%M") if tag == jetzt.date().isoformat() else f"{jetzt:%Y-%m-%d %H:%M}"
        try:
            if ohne_wsl:
                ergebnis = json.loads(Path(ohne_wsl).read_text(encoding="utf-8"))
            elif quelle == "dockerhost":
                ergebnis = fahre_dockerhost(tag, teil)
            else:
                ergebnis = fahre_wsl(tag, teil)
            if not isinstance(ergebnis, dict) or not isinstance(ergebnis.get("befunde"), list):
                raise RuntimeError("Ergebnis des WSL-Teils in unerwarteter Form")
            # Ein Ergebnis ohne teil stammt von einem WSL-Teil vor IB2 und gilt als alpaca. Passt
            # der Teil nicht, landete es sonst still in der falschen Ergebnisdatei und Zeile.
            if ergebnis.get("teil", "alpaca") != teil:
                raise RuntimeError(f"Ergebnis des WSL-Teils gehoert zu {ergebnis.get('teil')!r}, nicht zu {teil!r}")
        except Exception as ausnahme:
            code = 1
            logge(ablage, f"{vorsatz}{tag}: gescheitert {ausnahme!r}\n{traceback.format_exc()}")
            try:
                schreibe_notiz(
                    vault, tag,
                    f"- {uhrzeit} Uhr — {KOEPFE[teil]}: **gescheitert** "
                    f"({type(ausnahme).__name__}: {str(ausnahme)[:200]}) #needs-review. Der nächste Lauf versucht es erneut.",
                )
            except Exception as zweite:
                logge(ablage, f"{vorsatz}{tag}: auch die Notiz scheiterte {zweite!r}")
            continue
        # Ein Abbruch im WSL-Teil (Exit 2) wird wiederholt; alles andere gilt als gelaufen.
        if ergebnis.get("exit") != 2:
            (ablage / f"{tag}{endung}.json").write_text(
                json.dumps(ergebnis, ensure_ascii=False, indent=1), encoding="utf-8", newline="\n"
            )
        else:
            code = 1
        logge(ablage, f"{vorsatz}{tag}: {len(ergebnis['befunde'])} Befund(e), exit {ergebnis.get('exit')}")
        try:
            schreibe_notiz(vault, tag, zeile_von(ergebnis, uhrzeit))
        except Exception as ausnahme:
            code = 1
            logge(ablage, f"{vorsatz}{tag}: Notiz nicht geschrieben {ausnahme!r}")
    return code


if __name__ == "__main__":
    sys.exit(main())