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

"""Richtet in Home Assistant den Totmann der Trading Plattform ein (Paket Z2, Anforderung Nr. 7).

Koordinator-Werkzeug, nicht Teil des Repos. Zwei Teile:
- dieses Skript laeuft unter WINDOWS mit dem Python des HA-Projekts (dort liegen websockets,
  requests, src/ha_verbindung.py, src/config_datei.py und src/vorlage_auswerten.py; Adresse
  der Instanz und Name der Token-Variable stehen dort, nicht hier) und schreibt nach Home
  Assistant;
- ha_totmann_wsl.py laeuft in WSL mit dem Python der Trading Plattform, liest dort die
  Geheimnisdatei und den Kalender von Alpaca und gibt Thema, Token und Kalender als eine
  JSON-Zeile zurueck. Dieses Skript liest sie nur als Pipe (aus_wsl) und gibt davon nichts
  aus, nur Laengen. Die Geheimnisdatei verlaesst WSL nie.
Nichts in diesen Dateien ist ein Geheimnis. Thema und Token landen nur in secrets.yaml von
Home Assistant.

Was der Totmann tut: Eine Automation prueft alle fuenf Minuten, ob das Lebenszeichen der
Plattform (input_datetime.trading_gesendet; der Hausmelder schickt es spaetestens alle fuenf
Minuten) aelter als 15 Minuten ist -- nur an Handelstagen, von 15 Minuten nach der Oeffnung
bis zum Schluss. Dann meldet sie einmal je Stille und Handelstag ueber die HA-App (der Dienst
ist das Argument --app-dienst, kein Literal hier) und ueber ntfy (rest_command, Adresse und
Zugriff aus der secrets.yaml von Home Assistant).

Die Bedingung besteht aus vier Eingaben und einer Entscheidung (ENTSCHEIDUNG). Die Automation
setzt die Eingaben live (EINGABEN_LIVE); trocken rechnet die Entscheidung mit gestellten
Lagen (faelle) lokal, pruefen laesst Home Assistant dieselben Lagen per render_template
rechnen. So ist belegt, dass Home Assistant die Entscheidung so versteht wie gemeint.

Handelstage und Fenster kommen beim Einrichten aus dem Kalender von Alpaca (ueber den
WSL-Teil) fuer die naechsten --tage Tage und stehen als Epoch-Sekunden in der Bedingung. Nach
dem letzten Datum schweigt der Totmann; der Alias nennt es, "einrichten" verlaengert. Die New
Yorker Zeit rechnet das Werkzeug selbst (_ny_versatz_h), weil Python unter Windows ohne das
Paket tzdata keine Zeitzonen kennt.

Schreibwege nach /config: secrets.yaml und configuration.yaml ueber die API des
File-editor-Add-ons (src/config_datei.py des HA-Projekts; dessen Pfadregel erlaubt nur
themes/, dieses Werkzeug ersetzt sie zur Laufzeit fuer genau diese zwei Dateien). Jeder
eigene Block steht zwischen MARKE_ANFANG und MARKE_ENDE; "entfernen" nimmt genau diese
Bloecke wieder heraus. Die Automation geht ueber /api/config/automation/config/<id>, wie in
ha_einrichten.py.

Aktionen:
  pruefen    lesend: Dienste, Bloecke (nur ja oder nein), Automation, Alter des
             Lebenszeichens, die Eingaben heute/jetzt/gesendet live, die juengste Spur der
             Automation (ob die Bedingung mit this ohne Fehler rechnet) und die Rechenprobe
             der Bedingung in Home Assistant
  trocken    ohne Netz: baut Automation und Bloecke aus erfundenen Werten, rechnet die
             gestellten Lagen lokal und prueft die Regeln des WSL-Teils
  einrichten holt Thema, Token und Kalender aus WSL, schreibt die Bloecke (wenn noetig),
             prueft die Konfiguration, laedt rest_command neu oder verlangt einen Neustart
             von Home Assistant (Exit 4) und legt die Automation an oder ersetzt sie
  probe      ruft rest_command und den App-Dienst je einmal -- zwei echte Meldungen
  entfernen  loescht die Automation und nimmt beide Bloecke heraus

Aufruf: <ha-projekt>\\.venv\\Scripts\\python.exe ha_totmann.py <aktion> [<ha-projektordner>
        <trading-projektordner>] [--app-dienst notify.<name>] [--tage 400]
Exit: 0 fertig, 2 Abbruch oder BEFUND, 3 Voraussetzung fehlt, 4 Neustart von HA noetig.
"""

import argparse
import json
import re
import subprocess
import sys
import time
import tomllib
from datetime import date
from datetime import datetime
from datetime import timedelta
from datetime import timezone
from pathlib import Path
from pathlib import PureWindowsPath

sys.path.insert(0, str(Path(__file__).resolve().parent))
import ha_totmann_wsl as wsl_teil  # noqa: E402

AUTOMATION_ID = "trading_plattform_totmann"
ALIAS = "Trading Plattform: Totmann"
LEBENSZEICHEN = "input_datetime.trading_gesendet"
REST_DIENST = "trading_totmann_ntfy"
SECRET_URL = "trading_totmann_ntfy_url"
SECRET_AUTH = "trading_totmann_ntfy_auth"
MARKE_ANFANG = "# >>> trading-totmann (ha_totmann.py, Paket Z2)"
MARKE_ENDE = "# <<< trading-totmann"
ERLAUBT = ("secrets.yaml", "configuration.yaml")
STILLE_S = 900
# Wie weit die Uhr von Home Assistant von der dieses Rechners abweichen darf, bevor
# eingaben_befunde es als Befund nennt.
UHR_TOLERANZ_S = 120
PRUEFTAKT = "/5"
APP_DIENST_MUSTER = re.compile(r"^notify\.[a-z0-9_]+$")
FREMDES_REST_COMMAND = re.compile(r"(?m)^rest_command:")
WSL = r"C:\Windows\System32\wsl.exe"
WSL_PY = "/home/claude/venvs/trading/bin/python"
WSL_TEIL = Path(__file__).resolve().with_name("ha_totmann_wsl.py")
WSL_TIMEOUT_S = 120
TEXT = (
    "Die Trading Plattform schweigt im Handelsfenster seit mehr als 15 Minuten. Letztes "
    "Lebenszeichen: {{ states('" + LEBENSZEICHEN + "') }}. Nachsehen: Rechner, WSL, Dashboard."
)
PROBE_TEXT = "Probe des Totmanns (ha_totmann.py probe): Diese Meldung kam an. Nichts zu tun."
# Erfundene Werte fuer trocken. Kein echtes Thema, kein echter Dienst, kein echter Token.
TROCKEN_THEMA = "probe-thema-nicht-echt"
TROCKEN_TOKEN = "tk_probe_nicht_echt"
TROCKEN_DIENST = "notify.probe_dienst"

