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

"""Eigener Lauf zu Paket D: Der echte Melder schickt ueber echtes HTTP an einen lokalen ntfy-Probeserver.

Koordinator-Werkzeug, nicht fuers Repo. Frage (Herkunft: Anforderung Paket D vom 2026-09-18, "Lauf mit echten
Daten"): Was geht wirklich ueber die Leitung, wenn der Runner und der Stille-Waechter mit dem ausgelieferten
Melder arbeiten -- wie viele Anfragen, welche Titel und Prioritaeten, steht ein Kontowert oder das Thema im Text,
und was passiert bei abgelehnten und bei langsamen Antworten? Die Tests des Plans spritzen einen Transport ein;
hier laeuft requests.post gegen 127.0.0.1.
Der Sitzungsaufbau (Probe-Broker, simulierte Uhr, Wegwerf-Register) kommt aus den Helfern der Testdatei
tests/test_paper_sitzung_melder.py des Codebaums; der Melder ist der echte, ohne eingespritzten Transport.
Was es nicht sieht: ntfy.sh selbst (Themenregeln des Dienstes, Ratenbegrenzung, Zustellung aufs Telefon) und den
echten Broker -- dafuer braucht es ein Thema von Pascal und einen Handelstag.
Regressionsnachweis: --sabotage erwartet im ersten Fall absichtlich zwei Meldungen statt einer und muss rot enden.

Aufruf (WSL, PYTHONPATH=<codebaum>, Arbeitsverzeichnis beliebig): python d_eigenlauf.py <codebaum> [--sabotage]
Exit 0 ohne Befund, 2 mit Befund.
"""

import json
import os
import subprocess
import sys
import tempfile
import time
from datetime import UTC
from datetime import datetime
from datetime import timedelta
from pathlib import Path

CODE = Path(sys.argv[1])
sys.path.insert(0, str(CODE))
sys.path.insert(0, str(CODE / "tests"))

import test_paper_sitzung_melder as helfer  # noqa: E402
from trading.broker.schnittstelle import BrokerFehler  # noqa: E402
from trading.dashboard.waechter import StilleWaechter  # noqa: E402
from trading.paper.melder import Melder  # noqa: E402
from trading.paper.melder import MelderConfig  # noqa: E402

SERVER = Path(__file__).with_name("ntfy_probe_server.py")
THEMA = "lokales-probe-thema"
TOKEN = "tk_nur_fuer_die_probe"
befunde: list[str] = []


class Probeserver:
    def __init__(self, port: int, *zusatz: str) -> None:
        self.port = port
        self.ordner = tempfile.mkdtemp(prefix="ntfy-probe-")
        self.mitschrift = Path(self.ordner) / "mitschrift.jsonl"
        self.prozess = subprocess.Popen(
            [sys.executable, str(SERVER), str(port), str(self.mitschrift), *zusatz],
            stdout=subprocess.DEVNULL,
        )
        time.sleep(0.8)

    def zeilen(self) -> list[dict]:
        if not self.mitschrift.exists():
            return []
        with open(self.mitschrift, encoding="utf-8") as datei:
            return [json.loads(zeile) for zeile in datei if zeile.strip()]

    def ende(self) -> None:
        self.prozess.terminate()
        self.prozess.wait(timeout=5)


def melder(port: int, zeitlimit_s: float = 3.0) -> Melder:
    return Melder(
        config=MelderConfig(
            aktiv=True,
            ntfy_url=f"http://127.0.0.1:{port}",
            thema=THEMA,
            zeitlimit_s=zeitlimit_s,
            stille_nach_s=180.0,
        ),
        token=TOKEN,
    )


def pruefe(name: str, bedingung: bool, einzelheit: str) -> None:
    print(f"  {'ok    ' if bedingung else 'BEFUND'} {name}: {einzelheit}")
    if not bedingung:
        befunde.append(name)


def auf_der_leitung(name: str, zeilen: list[dict], erwartete_titel: list[str]) -> None:
    titel = [z["title"] for z in zeilen]
    pruefe(f"{name}: Titel", titel == erwartete_titel, f"{titel}")
    for z in zeilen:
        text = z["text"]
        pruefe(f"{name}: Pfad ist das Thema", z["pfad"] == f"/{THEMA}", z["pfad"])
        pruefe(f"{name}: Titel ist ASCII", z["title"].isascii(), z["title"])
        pruefe(f"{name}: Bearer-Kopf da", z["authorization_schema"] == "Bearer", f"Laenge {z['authorization_laenge']}")
        pruefe(f"{name}: weder Thema noch Token im Text", THEMA not in text and TOKEN not in text, f"{len(text)} Zeichen")
        pruefe(
            f"{name}: kein Kontowert im Text",
            "100000" not in text.replace(".", "").replace(",", "") and "Konto" not in text and "cash" not in text.lower(),
            text.replace("\n", " | ")[:160],
        )


def fall_sitzung(name, port, server_zusatz, erwartete_titel, sabotage=False, zeitlimit_s=3.0, **aufbau_kw):
    print(f"== {name}")
    server = Probeserver(port, *server_zusatz)
    try:
        with tempfile.TemporaryDirectory() as ordner:
            ohne = helfer.Aufbau(Path(ordner) / "ohne", melder=None, **_kw(aufbau_kw))
            code_ohne = ohne.sitzung.laufe()
            mit = helfer.Aufbau(Path(ordner) / "mit", melder=melder(port, zeitlimit_s), **_kw(aufbau_kw))
            beginn = time.monotonic()
            code_mit = mit.sitzung.laufe()
            dauer = time.monotonic() - beginn
            pruefe(f"{name}: Exit-Code wie ohne Melder", code_mit == code_ohne, f"{code_mit} gegen {code_ohne}")
            pruefe(f"{name}: Grund wie ohne Melder", mit.ende()["grund"] == ohne.ende()["grund"], mit.ende()["grund"])
            zeilen = server.zeilen()
            if sabotage:
                erwartete_titel = [*erwartete_titel, "Paper-Runner beendet"]
            auf_der_leitung(name, zeilen, erwartete_titel)
            return mit, zeilen, dauer
    finally:
        server.ende()