# Die vier Eingaben der Bedingung, wie die Automation sie liest. this ist die Automation
# selbst; ihr last_triggered ist die letzte Ausloesung.
EINGABEN_LIVE = {
    "heute": "now().strftime('%Y-%m-%d')",
    "jetzt": "now().timestamp()",
    "gesendet": "state_attr('" + LEBENSZEICHEN + "', 'timestamp') | float(0)",
    "zuletzt": "as_timestamp(this.attributes.last_triggered, 0)",
}
# Die Entscheidung, gemeinsam fuer Automation, trocken und pruefen. Ausloesen, wenn heute
# ein Handelstag ist, jetzt im Fenster liegt, das Lebenszeichen aelter als STILLE_S ist und
# die letzte Ausloesung vor dem Lebenszeichen lag -- oder vor dem heutigen Fenster: Haelt
# eine Stille ueber Nacht an, meldet der neue Handelstag sie einmal.
ENTSCHEIDUNG = (
    "f is not none and f[0] <= jetzt <= f[1] and jetzt - gesendet > "
    + str(STILLE_S)
    + " and (zuletzt <= gesendet or zuletzt < f[0])"
)
# Ein erfundener Kalender fuer trocken und die Rechenprobe in pruefen: zwei gewoehnliche Tage
# und der verkuerzte Tag nach Thanksgiving 2026.
PROBE_KALENDER = [
    {"date": "2026-09-28", "open": "09:30", "close": "16:00"},
    {"date": "2026-09-29", "open": "09:30", "close": "16:00"},
    {"date": "2026-11-27", "open": "09:30", "close": "13:00"},
]


class Voraussetzung(Exception):
    """Etwas fehlt fuer den Lauf. Der Text traegt keinen Wert aus der Geheimnisdatei."""


def _ny_versatz_h(tag: date) -> int:
    """UTC-Versatz der New Yorker Boersenzeit an einem Tag: -4 in der Sommerzeit, sonst -5.

    US-Regel seit 2007: vom zweiten Sonntag im Maerz bis zum ersten Sonntag im November.
    Umgestellt wird sonntags um 02:00; an einem Handelstag gilt der Versatz des ganzen Tages.
    """
    maerz = date(tag.year, 3, 1)
    zweiter_sonntag_maerz = maerz + timedelta(days=(6 - maerz.weekday()) % 7 + 7)
    november = date(tag.year, 11, 1)
    erster_sonntag_november = november + timedelta(days=(6 - november.weekday()) % 7)
    return -4 if zweiter_sonntag_maerz <= tag < erster_sonntag_november else -5


def fenster_je_tag(kalender: list[dict]) -> dict[str, list[int]]:
    """Je Handelstag [Oeffnung plus STILLE_S, Schluss] als Epoch-Sekunden."""
    fenster: dict[str, list[int]] = {}
    for eintrag in kalender:
        tag = date.fromisoformat(eintrag["date"])
        zone = timezone(timedelta(hours=_ny_versatz_h(tag)))
        grenzen = []
        for feld in ("open", "close"):
            stunde, minute = (int(teil) for teil in eintrag[feld].split(":"))
            grenzen.append(
                int(datetime(tag.year, tag.month, tag.day, stunde, minute, tzinfo=zone).timestamp())
            )
        fenster[tag.isoformat()] = [grenzen[0] + STILLE_S, grenzen[1]]
    return fenster


def gestellt(heute: str, jetzt: int, gesendet: int, zuletzt: int) -> dict[str, str]:
    """Gestellte Eingaben als Vorlagen-Literale, fuer die Rechenprobe in pruefen."""
    return {
        "heute": "'" + heute + "'",
        "jetzt": str(jetzt),
        "gesendet": str(gesendet),
        "zuletzt": str(zuletzt),
    }


def bedingung(fenster: dict[str, list[int]], eingaben: dict[str, str] | None = None) -> str:
    """Die Vorlage der Bedingung; ohne eingaben die der Automation (EINGABEN_LIVE)."""
    werte = EINGABEN_LIVE if eingaben is None else eingaben
    return (
        "{% set fenster = " + json.dumps(fenster, separators=(",", ":")) + " %}"
        + "".join(
            "{% set " + name + " = " + werte[name] + " %}"
            for name in ("heute", "jetzt", "gesendet", "zuletzt")
        )
        + "{% set f = fenster.get(heute) %}"
        + "{{ " + ENTSCHEIDUNG + " }}"
    )


def _ohne_eingaben(vorlage: str) -> str:
    return re.sub(r"\{% set (heute|jetzt|gesendet|zuletzt) = .*? %\}", "", vorlage)


def _lage(heute: str, jetzt: int, gesendet: int, zuletzt: int) -> dict:
    return {"heute": heute, "jetzt": jetzt, "gesendet": gesendet, "zuletzt": zuletzt}


def faelle(fenster: dict[str, list[int]]) -> list[tuple[str, dict, bool]]:
    """Gestellte Lagen: Name, Eingaben, ob die Bedingung ausloesen soll.

    Montag 28.09. und Dienstag 29.09.2026 aus PROBE_KALENDER, alle Zeiten erfunden. mitte
    liegt eine Stunde nach dem Beginn des Montagsfensters.
    """
    montag, dienstag = fenster["2026-09-28"], fenster["2026-09-29"]
    mitte = montag[0] + 3600
    return [
        ("stille_im_fenster", _lage("2026-09-28", mitte, mitte - 1800, 0), True),
        ("lebenszeichen_frisch", _lage("2026-09-28", mitte, mitte - 300, 0), False),
        (
            "schon_gemeldet_nach_dem_letzten_lebenszeichen",
            _lage("2026-09-28", mitte, mitte - 1800, mitte - 600),
            False,
        ),
        ("vor_dem_fenster", _lage("2026-09-28", montag[0] - 60, montag[0] - 7200, 0), False),
        ("nach_dem_schluss", _lage("2026-09-28", montag[1] + 600, mitte, 0), False),
        ("kein_handelstag", _lage("2026-09-27", mitte - 86400, 0, 0), False),
        (
            "stille_ueber_die_tagesgrenze",
            _lage("2026-09-29", dienstag[0] + 600, mitte - 1800, mitte - 600),
            True,
        ),
        (
            "am_neuen_tag_schon_gemeldet",
            _lage("2026-09-29", dienstag[0] + 1200, mitte - 1800, dienstag[0] + 600),
            False,
        ),
    ]


def _entscheide_lokal(fenster: dict[str, list[int]], lage: dict) -> bool:
    # ENTSCHEIDUNG ist eine Konstante dieser Datei, kein fremder Text; eval ohne Builtins,
    # none steht fuer das None der Vorlagensprache.
    return bool(
        eval(  # noqa: S307
            ENTSCHEIDUNG,
            {"__builtins__": {}},
            {
                "none": None,
                "f": fenster.get(lage["heute"]),
                "jetzt": lage["jetzt"],
                "gesendet": lage["gesendet"],
                "zuletzt": lage["zuletzt"],
            },
        )
    )


def _wahrheit(wert: object) -> bool | None:
    if isinstance(wert, bool):
        return wert
    if isinstance(wert, str) and wert.strip().lower() in ("true", "false"):
        return wert.strip().lower() == "true"
    return None


def automation(fenster: dict[str, list[int]], app_dienst: str) -> dict:
    letzter = max(fenster) if fenster else "keiner"
    return {
        "alias": f"{ALIAS} (Handelstage bis {letzter})",
        "description": (
            "Meldet ueber die HA-App und ntfy, wenn das Lebenszeichen der Trading Plattform "
            f"({LEBENSZEICHEN}) im Handelsfenster aelter als 15 Minuten ist; einmal je "
            f"Stille und Handelstag. Handelstage aus dem Kalender von Alpaca bis {letzter}. "
            "Angelegt ueber ha_totmann.py (Paket Z2); Rueckbau mit der Aktion entfernen."
        ),
        "trigger": [{"platform": "time_pattern", "minutes": PRUEFTAKT}],
        "condition": [{"condition": "template", "value_template": bedingung(fenster)}],
        "action": [
            {
                "service": f"rest_command.{REST_DIENST}",
                "data": {"nachricht": TEXT},
                "continue_on_error": True,
            },
            {
                "service": app_dienst,
                "data": {"title": "Trading Totmann", "message": TEXT},
                "continue_on_error": True,
            },
        ],
        "mode": "single",
        "max_exceeded": "silent",
    }


def config_block(mit_token: bool) -> str:
    """Der Block fuer configuration.yaml. Kein Thema: Adresse und Zugriff kommen per !secret."""
    zeilen = [
        MARKE_ANFANG,
        "# Totmann der Trading Plattform: Meldung ueber ntfy. Adresse und Zugriff stehen in",
        "# secrets.yaml; hier steht kein Thema.",
        "rest_command:",
        f"  {REST_DIENST}:",
        f"    url: !secret {SECRET_URL}",
        "    method: POST",
        '    content_type: "text/plain; charset=utf-8"',
        "    timeout: 10",
        '    payload: "{{ nachricht }}"',
        "    headers:",
        '      Title: "Trading Totmann"',
        '      Priority: "high"',
        '      Tags: "rotating_light"',
    ]
    if mit_token:
        zeilen.append(f"      Authorization: !secret {SECRET_AUTH}")
    zeilen.append(MARKE_ENDE)
    return "\n".join(zeilen) + "\n"


def secrets_block(url: str, token: str) -> str:
    """Der Block fuer secrets.yaml: die ganze Adresse mit Thema, dazu der Zugriff."""
    zeilen = [MARKE_ANFANG, f'{SECRET_URL}: "{url}"']
    if token:
        zeilen.append(f'{SECRET_AUTH}: "Bearer {token}"')
    zeilen.append(MARKE_ENDE)
    return "\n".join(zeilen) + "\n"


def ohne_block(text: str) -> str:
    """Der Text ohne den eigenen Block zwischen den Marken, mit genau einem Umbruch am Ende."""
    muster = re.compile(
        r"\n*" + re.escape(MARKE_ANFANG) + r".*?" + re.escape(MARKE_ENDE) + r"\n?", re.DOTALL
    )
    rest = muster.sub("\n", text).rstrip("\n")
    return rest + "\n" if rest else ""


def anhaengen(text: str, block: str) -> str:
    rest = text.rstrip("\n")
    return (rest + "\n\n" if rest else "") + block


def _ohne(text: str, *geheim: str) -> str:
    for wert in geheim:
        if wert:
            text = text.replace(wert, "***")
    return text


def _wsl_pfad(pfad: str) -> str:
    """Ein Windows-Pfad mit Laufwerksbuchstaben als Pfad unter /mnt in WSL."""
    windows = PureWindowsPath(pfad)
    if len(windows.drive) != 2 or not windows.drive.endswith(":"):
        raise Voraussetzung(f"kein Laufwerkspfad: {pfad}")
    return "/mnt/" + windows.drive[0].lower() + "/" + "/".join(windows.parts[1:])


def pruefe_wsl_ausgabe(daten: object) -> dict:
    """Die Zeile des WSL-Teils, nach denselben Regeln geprueft, die er selbst anlegt."""
    if not isinstance(daten, dict) or set(daten) != {"thema", "token", "kalender"}:
        raise Voraussetzung("WSL-Teil: Ausgabe in unerwarteter Form")
    thema, token = wsl_teil.zugang({"ntfy": {"thema": daten["thema"], "token": daten["token"]}})
    return {"thema": thema, "token": token, "kalender": wsl_teil.pruefe_kalender(daten["kalender"])}