def _kw(aufbau_kw: dict) -> dict:
    kw = dict(aufbau_kw)
    kw.setdefault("kerzen", helfer._kerzen())
    kw.setdefault("agent", helfer.Agent(aktion=3))
    return kw


def main() -> int:
    sabotage = "--sabotage" in sys.argv
    basis = 18200 + (os.getpid() % 500)

    fall_sitzung("ruhiger Tag", basis, (), ["Paper-Runner beendet"], sabotage=sabotage)
    fall_sitzung(
        "Watchdog haelt an", basis + 1, (), ["Paper-Runner pausiert", "Paper-Runner beendet"],
        wirft_bei={"letzter_kurs": BrokerFehler("500 Internal Server Error")},
    )
    fall_sitzung(
        "Tagesverlust", basis + 2, (), ["Grenze verletzt", "Paper-Runner beendet"],
        kerzen=helfer._kerzen(60, trend=-0.15, sigma=0.0), agent=helfer.Agent(aktion=4),
        config=helfer._config(basis_menge=200), limits=helfer._limits(max_tagesverlust_anteil=0.002),
    )

    mit, zeilen, _ = fall_sitzung(
        "Server lehnt ab (403)", basis + 3, ("--status", "403"), ["Paper-Runner gestoert", "Paper-Runner beendet"],
        fuellung="nie",
    )
    fehlschlaege = [z.get("grund") or z.get("text") or z for z in mit.fehlschlaege()]
    pruefe("403: je Meldung eine Protokollzeile", len(mit.fehlschlaege()) == len(zeilen) == 2, f"{fehlschlaege}")
    pruefe("403: die Zeile nennt den Status und nicht das Thema", all("403" in json.dumps(z) and THEMA not in json.dumps(z) for z in mit.fehlschlaege()), "geprueft an json.dumps der Zeilen")

    mit, zeilen, dauer = fall_sitzung(
        "Server antwortet zu langsam", basis + 4, ("--verzoegerung", "2.5"), ["Paper-Runner beendet"], zeitlimit_s=1.0,
    )
    pruefe("langsam: Fehlschlag protokolliert", len(mit.fehlschlaege()) == 1, f"{[json.dumps(z)[:120] for z in mit.fehlschlaege()]}")
    pruefe("langsam: der Lauf wartet hoechstens das Zeitlimit", dauer < 2.4, f"{dauer:.2f} s fuer den ganzen Tag mit einer Meldung, Zeitlimit 1,0 s")

    mit, zeilen, _ = fall_sitzung(
        "Server leitet weiter (307)", basis + 6, ("--status", "307"), ["Paper-Runner beendet"],
    )
    pruefe("307: genau eine Anfrage, keiner Weiterleitung gefolgt", len(zeilen) == 1 and zeilen[0]["pfad"] == f"/{THEMA}", f"{[z['pfad'] for z in zeilen]}")
    pruefe("307: als Fehlschlag protokolliert", [z.get("text") for z in mit.fehlschlaege()] == ["HTTP 307"], f"{[z.get('text') for z in mit.fehlschlaege()]}")

    print("== Stille-Waechter wiederholt bis zur Zustellung")
    server = Probeserver(basis + 5, "--status-folge", "403,403,200")
    try:
        with tempfile.TemporaryDirectory() as ordner:
            status = Path(ordner) / "status.json"
            jetzt = datetime.now(UTC)
            status.write_text(json.dumps({"zeit": (jetzt - timedelta(seconds=600)).isoformat(), "phase": "handel", "position": 0, "pid": 1}), encoding="utf-8")
            waechter = StilleWaechter(melder=melder(basis + 5), status_datei=status, stille_nach_s=180.0)
            for _ in range(5):
                waechter.pruefe()
            zeilen = server.zeilen()
            pruefe("Waechter: drei Anfragen, dann Ruhe", [z["status"] for z in zeilen] == [403, 403, 200], f"{[z['status'] for z in zeilen]}")
            pruefe("Waechter: Titel und Prioritaet", all(z["title"] == "Runner schweigt" and z["priority"] == "urgent" for z in zeilen), f"{[(z['title'], z['priority']) for z in zeilen]}")
            pruefe("Waechter: zwei Fehlschlaege gemerkt", waechter.fehlschlaege == ["HTTP 403", "HTTP 403"], f"{waechter.fehlschlaege}")
            status.write_text(json.dumps({"zeit": datetime.now(UTC).isoformat(), "phase": "handel", "position": 0, "pid": 1}), encoding="utf-8")
            waechter.pruefe()
            waechter.pruefe()
            zeilen = server.zeilen()
            pruefe("Waechter: genau eine Rueckmeldung", [z["title"] for z in zeilen[3:]] == ["Runner meldet sich wieder"], f"{[z['title'] for z in zeilen[3:]]}")
    finally:
        server.ende()

    print(f"{len(befunde)} Befund(e).")
    return 2 if befunde else 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as ausnahme:  # noqa: BLE001 - ein Absturz des Werkzeugs ist selbst ein Befund
        print(f"BEFUND: das Werkzeug ist abgebrochen: {ausnahme!r}")
        raise