def aus_wsl(tage: int) -> dict:
    """Startet den WSL-Teil und liest seine eine Zeile als Pipe. stdout wird nie ausgegeben."""
    lauf = subprocess.run(
        [
            WSL, "-d", "Debian", "-u", "claude", "--exec",
            WSL_PY, _wsl_pfad(str(WSL_TEIL)), "--tage", str(tage),
        ],
        capture_output=True,
        timeout=WSL_TIMEOUT_S,
    )
    if lauf.returncode != 0:
        # stderr des WSL-Teils traegt nur Befunde ohne Werte.
        fehler = lauf.stderr.decode("utf-8", errors="replace").replace("\x00", "").strip()
        raise Voraussetzung(
            f"WSL-Teil endete mit Exit {lauf.returncode}: {fehler[:300] or 'ohne Meldung'}"
        )
    zeilen = [
        zeile
        for zeile in lauf.stdout.decode("utf-8", errors="replace").split("\n")
        if zeile.startswith("{")
    ]
    if len(zeilen) != 1:
        raise Voraussetzung(f"WSL-Teil: {len(zeilen)} statt genau einer JSON-Zeile")
    return pruefe_wsl_ausgabe(json.loads(zeilen[0]))


def trocken() -> int:
    """Prueft die eigenen Bausteine ohne Netz, mit erfundenen Werten."""
    befunde = []
    for tag, erwartet in (
        (date(2026, 3, 6), -5),
        (date(2026, 3, 9), -4),
        (date(2026, 10, 30), -4),
        (date(2026, 11, 2), -5),
        (date(2027, 3, 12), -5),
        (date(2027, 3, 15), -4),
    ):
        if _ny_versatz_h(tag) != erwartet:
            befunde.append(f"Versatz am {tag}: {_ny_versatz_h(tag)} statt {erwartet}")
    fenster = fenster_je_tag(PROBE_KALENDER)
    # 09:45 EDT ist 13:45 UTC, 16:00 EDT 20:00 UTC; 09:45 EST ist 14:45 UTC, 13:00 EST 18:00 UTC.
    for tag, soll in (("2026-09-28", ["13:45", "20:00"]), ("2026-11-27", ["14:45", "18:00"])):
        ist = [datetime.fromtimestamp(s, timezone.utc).strftime("%H:%M") for s in fenster[tag]]
        if ist != soll:
            befunde.append(f"Fenster am {tag}: {ist} statt {soll}")
    # Die Bedingung: jede gestellte Lage lokal, mit derselben ENTSCHEIDUNG wie in HA.
    lagen = faelle(fenster)
    for name, lage, soll in lagen:
        if _entscheide_lokal(fenster, lage) is not soll:
            befunde.append(f"Bedingung {name}: soll {'ausloesen' if soll else 'schweigen'}")
    live = bedingung(fenster)
    for name, ausdruck in EINGABEN_LIVE.items():
        if "{% set " + name + " = " + ausdruck + " %}" not in live:
            befunde.append(f"die Automation liest {name} nicht wie vorgesehen")
    if _ohne_eingaben(live) != _ohne_eingaben(bedingung(fenster, gestellt(**lagen[0][1]))):
        befunde.append("die Rechenprobe entschiede mit einem anderen Ausdruck als die Automation")
    roh = json.dumps(automation(fenster, TROCKEN_DIENST))
    anders = json.dumps(automation(fenster, "notify.anderer_dienst"))
    konf = config_block(True)
    geheim = secrets_block("https://ntfy.example/" + TROCKEN_THEMA, TROCKEN_TOKEN)
    if TROCKEN_THEMA in roh or TROCKEN_THEMA in konf:
        befunde.append("das Thema stuende in der Automation oder in configuration.yaml")
    if TROCKEN_TOKEN in roh or TROCKEN_TOKEN in konf:
        befunde.append("der Token stuende in der Automation oder in configuration.yaml")
    if TROCKEN_THEMA not in geheim or f"Bearer {TROCKEN_TOKEN}" not in geheim:
        befunde.append("Thema oder Token fehlen im Block fuer secrets.yaml")
    if roh.count(TROCKEN_DIENST) != 1 or "notify.anderer_dienst" not in anders:
        befunde.append("der App-Dienst kommt nicht aus dem Argument")
    if TROCKEN_DIENST in anders:
        befunde.append("der App-Dienst steht fest in der Automation")
    if f"!secret {SECRET_URL}" not in konf or f"!secret {SECRET_AUTH}" not in konf:
        befunde.append("configuration.yaml nennt die Geheimnisse nicht per !secret")
    if f"!secret {SECRET_AUTH}" in config_block(False):
        befunde.append("ohne Token verwiese configuration.yaml auf ein fehlendes Geheimnis")
    if ohne_block(anhaengen("a: 1\n", konf)) != "a: 1\n":
        befunde.append("entfernen liesse vom Block etwas stehen")
    if ohne_block(konf) != "":
        befunde.append("eine Datei nur mit dem Block bliebe nicht leer")
    if not FREMDES_REST_COMMAND.search(ohne_block("x: 1\nrest_command:\n  y: 2\n")):
        befunde.append("ein fremdes rest_command wuerde uebersehen")
    if FREMDES_REST_COMMAND.search(ohne_block(anhaengen("x: 1\n", konf))):
        befunde.append("der eigene Block gaelte als fremdes rest_command")
    # Der WSL-Teil, ohne ihn zu starten: seine Regeln und der Weg ueber die Pipe.
    if _wsl_pfad(r"D:\Claude\.vault-tools\x\y.py") != "/mnt/d/Claude/.vault-tools/x/y.py":
        befunde.append("der Pfad fuer WSL ist falsch umgesetzt")
    try:
        wsl_teil.zugang({"ntfy": {"thema": "kein thema!", "token": ""}})
        befunde.append("ein Thema ausserhalb der Regel ginge durch")
    except wsl_teil.Befund as befund:
        if "kein thema" in str(befund):
            befunde.append("der Befund des WSL-Teils nennt den Wert")
    try:
        wsl_teil.pruefe_kalender([{"date": "2026-09-28", "open": "09:30", "close": "4pm"}])
        befunde.append("ein Kalender mit falscher Uhrzeit ginge durch")
    except wsl_teil.Befund:
        pass
    try:
        zurueck = pruefe_wsl_ausgabe(
            json.loads(wsl_teil.ausgabe(TROCKEN_THEMA, TROCKEN_TOKEN, PROBE_KALENDER))
        )
        if zurueck != {"thema": TROCKEN_THEMA, "token": TROCKEN_TOKEN, "kalender": PROBE_KALENDER}:
            befunde.append("die Zeile des WSL-Teils kommt nicht unveraendert an")
    except (Voraussetzung, wsl_teil.Befund) as fehler:
        befunde.append(f"die Zeile des WSL-Teils wird abgelehnt: {fehler}")
    try:
        pruefe_wsl_ausgabe({"thema": TROCKEN_THEMA})
        befunde.append("eine unvollstaendige Zeile des WSL-Teils ginge durch")
    except Voraussetzung:
        pass
    # Die Plausibilitaet der Live-Eingaben, mit gestellten Antworten: Mittag in Berlin.
    jetzt_probe = datetime(2026, 9, 28, 10, 0, tzinfo=timezone.utc).timestamp()
    gut = {
        "heute": datetime.fromtimestamp(jetzt_probe).strftime("%Y-%m-%d"),
        "jetzt": jetzt_probe,
        "gesendet": jetzt_probe - 100,
    }
    if eingaben_befunde(gut, jetzt_probe - 100, jetzt_probe):
        befunde.append("plausible Live-Eingaben gaelten als Befund")
    for name, schlecht, zeichen in (
        ("gesendet_null", dict(gut, gesendet=0.0), None),
        ("uhr_daneben", dict(gut, jetzt=jetzt_probe + 600), jetzt_probe - 100),
        ("falscher_tag", dict(gut, heute="2026-09-27"), jetzt_probe - 100),
        ("anderer_stempel", gut, jetzt_probe - 5000),
        ("jetzt_kein_wert", dict(gut, jetzt=None), jetzt_probe - 100),
    ):
        if not eingaben_befunde(schlecht, zeichen, jetzt_probe):
            befunde.append(f"Live-Eingaben {name} gingen durch")
    # Die Spur, mit gestellten Antworten von trace/list und trace/get.
    in_ordnung = {"timestamp": {"start": "x"}, "trace": {"condition/0": [{"result": {"result": False}}]}}
    for name, liste, lauf, soll in (
        ("ohne_lauf", [], None, False),
        ("in_ordnung", [{"run_id": "1"}], in_ordnung, False),
        ("renderfehler", [{"run_id": "1"}], {"trace": {"condition/0": [{"error": "UndefinedError"}]}}, True),
        ("ohne_bedingung", [{"run_id": "1"}], {"trace": {}}, True),
        ("ohne_wahrheitswert", [{"run_id": "1"}], {"trace": {"condition/0": [{"result": {}}]}}, True),
    ):
        if spur_befund(liste, lauf)[1] is not soll:
            befunde.append(f"Spur {name}: soll {'Befund' if soll else 'kein Befund'}")
    for befund in befunde:
        print(f"BEFUND {befund}")
    print(
        f"trocken: {len(fenster)} Tage, {len(lagen)} Lagen, Automation {len(roh)} Zeichen, "
        f"{len(befunde)} Befund(e)"
    )
    return 2 if befunde else 0


def _nur_zwei_dateien(pfad: object, *, datei: bool) -> str:
    """Ersetzt config_datei.pruefe_pfad fuer dieses Werkzeug: genau zwei Dateien in /config."""
    from config_datei import PfadVerboten

    if pfad not in ERLAUBT:
        raise PfadVerboten(f"Pfad {pfad!r} ist keine der Dateien {ERLAUBT}")
    return str(pfad)


def _dienste(ha) -> dict:
    return ha.befehl_ok("get_services") or {}


def bedingung_pruefen(ha) -> int:
    """Die Rechenprobe: Home Assistant rechnet jede gestellte Lage per render_template.

    Lesend (render_template steht in der Leseliste des HA-Projekts). Gerechnet wird mit dem
    erfundenen PROBE_KALENDER und derselben ENTSCHEIDUNG wie in der Automation; nur die vier
    Eingaben sind gestellt. Zurueck kommt die Zahl der Befunde.
    """
    from vorlage_auswerten import auswerten

    fenster = fenster_je_tag(PROBE_KALENDER)
    befunde = 0
    for name, lage, soll in faelle(fenster):
        nachrichten = ha.roh_sammeln(
            "render_template",
            5.0,
            template=bedingung(fenster, gestellt(**lage)),
            strict=True,
            report_errors=True,
        )
        art, wert = auswerten(nachrichten)
        ist = _wahrheit(wert) if art == "ERGEBNIS" else None
        zeile = (
            f"Bedingung {name}: soll {'ausloesen' if soll else 'schweigen'}, "
            f"Home Assistant: {art} {wert!r}"
        )
        if ist is soll:
            print(zeile)
        else:
            befunde += 1
            print("BEFUND " + zeile)
    return befunde


def eingaben_befunde(werte: dict, zeichen_ha: object, jetzt_hier: float) -> list[str]:
    """Plausibilitaet der drei live gerechneten Eingaben ohne this (Nachkritik 2026-09-24).

    werte: heute, jetzt und gesendet, wie render_template sie liefert. zeichen_ha: das
    Attribut timestamp des Lebenszeichens aus get_states. jetzt_hier: time.time() dieses
    Rechners; er steht in derselben Zeitzone wie Home Assistant (Berlin).
    """
    befunde = []
    try:
        jetzt = float(werte["jetzt"])
    except (TypeError, ValueError, KeyError):
        return [f"jetzt ist keine Zahl: {werte.get('jetzt')!r}"]
    if abs(jetzt - jetzt_hier) > UHR_TOLERANZ_S:
        befunde.append(f"jetzt weicht {jetzt - jetzt_hier:+.0f} s von der Uhr dieses Rechners ab")
    heute_hier = datetime.fromtimestamp(jetzt).strftime("%Y-%m-%d")
    if str(werte.get("heute")) != heute_hier:
        befunde.append(f"heute ist {werte.get('heute')!r}, nach jetzt in Ortszeit {heute_hier}")
    try:
        gesendet = float(werte["gesendet"])
    except (TypeError, ValueError, KeyError):
        return befunde + [f"gesendet ist keine Zahl: {werte.get('gesendet')!r}"]
    if gesendet <= 0.0:
        befunde.append("gesendet ist 0: das Attribut timestamp des Lebenszeichens fehlt oder ist leer")
    elif not isinstance(zeichen_ha, (int, float)) or abs(gesendet - float(zeichen_ha)) > 1.0:
        befunde.append(f"gesendet ist nicht der Zeitstempel des Lebenszeichens ({zeichen_ha!r})")
    elif gesendet > jetzt + UHR_TOLERANZ_S:
        befunde.append("gesendet liegt in der Zukunft")
    return befunde


def eingaben_pruefen(ha, zeichen_ha: object) -> int:
    """Rechnet heute, jetzt und gesendet live per render_template, wie die Automation sie liest.

    zuletzt liest this und ist nur im Lauf der Automation definiert; das belegt spur_pruefen.
    Zurueck kommt die Zahl der Befunde.
    """
    from vorlage_auswerten import auswerten

    werte: dict[str, object] = {}
    for name in ("heute", "jetzt", "gesendet"):
        nachrichten = ha.roh_sammeln(
            "render_template",
            5.0,
            template="{{ " + EINGABEN_LIVE[name] + " }}",
            strict=True,
            report_errors=True,
        )
        art, wert = auswerten(nachrichten)
        if art != "ERGEBNIS":
            print(f"BEFUND Eingabe {name}: Home Assistant: {art} {wert!r}")
            return 1
        werte[name] = wert
    befunde = eingaben_befunde(werte, zeichen_ha, time.time())
    for befund in befunde:
        print(f"BEFUND Eingabe {befund}")
    if not befunde:
        abstand = float(werte["jetzt"]) - float(werte["gesendet"])
        print(f"Eingaben live: heute {werte['heute']}, jetzt und gesendet plausibel (Abstand {abstand:.0f} s)")
    return len(befunde)


def spur_befund(liste: object, lauf: object) -> tuple[str, bool]:
    """Die juengste Spur der Automation: (Zeile, ob Befund).

    liste ist die Antwort von trace/list, lauf die von trace/get fuer deren juengsten Lauf.
    Belegt, was render_template nicht kann: dass die Bedingung im Lauf der Automation mit
    this rechnet (Home Assistant 2026.9.2: automation/__init__.py:707-711 setzt this,
    helpers/trace.py:113-118 schreibt error oder result in den Schritt).
    """
    if not isinstance(liste, list) or not liste:
        return "Spur: noch kein Lauf (die Automation prueft alle fuenf Minuten)", False
    if not isinstance(lauf, dict) or not isinstance(lauf.get("trace"), dict):
        return f"Spur: Form unbekannt ({type(lauf).__name__})", True
    schritte = lauf["trace"].get("condition/0")
    if not isinstance(schritte, list) or not schritte or not isinstance(schritte[0], dict):
        return "Spur: kein Schritt condition/0 im juengsten Lauf", True
    schritt = schritte[0]
    if "error" in schritt:
        return f"Spur: Bedingung mit Fehler: {str(schritt['error'])[:200]}", True
    ergebnis = schritt.get("result")
    wert = ergebnis.get("result") if isinstance(ergebnis, dict) else None
    if not isinstance(wert, bool):
        return f"Spur: Bedingung ohne Wahrheitswert ({str(ergebnis)[:200]})", True
    stempel = lauf.get("timestamp")
    start = stempel.get("start", "?") if isinstance(stempel, dict) else "?"
    return f"Spur: juengster Lauf {start}, Bedingung {wert}, kein Fehler", False


def spur_pruefen(ha) -> int:
    """Liest die juengste Spur der Automation (trace/list, trace/get). 1 bei Befund."""
    liste = ha.befehl_ok("trace/list", domain="automation", item_id=AUTOMATION_ID)
    lauf = None
    laeufe = [eintrag for eintrag in liste or [] if isinstance(eintrag, dict)]
    if laeufe:
        juengster = max(
            laeufe, key=lambda eintrag: str((eintrag.get("timestamp") or {}).get("start", ""))
        )
        lauf = ha.befehl_ok(
            "trace/get", domain="automation", item_id=AUTOMATION_ID, run_id=juengster.get("run_id")
        )
    zeile, befund = spur_befund(laeufe, lauf)
    print(("BEFUND " if befund else "") + zeile)
    return 1 if befund else 0


def pruefen(ha, datei) -> int:
    dienste = _dienste(ha)
    rest = dienste.get("rest_command", {})
    print(f"rest_command.{REST_DIENST}: {'vorhanden' if REST_DIENST in rest else 'fehlt'}")
    print(f"rest_command.reload: {'vorhanden' if 'reload' in rest else 'fehlt'}")
    app = sorted(name for name in dienste.get("notify", {}) if name.startswith("mobile_app_"))
    print("notify-Dienste der App: " + (", ".join(f"notify.{n}" for n in app) or "keiner"))
    for name in ERLAUBT:
        inhalt = datei.lies(name)
        text = "" if inhalt is None else inhalt.decode("utf-8")
        zeile = f"{name}: {'fehlt' if inhalt is None else 'vorhanden'}, Block des Totmanns "
        zeile += "da" if MARKE_ANFANG in text else "nicht da"
        if name == "configuration.yaml":
            fremd = FREMDES_REST_COMMAND.search(ohne_block(text))
            zeile += f", fremdes rest_command {'ja' if fremd else 'nein'}"
        print(zeile)
    zustaende = ha.befehl_ok("get_states") or []
    auto = [s for s in zustaende if s.get("attributes", {}).get("id") == AUTOMATION_ID]
    if auto:
        print(
            f"Automation: {auto[0]['entity_id']} = {auto[0]['state']}, "
            f"{auto[0]['attributes'].get('friendly_name')}"
        )
    else:
        print("Automation: fehlt")
    zeichen = [s for s in zustaende if s.get("entity_id") == LEBENSZEICHEN]
    stempel = zeichen[0].get("attributes", {}).get("timestamp") if zeichen else None
    if isinstance(stempel, (int, float)):
        print(f"Lebenszeichen: vor {time.time() - stempel:.0f} s")
    else:
        print("Lebenszeichen: nicht lesbar")
    live = eingaben_pruefen(ha, stempel)
    if auto:
        live += spur_pruefen(ha)
    befunde = bedingung_pruefen(ha)
    print(f"Rechenprobe der Bedingung: {len(faelle(fenster_je_tag(PROBE_KALENDER)))} Lagen, {befunde} Befund(e)")
    return 2 if befunde or live else 0


def einrichten(ha, datei, trading: Path, app_dienst: str, zugang: dict) -> int:
    dienste = _dienste(ha)
    if app_dienst.split(".", 1)[1] not in dienste.get("notify", {}):
        print(f"ABBRUCH: {app_dienst} steht nicht in der Dienstliste von Home Assistant")
        return 3
    thema, zugriff = zugang["thema"], zugang["token"]
    paper = tomllib.loads((trading / "config" / "paper.toml").read_text(encoding="utf-8"))
    ntfy_url = str(paper["benachrichtigung"]["ntfy_url"]).rstrip("/")
    fenster = fenster_je_tag(zugang["kalender"])
    if not fenster:
        print("ABBRUCH: Der Kalender von Alpaca nennt keinen Handelstag")
        return 3
    print(f"Kalender: {len(fenster)} Handelstage, {min(fenster)} bis {max(fenster)}")

    alt = {name: datei.lies(name) for name in ERLAUBT}
    if alt["configuration.yaml"] is None:
        print("ABBRUCH: configuration.yaml nicht lesbar")
        return 3
    konf_alt = alt["configuration.yaml"].decode("utf-8")
    geh_alt = (alt["secrets.yaml"] or b"").decode("utf-8")
    if FREMDES_REST_COMMAND.search(ohne_block(konf_alt)):
        print("ABBRUCH: configuration.yaml hat schon einen Abschnitt rest_command; von Hand zusammenfuehren")
        return 3
    neu = {
        "secrets.yaml": anhaengen(
            ohne_block(geh_alt), secrets_block(f"{ntfy_url}/{thema}", zugriff)
        ),
        "configuration.yaml": anhaengen(ohne_block(konf_alt), config_block(bool(zugriff))),
    }
    vorher = {"secrets.yaml": geh_alt, "configuration.yaml": konf_alt}
    geschrieben = []
    for name in ERLAUBT:
        if neu[name] != vorher[name]:
            datei.schreibe_und_pruefe(name, neu[name].encode("utf-8"))
            geschrieben.append(name)
    status, antwort = ha.http_post("/api/config/core/check_config", {})
    ergebnis = antwort.get("result") if isinstance(antwort, dict) else None
    if status != 200 or ergebnis != "valid":
        fehler = antwort.get("errors") if isinstance(antwort, dict) else antwort
        print("BEFUND Konfiguration ungueltig: " + _ohne(str(fehler), thema, zugriff)[:300])
        for name in geschrieben:
            datei.schreibe_und_pruefe(name, alt[name] or b"")
        print(f"Zurueckgeschrieben: {geschrieben}")
        return 2
    print(f"Geschrieben: {geschrieben or 'nichts, die Bloecke standen schon da'}; Konfiguration gueltig")

    rest = _dienste(ha).get("rest_command", {})
    if REST_DIENST not in rest and "reload" in rest:
        ha.befehl_ok("call_service", domain="rest_command", service="reload", service_data={})
        rest = _dienste(ha).get("rest_command", {})
    if REST_DIENST not in rest:
        print(
            f"rest_command.{REST_DIENST} ist noch nicht geladen: Home Assistant neu starten "
            "(Einstellungen, System, Neu starten), danach einrichten noch einmal -- die Bloecke "
            "stehen dann schon da."
        )
        return 4

    status, antwort = ha.http_post(
        f"/api/config/automation/config/{AUTOMATION_ID}", automation(fenster, app_dienst)
    )
    print(f"Automation angelegt oder ersetzt: HTTP {status}")
    if status != 200:
        print("ABBRUCH: Die Automation wurde nicht angenommen: " + _ohne(str(antwort), thema, zugriff)[:200])
        return 2
    return 0


def probe(ha, app_dienst: str) -> int:
    dienste = _dienste(ha)
    if REST_DIENST not in dienste.get("rest_command", {}):
        print(f"ABBRUCH: rest_command.{REST_DIENST} fehlt; erst einrichten")
        return 3
    name = app_dienst.split(".", 1)[1]
    if name not in dienste.get("notify", {}):
        print(f"ABBRUCH: {app_dienst} steht nicht in der Dienstliste von Home Assistant")
        return 3
    ha.befehl_ok(
        "call_service", domain="rest_command", service=REST_DIENST, service_data={"nachricht": PROBE_TEXT}
    )
    ha.befehl_ok(
        "call_service",
        domain="notify",
        service=name,
        service_data={"title": "Trading Totmann (Probe)", "message": PROBE_TEXT},
    )
    print("Zwei Probemeldungen abgeschickt (ntfy und HA-App). Ob sie ankamen, sieht nur Pascal.")
    return 0


def entfernen(ha, datei) -> int:
    status = ha.http_delete(f"/api/config/automation/config/{AUTOMATION_ID}")
    print(f"Automation geloescht: HTTP {status}")
    for name in ERLAUBT:
        inhalt = datei.lies(name)
        if inhalt is None:
            print(f"{name}: fehlt")
            continue
        text = inhalt.decode("utf-8")
        if MARKE_ANFANG not in text:
            print(f"{name}: kein Block")
            continue
        datei.schreibe_und_pruefe(name, ohne_block(text).encode("utf-8"))
        print(f"{name}: Block entfernt")
    status, antwort = ha.http_post("/api/config/core/check_config", {})
    print("Konfiguration: " + str(antwort.get("result") if isinstance(antwort, dict) else f"HTTP {status}"))
    rest = _dienste(ha).get("rest_command", {})
    if "reload" in rest:
        ha.befehl_ok("call_service", domain="rest_command", service="reload", service_data={})
        print("rest_command neu geladen")
    else:
        print(
            f"rest_command.{REST_DIENST} bleibt bis zum naechsten Neustart von Home Assistant "
            "geladen; ohne Automation ruft ihn niemand."
        )
    return 0


# Die Befehle, die pruefen an Home Assistant schickt. Alle lesen nur.
LESEND = frozenset({"get_services", "get_states", "trace/list", "trace/get", "render_template"})


class _NurLesend:
    """Die Verbindung fuer pruefen: laesst nur LESEND und das Lesen von Dateien durch.

    Seit der Pruefung Z2 (2026-09-25). Vorher lief pruefen mit nur_lesen=False, und ein
    Fehler in pruefen, der einen schreibenden Befehl schickte, waere nicht abgefangen
    worden. Ueber supervisor/api geht nur GET und das Anlegen der Ingress-Sitzung
    (ENDPUNKT_SITZUNG), das ConfigDatei auch lesend braucht. Jede andere Methode der
    Verbindung -- http_post, http_delete -- ist hier gesperrt.
    """

    def __init__(self, ha, sitzung_endpunkt: str) -> None:
        self._ha = ha
        self._sitzung = sitzung_endpunkt

    def _pruefe(self, typ: str, felder: dict) -> None:
        if typ in LESEND:
            return
        if typ == "supervisor/api":
            methode = str(felder.get("method", "")).lower()
            if methode == "get":
                return
            if methode == "post" and felder.get("endpoint") == self._sitzung:
                return
        raise RuntimeError(f"pruefen ist lesend: {typ} {felder.get('method', '')} gesperrt".strip())

    def befehl_ok(self, typ: str, **felder):
        self._pruefe(typ, felder)
        return self._ha.befehl_ok(typ, **felder)

    def roh_sammeln(self, typ: str, dauer: float, **felder):
        self._pruefe(typ, felder)
        return self._ha.roh_sammeln(typ, dauer, **felder)

    def __getattr__(self, name: str):
        raise RuntimeError(f"pruefen ist lesend: {name} gesperrt")


def online(argumente) -> int:
    ha_projekt, trading = Path(argumente.ha_projekt), Path(argumente.trading)
    zugang = None
    if argumente.aktion == "einrichten":
        # Vor der Verbindung zu Home Assistant: Der WSL-Teil braucht bis zu WSL_TIMEOUT_S,
        # und so lange soll keine Verbindung offen warten.
        zugang = aus_wsl(argumente.tage)
        print(
            f"Aus WSL: Thema Laenge {len(zugang['thema'])}, Token Laenge "
            f"{len(zugang['token'])}, {len(zugang['kalender'])} Handelstage"
        )
    sys.path.insert(0, str(ha_projekt / "src"))
    import config_datei
    import ha_verbindung as hv

    # Die Pfadregel des HA-Projekts erlaubt nur themes/. Dieses Werkzeug braucht genau
    # secrets.yaml und configuration.yaml und ersetzt die Regel dafuer zur Laufzeit; die
    # Methoden von ConfigDatei schlagen pruefe_pfad erst beim Aufruf nach.
    config_datei.pruefe_pfad = _nur_zwei_dateien
    token = hv.lies_token()
    print(f"Token gelesen (Laenge {len(token)}), Instanz aus dem HA-Projekt, Aktion {argumente.aktion}")
    # nur_lesen=False auch fuer pruefen: get_services, get_states, trace/list und trace/get
    # stehen nicht in der Leseliste des HA-Projekts (src/ha_verbindung.py, LESELISTE; dort
    # nur render_template). Fuer pruefen sperrt statt dessen _NurLesend alles ausser genau
    # diesen Befehlen (Pruefung Z2); die Dateien oeffnet es ueber ConfigDatei mit
    # nur_lesen=True.
    with hv.HaVerbindung(hv.BASIS_URL, token, nur_lesen=False) as verbindung:
        ha = (
            _NurLesend(verbindung, config_datei.ENDPUNKT_SITZUNG)
            if argumente.aktion == "pruefen"
            else verbindung
        )
        datei = config_datei.ConfigDatei(
            hv.BASIS_URL,
            lambda endpunkt, methode: ha.befehl_ok("supervisor/api", endpoint=endpunkt, method=methode),
            nur_lesen=argumente.aktion == "pruefen",
        )
        with datei:
            if argumente.aktion == "pruefen":
                return pruefen(ha, datei)
            if argumente.aktion == "einrichten":
                return einrichten(ha, datei, trading, argumente.app_dienst, zugang)
            if argumente.aktion == "probe":
                return probe(ha, argumente.app_dienst)
            return entfernen(ha, datei)


def main(argv=None) -> int:
    zerleger = argparse.ArgumentParser(
        description="Totmann der Trading Plattform in Home Assistant (Paket Z2).",
        allow_abbrev=False,
    )
    zerleger.add_argument("aktion", choices=["pruefen", "trocken", "einrichten", "probe", "entfernen"])
    zerleger.add_argument("ha_projekt", nargs="?")
    zerleger.add_argument("trading", nargs="?")
    zerleger.add_argument("--app-dienst", default=None)
    zerleger.add_argument("--tage", type=int, default=400)
    argumente = zerleger.parse_args(argv)
    if argumente.aktion == "trocken":
        return trocken()
    if not argumente.ha_projekt or not argumente.trading:
        zerleger.error("pruefen, einrichten, probe und entfernen brauchen beide Projektordner")
    if argumente.aktion in ("einrichten", "probe") and (
        argumente.app_dienst is None or not APP_DIENST_MUSTER.match(argumente.app_dienst)
    ):
        zerleger.error("--app-dienst notify.<name> fehlt oder hat nicht diese Form")
    try:
        return online(argumente)
    except (Voraussetzung, wsl_teil.Befund) as fehler:
        # Diese Texte tragen keinen Wert aus der Geheimnisdatei, nur Feldnamen und Laengen.
        print(f"ABBRUCH: {fehler}")
        return 3
    except Exception as ausnahme:
        # Nur der Typ: Eine Ausnahme kann eine Adresse mit dem Thema oder einen Kopf mit
        # einem Schluessel tragen.
        print(f"ABBRUCH: {type(ausnahme).__name__}")
        return 2


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