Automatischer Push: 14 Datei(en) geändert

- .vault-tools/trading-plattform-pruefung/LIESMICH.md
- .vault-tools/trading-plattform-pruefung/bremse_echtlauf.py
- .vault-tools/trading-plattform-pruefung/browser_w.py
- .vault-tools/trading-plattform-pruefung/generalprobe_echtlauf.py
- .vault-tools/trading-plattform-pruefung/ha_aktiv_setzen.py
- .vault-tools/trading-plattform-pruefung/handelstag_auswertung.py
- .vault-tools/trading-plattform-pruefung/hausmelder_echtlauf.py
- .vault-tools/trading-plattform-pruefung/killswitch_simulation.py
- .vault-tools/trading-plattform-pruefung/killswitch_simulation_mutanten.py
- .vault-tools/trading-plattform-pruefung/latenz_auswertung.py
- .vault-tools/trading-plattform-pruefung/s1_alpaca_lesen.py
- .vault-tools/trading-plattform-pruefung/s5_echtlauf.py
- … und 2 weitere

Erzeugt vom PostToolUse-Hook .vault-tools/auto-push/auto_push.py.

Co-Authored-By: Claude Opus 5 <[email protected]>
This commit is contained in:
Pascal MichelandClaude Opus 5 committed 2026-09-22 04:00:01 +02:00
1 parent 246af19683
commit 593acdd122
14 files changed
+2090 -5

No files matched your search

@@ -26,12 +26,12 @@ Koordinator-Werkzeuge, nicht fürs Repo. Sie starten **nie** den echten Broker:
| Datei | Was sie tut | Aufruf |
|---|---|---|
| `start_probe_zugang.py` | die echte App mit Attrappe, Wegwerf-Register und Wegwerf-Zugang (Benutzer `probe`, bekanntes Passwort, TOTP-Beispielgeheimnis der RFC-Literatur; beides in `zugang.json` der Ablage), auf Wunsch mit einer Paper-Statusdatei | aus dem Projektordner in WSL: `PYTHONPATH=. python start_probe_zugang.py <port> <ablage> [mit-paperstatus]` |
| `start_probe_zugang.py` | die echte App mit Attrappe, Wegwerf-Register und Wegwerf-Zugang (Benutzer `probe`, bekanntes Passwort, TOTP-Beispielgeheimnis der RFC-Literatur; beides in `zugang.json` der Ablage), auf Wunsch mit einer Paper-Statusdatei. **Seit 2026-09-21 bricht es ab, wenn im Arbeitsordner eine `config/secrets.toml` liegt** — `erzeuge_app` startete bis Paket S5 immer den echten Stille-Wächter, und eine Probe-Instanz im echten Projektordner hätte nach 180 s aufs Telefon geklingelt | aus einer **Kopie** des Projektordners (ohne Geheimnisdatei) in WSL: `PYTHONPATH=. python start_probe_zugang.py <port> <ablage> [mit-paperstatus]` |
| `angriffe.py` | 56 Fälle gegen Host-Positivliste, Origin, Sitzungszwang, Cookie-Attribute, Bremse, CSP und die anderen Header, WebSocket ohne Sitzung; Regressionsnachweis gegen das Backend vor Teil B: 45 rot | `python angriffe.py <basis-url> <ablage-mit-zugang.json>` |
| `browserlauf_zugang.py` | Anmeldeformular ohne Sitzung, falsche und richtige Anmeldung (frischer TOTP-Code), Name und „Abmelden" in der Kopfzeile, Strom, Karte „Paper-Betrieb", alle Ansichten unter der CSP, Abmelden und Neuladen, 375 px; protokolliert jede Antwort ab 400 mit Phase und Pfad (`antworten-ab-400.json`); `--sabotage` muss rot enden; Screenshots — **ansehen** (der Lauf fand so das `GET /api/paper` vor dem Tor und die unbeschrifteten Felder) | `python -X utf8 browserlauf_zugang.py http://localhost:<port> <ablage>/zugang.json <ausgabeordner> [--sabotage]` |
| `pruef_beobachtung_echt.py` | `baue_beobachtung` gegen `NautilusEnv._beobachtung` auf den echten IEX-Kerzen eines Tages, beide Seiten unabhängig gerechnet; Sabotage rot. Seit 2026-09-18 mit zweitem Teil: eine feste Aktionsfolge über alle fünf Stufen, Vergleich nach jedem Schritt bei gleichem Zustand (gemessen am 2026-09-17: 120 Beobachtungen bitgleich, davon 50 long, 50 short, Kontowert in allen ungleich dem Startkapital, Zähler in 60 größer null); `--mutant-vorzeichen` (Betrag statt Vorzeichen der Position) lässt genau die 50 Short-Beobachtungen rot fallen, während der erste Teil grün bleibt | `PYTHONPATH=. python pruef_beobachtung_echt.py JJJJ-MM-TT` aus dem Projektordner |
| `lies_paper_protokoll.py` | fasst ein Paper- oder Probe-Protokoll zusammen (Takte je Phase, Killswitch je Takt, frische Kerzen, erste Entscheidung, Orders, Schlussphase, Ende) | `python lies_paper_protokoll.py <jsonl>` |
| `handelstag_auswertung.py` | wertet einen Paper-Handelstag aus: Takte je Phase, frische Kerzen, Spanne Taktzeile gegen Minutenende (Obergrenze für „Kerze abrufbar"), Orders, Ende; mit `--broker` lesend die Orders des Paper-Kontos (Füllzeit aus `submitted_at`/`filled_at` gegen `order_timeout_s`, Vorzeichen offener Positionen), mit `--kerzen` die Kosten je Order gegen den Schluss der letzten abgeschlossenen IEX-Kerze in Basispunkten. Am 2026-09-18 an den zwei Protokollen der Nacht erprobt (Probe: Spanne ausgelassen, weil die Kerzen gespielt sind) | aus dem Projektordner: `python handelstag_auswertung.py <protokoll.jsonl> [--broker] [--kerzen]` |
| `handelstag_auswertung.py` | wertet einen Paper-Handelstag aus: Takte je Phase, frische Kerzen, Spanne Taktzeile gegen Minutenende (Obergrenze für „Kerze abrufbar"), Orders, Ende; mit `--broker` lesend die Orders des Paper-Kontos (Annahme bis Füllung aus `created_at`/`filled_at` gegen `order_timeout_s`, dazu die Teilspannen bis `submitted_at` und von dort bis zur Füllung — seit dem ersten Handelstag, an dem der Broker eine Order 8,7 s zwischen Annahme und Weitergabe liegen ließ und `submitted_at` allein das verschwiegen hätte; ein Hinweis ab der halben Grenze; Vorzeichen offener Positionen), mit `--kerzen` die Kosten je Order gegen den Schluss der letzten abgeschlossenen IEX-Kerze in Basispunkten. Am 2026-09-18 an den zwei Protokollen der Nacht erprobt (Probe: Spanne ausgelassen, weil die Kerzen gespielt sind) | aus dem Projektordner: `python handelstag_auswertung.py <protokoll.jsonl> [--broker] [--kerzen]` |
| `vorstart_pruefung.py` | liest vor dem Runner-Start: läuft schon ein Runner, Killswitch-Zustand im Register (nur Auskunft — der Start stellt nach der Neustartregel immer auf `angehalten`; ein Befund ist allein `gesperrt`), Paper-Konto glatt, Marktuhr des Brokers, eigene Uhr gegen den `Date`-Kopf des Brokers. Am 2026-09-18 fand der erste trockene Lauf einen Killswitch auf `aktiv` (Klick um 08:55 Uhr), den der Start überschreibt | aus dem Projektordner: `python vorstart_pruefung.py` |
| `runner_stand.py` | zeigt den Stand eines laufenden Paper-Runners aus seinem Protokoll: Zeilen je Art, Takte je Phase und Killswitch-Zustand, letzter Takt, Order- und Ende-Zeilen. Urteilt nicht und gibt keine Kontowerte aus. Am 2026-09-18 am Protokoll der Nacht erprobt (drei Zeilen: `kopf`, `ereignis`, `ende`); an einem Protokoll mit Takten noch nicht | in WSL: `python runner_stand.py [<protokoll.jsonl>]`, ohne Angabe das neueste `*-paper.jsonl` im Projektordner |
| `protokoll_vergleich.py` | vergleicht zwei JSONL-Protokolle der Messwerkzeuge Wert für Wert (Bitgleichheit nach einem Umbau); trennt `ABWEICHUNG`, `NUR NEU` (neuer Schlüssel, kein Befund ohne `--streng`) und `NUR ALT`. Erprobt am 2026-09-18: Selbstvergleich 5.190 Werte ohne Befund, `--sabotage` (ein ULP) rot, der Ein-Jahres-Lauf mit `OMP_NUM_THREADS=8` gegen den ohne 621 Abweichungen ab der dreizehnten Stelle | `python protokoll_vergleich.py <alt.jsonl> <neu.jsonl> [--ohne …] [--streng] [--sabotage]` |
@@ -58,6 +58,27 @@ Koordinator-Werkzeuge, nicht fürs Repo. Sie starten **nie** den echten Broker:
|---|---|---|
| `ha_einrichten.py` | richtet in Home Assistant die Gegenstelle für den Hausmelder ein: acht Helfer mit dem Präfix `trading_` und eine Automation mit Webhook-Auslöser (`local_only`, nur POST). Die Webhook-ID entsteht im Skript (64 Hex-Zeichen), wird nie ausgegeben und landet nur in `[homeassistant]` der `config/secrets.toml` des Trading-Projekts, mit `aktiv = false`. `pruefen` und `lesen` lesen nur (`lesen` zeigt zusätzlich die Zustände der Helfer), `probe` schickt eine Beispiel-Nutzlast und liest zurück, `entfernen` baut Helfer und Automation wieder ab. **Läuft unter Windows mit dem Python des Home-Assistant-Projekts** — dort liegen der Klient `src/ha_verbindung.py`, die Adresse der Instanz und der Name der Token-Variable; das Token bleibt auf der Windows-Seite und kommt nie auf den Trading-Rechner. Gemessen am 2026-09-20/21: eingerichtet, `probe` rückgelesen wie gesendet, und ein POST des echten `Hausmelder` aus WSL kam in den Helfern an | `<ha-projekt>\.venv\Scripts\python.exe -X utf8 ha_einrichten.py <pruefen|lesen|einrichten|probe|entfernen> "<ha-projekt>" "<trading-projekt>"` |
## Nacht vom 2026-09-20 auf den 2026-09-21: Generalprobe, Hausmelder, YubiKey, Härtung
| Datei | Was sie tut | Aufruf |
|---|---|---|
| `generalprobe_echtlauf.py` | ruft die **echte** Funktion `_generalprobe` aus `scripts/paper.py` — dieselbe Verdrahtung wie im Betrieb — mit den echten IEX-Kerzen eines Tages (JSON aus `hole_kerzen.py`), einer Sicherungskopie des Registers (das echte wird nur lesend geöffnet), dem echten Registermodell, einem mitschreibenden „echten" Konto, das bei jedem schreibenden Aufruf wirft, einer Netzsperre auf `requests.Session.request` und einem Faden, der wie das Dashboard klickt (fortsetzen, anhalten, fortsetzen, glattstellen, freigeben). Urteilt über Exit-Code, Modus, Meldungsvorsatz, Netzversuche, Aufrufe am Konto, Position nach dem Glattstellen, Alter des Herzschlags, `tagesstart.json`. Gemessen am 2026-09-20/21 mit den Kerzen des 2026-09-18: bei 1 s je Marktminute 435 s, 389 Takte, drei Orders, 0 Befunde | in WSL: `python generalprobe_echtlauf.py <code-baum> <echter-projektordner> <kerzen.json> <wegwerf-ordner> [tempo]` |
| `hausmelder_echtlauf.py` | der echte `Hausmelder`-Faden gegen einen mitschreibenden HTTP-Server auf `127.0.0.1`: erste Nutzlast ohne Statusdatei, Änderung, Stille ohne Änderung, Lebenszeichen, Killswitch-Wechsel, HTTP 500 (Merker kippt nicht), 302 (nicht gefolgt), `veraltet`, Adresse in keinem Fehlschlag und keinem `repr`, `starte_hausmelder` über eine Geheimnisdatei. Mit `--echt <secrets.toml>` zusätzlich **ein** POST an die dort eingetragene Adresse; die Adresse wird nie ausgegeben. Zurücklesen mit `ha_einrichten.py lesen` | in WSL: `python hausmelder_echtlauf.py <code-baum> <wegwerf-ordner> [--echt <secrets.toml>]` |
| `ha_aktiv_setzen.py` | setzt `[homeassistant] aktiv` in einer Geheimnisdatei auf `true` oder `false` — genau eine Zeile, mit Probe vor dem Schreiben, dass sich nur dieser Wert ändert; gibt nur Abschnittsnamen und die Länge der Adresse aus | `python ha_aktiv_setzen.py <secrets.toml> <true|false>` |
| `start_probe_w.py` | Probe-Backend wie `start_probe_zugang.py`, dazu die Konfiguration für Sicherheitsschlüssel (RP-ID `localhost`, Origin `http://localhost:<port>`) und wahlweise ein nur gelesener Marktkatalog. **Bricht ab, wenn im Arbeitsordner eine `config/secrets.toml` liegt** — Probe-Instanzen starten nur aus einer Kopie oder einem Arbeitsbaum ohne Geheimnisse, sonst sprächen sie mit dem echten ntfy und dem echten Home Assistant | in WSL aus der Kopie: `python start_probe_w.py <port> <ablage> [<katalog>]` |
| `browser_w.py` | Browserlauf zum YubiKey-Weg: Playwright/Chromium mit einem **virtuellen Authenticator** über das DevTools-Protokoll (`WebAuthn.addVirtualAuthenticator`) gegen die Probe-Instanz unter `http://localhost:<port>` — Anmelden mit Code, Registrieren in „Zugang", Anmelden mit Passwort und Schlüssel (der Zähler des Authenticators muss steigen, sonst wurde er nie gefragt), falsches Passwort, Code-Weg danach, Entfernen, Gegenprobe unter `127.0.0.1` (409), Schlüsseldatei und Anmeldeprotokoll ohne Passwort, keine CSP- oder Permissions-Policy-Meldung. Gemessen am 2026-09-21: 17 von 17, auch gegen eine gehärtete Systemeinheit | unter Windows: `python -X utf8 browser_w.py <port> <ablage>` |
| `latenz_auswertung.py` | rechnet Reaktionszeiten aus dem, was ein Lauf ohnehin hinterlässt — nur lesend, Register mit `mode=ro`: Klick → Runner reagiert (Ereignis in `killswitch_ereignisse` mit Auslöser `pascal` gegen die erste Taktzeile mit dem neuen Zustand; obere Schranke, weil die Taktzeile am Ende des Takts entsteht), Kerze → Taktende (nur im Modus `paper`) und den Taktabstand als Rahmen. Für Tage, an denen der Runner seine Latenzen noch nicht selbst misst. Gibt nie einen Kontowert aus | in WSL aus dem Projektordner: `python latenz_auswertung.py <protokoll.jsonl> [<register.db>]` |
| `bremse_echtlauf.py` | Misst das Wachstum des Anmeldeprotokolls im gesperrten Zustand gegen eine laufende Probe-Instanz: fünf Fehlversuche, dann je 5 s Dauerfeuer auf `beginn`, `abschluss` und den Weg mit Code, gezählt in Zeilen und Bytes; zweiter Durchgang mit dem Abschluss als erster gesperrter Anfrage; dazu der 409 des Abschlusses unter `127.0.0.1`. Anlass: Der Prüfer maß am 2026-09-21 im Schlüsselweg 22 474 B/s, weil die Protokollbremse nur im Weg mit Code stand. Nach Task 229 gemessen: 2721 Anfragen in 5 s → eine Zeile | in WSL: `python bremse_echtlauf.py <port> <ablage> <beginn\|abschluss>` |
| `s5_echtlauf.py` | Gegenmessung der Dashboard-Härtung gegen eine laufende Probe-Instanz: 422 ohne die Eingabe, 512 MiB Rumpf mit und ohne `content-length` samt Spitzenspeicher des Servers aus `/proc`, Latenz von `POST /api/killswitch/anhalten` unter 40 parallelen teuren Kerzenabrufen, überlanger Grund, Sitzungswiderruf nach Passwortwechsel. Gemessen am 2026-09-21: 413 in 0,19 s ohne Speicherzuwachs (vorher 1,5 GiB), Killswitch höchstens 0,51 s (vorher 118 s) | in WSL: `python s5_echtlauf.py <port> <ablage> <code-baum>` |
| `s1_alpaca_lesen.py` | liest mit dem Alpaca-Klienten eines Code-Baums die echten Antworten des Paper-Kontos — **nur lesend, ohne einen Betrag auszugeben** (Feldnamen, Typen, Statusverteilung der Orderhistorie, Felder mit `None`). Für den Vergleich alter gegen neuer Parser | in WSL: `python s1_alpaca_lesen.py <code-baum> <echter-projektordner>` |
## Killswitch-Simulation (2026-09-21)
| Datei | Was sie tut | Aufruf |
|---|---|---|
| `killswitch_simulation.py` | der Killswitch im Zeitraffer, **ohne Netz, ohne das echte Konto und ohne das echte Register** — die echte Funktion `_generalprobe` mit den Kerzen eines Tages (JSON aus `hole_kerzen.py`), verdrahtet wie `generalprobe_echtlauf.py`. Drei Szenarien: `hand` (fortsetzen, anhalten, fortsetzen, glattstellen, freigeben — gehandelt wird nur im Zustand `aktiv`, beim Anhalten bleibt die Position stehen, nach dem Glattstellen ist sie null und der Runner gesperrt), `sturz` (alle Kerzen ab 11:00 Uhr New Yorker Zeit zehn Prozent tiefer — der Runner reißt die Tagesverlustgrenze selbst, stellt glatt, sperrt, meldet „Grenze verletzt", und sperrt nach einem „Freigeben" im selben Takt wieder, ohne neue Order) und `neustart` (das Dashboard startet mitten am Tag neu — Pause mit stehender Position, Pausenmeldung mit dem Auslöser `neustart`, und die Latenzanzeige zählt den Neustart nicht als Klick). Der Kopfkommentar nennt je Prüfung die Herkunft der Erwartung und die Grenzen: Gefüllt wird gespielt, der Klick geht nicht über HTTP, der Watchdog wird nicht gezielt ausgelöst. **Anlass:** Die Killswitch-Übung des Projekts stellt echte Orders und schreibt ins gemeinsame Register — sie darf nicht laufen, solange der Paper-Runner läuft. Gemessen am 2026-09-21 gegen den Betriebscode (0.20.0) mit den Kerzen des 2026-09-18 bei 1 s je Marktminute: je Szenario 431 s, Exit 0, 389 Takte, kein Befund und kein Hinweis; Reaktion des Runners auf einen Klick 0,85 bis 0,99 s (neun Messungen) | in WSL: `python killswitch_simulation.py <hand\|sturz\|neustart> <code-baum> <echter-projektordner> <kerzen.json> <wegwerf-ordner> [tempo]` |
| `killswitch_simulation_mutanten.py` | der Regressionsnachweis dazu: kopiert den Code-Baum, fährt zuerst die Grundlinie, macht die Kopie dann an sechs Stellen einzeln kaputt (Runner übergeht den Killswitch; Glattstellen schließt nicht; Tagesverlustgrenze greift nicht; jeder Auslöser zählt als Klick; keine Pausenmeldung beim Neustart; Tagesverlust hält nur an) und erwartet je Mutant Exit 2. Ein Muster, das nicht genau einmal trifft, ist ein Befund des Treibers. In zwei Teilen, je unter zehn Minuten. Gemessen am 2026-09-21 gegen den Betriebscode: Grundlinie und Gegenprobe in allen drei Szenarien ohne Befund, alle sechs Mutanten gesehen — drei davon an genau einer Prüfung (die Position nach dem Glattstellen, die Zahl der Klick-Messungen, die fehlende Pausenmeldung). **Der erste Lauf der Grundlinie fand dabei einen Fehler der Prüfung, nicht des Runners:** Im Schnelltempo fiel ein Klick mitten in einen Takt; dessen Taktzeile trug schon `gesperrt`, aber noch die Phase `handelt`, ohne Entscheidung und ohne Order. Seitdem ist alles ab dem zweiten Takt eines nicht aktiven Abschnitts verbindlich, der erste wird als Hinweis gedruckt | in WSL: `python killswitch_simulation_mutanten.py <code-baum> <echter-projektordner> <kerzen.json> <arbeitsordner> <1\|2>` |
Das Einrichten selbst (`deploy/dienste-einrichten.sh` unter `sudo -S`, `deploy/windows/wsl-wachhalten.ps1 -Aktion einrichten`) und das Eintragen des ntfy-Themas liefen aus dem Scratchpad der Sitzung; die Befehle stehen in der Administrator-Anleitung des Projekts, Abschnitt „Produktivbetrieb als Dienst".
Die Compose-Probe (Bauen aus dem echten Baum mit verankerten Ausschlüssen, Start, Healthcheck, Angriffsfälle, Runner-Probe im Container, Aufräumen) lag als `compose_probe_echt.sh` im Scratchpad der Sitzung und braucht `sudo -S` mit dem WSL-Passwort aus `$WSL_PW`; sie ist deshalb nicht hier abgelegt. Muster: `tar --exclude=./data` — ohne Anker fällt auch `trading/data` weg.
@@ -0,0 +1,118 @@
"""Gegenmessung zu Task 229: Protokollwachstum im gesperrten Zustand, beide Haelften des Schluesselwegs.
Aufruf (WSL, Venv): bremse_echtlauf.py <port> <ordner> <beginn|abschluss>
beginn -- fuenf falsche Passwoerter am Schluesselweg, dann 5 s Dauerfeuer auf beginn, dann 5 s auf
abschluss, dann 5 s auf den Code-Weg. Erwartet: genau EINE Zeile "gesperrt".
abschluss -- fuenf falsche Anmeldungen am Code-Weg, dann 5 s Dauerfeuer auf abschluss. Erwartet: genau
eine Zeile "gesperrt" mit weg=schluessel und leerem Benutzer.
Die Probe-Instanz (start_probe_w.py) laeuft aus einer Kopie oder einem Arbeitsbaum ohne Geheimnisdatei, mit
eigener Ablage und eigenem Port. Die Bremse lebt im Prozess: je Durchgang eine frische Instanz starten.
"""
import http.client
import json
import sys
import time
from pathlib import Path
PORT = int(sys.argv[1])
ORDNER = Path(sys.argv[2])
ART = sys.argv[3]
HOST = "localhost:" + str(PORT)
ORIGIN = "http://localhost:" + str(PORT)
PROTOKOLL = ORDNER / "anmeldungen.jsonl"
daten = json.loads((ORDNER / "zugang.json").read_text(encoding="utf-8"))
befunde = 0
def sage(name, ok, text):
global befunde
if not ok:
befunde += 1
print("[" + ("OK " if ok else "FAIL") + "] " + name + ": " + str(text), flush=True)
def anfrage(pfad, koerper, host=HOST, origin=ORIGIN):
verbindung = http.client.HTTPConnection("127.0.0.1", PORT, timeout=30)
rumpf = json.dumps(koerper).encode("utf-8")
verbindung.request(
"POST", pfad, body=rumpf,
headers={"Host": host, "Origin": origin, "Content-Type": "application/json"},
)
antwort = verbindung.getresponse()
inhalt = antwort.read()
verbindung.close()
return antwort.status, inhalt
def zeilen():
if not PROTOKOLL.exists():
return []
return [json.loads(z) for z in PROTOKOLL.read_text(encoding="utf-8").splitlines() if z.strip()]
def groesse():
return PROTOKOLL.stat().st_size if PROTOKOLL.exists() else 0
def feuer(name, pfad, koerper, sekunden=5.0):
vorher_b, vorher_z = groesse(), len(zeilen())
ende = time.monotonic() + sekunden
stati = {}
anzahl = 0
while time.monotonic() < ende:
status, _ = anfrage(pfad, koerper)
stati[status] = stati.get(status, 0) + 1
anzahl += 1
zuwachs_b, zuwachs_z = groesse() - vorher_b, len(zeilen()) - vorher_z
print(
f" {name}: {anzahl} Anfragen in {sekunden:.1f} s, Stati {stati}, Zuwachs {zuwachs_b} B "
f"({zuwachs_b / sekunden:.0f} B/s), {zuwachs_z} Zeile(n)",
flush=True,
)
return anzahl, stati, zuwachs_b, zuwachs_z
BEGINN = "/api/anmeldung/schluessel/beginn"
ABSCHLUSS = "/api/anmeldung/schluessel/abschluss"
CODE = "/api/anmeldung"
falsch_beginn = {"benutzer": daten["benutzer"], "passwort": "falsch"}
falsch_code = {"benutzer": daten["benutzer"], "passwort": "falsch", "code": "000000"}
erfunden = {"vorgang": "A" * 43, "antwort": {}}
if ART == "beginn":
stati = [anfrage(BEGINN, falsch_beginn)[0] for _ in range(5)]
sage("M1 fuenf falsche Passwoerter am Schluesselweg", stati == [401] * 5, stati)
n, st, zb, zz = feuer("beginn im gesperrten Zustand", BEGINN, falsch_beginn)
sage("M2 beginn: alle 429, hoechstens eine Zeile", set(st) == {429} and zz <= 1, f"{zz} Zeile(n), {zb} B")
n, st, zb, zz = feuer("abschluss im gesperrten Zustand", ABSCHLUSS, erfunden)
sage("M3 abschluss: alle 429, keine weitere Zeile", set(st) == {429} and zz == 0, f"{zz} Zeile(n), {zb} B")
n, st, zb, zz = feuer("Code-Weg im gesperrten Zustand", CODE, falsch_code)
sage("M4 Code-Weg: alle 429, keine weitere Zeile", set(st) == {429} and zz == 0, f"{zz} Zeile(n), {zb} B")
alle = zeilen()
gesperrt = [z for z in alle if z["ergebnis"] == "gesperrt"]
sage(
"M5 genau eine Zeile gesperrt, vom Schluesselweg",
len(gesperrt) == 1 and gesperrt[0].get("weg") == "schluessel" and gesperrt[0]["benutzer"] == daten["benutzer"],
f"{len(gesperrt)} von {len(alle)} Zeilen: {gesperrt[:2]}",
)
status, inhalt = anfrage(ABSCHLUSS, erfunden, host="127.0.0.1:" + str(PORT), origin="http://127.0.0.1:" + str(PORT))
sage("M6 abschluss unter 127.0.0.1 ist 409 mit der richtigen Adresse", status == 409 and ORIGIN.encode() in inhalt, f"{status} {inhalt[:120]!r}")
else:
stati = [anfrage(CODE, falsch_code)[0] for _ in range(5)]
sage("N1 fuenf falsche Anmeldungen am Code-Weg", stati == [401] * 5, stati)
n, st, zb, zz = feuer("abschluss als erste gesperrte Anfrage", ABSCHLUSS, erfunden)
sage("N2 abschluss: alle 429, genau eine Zeile", set(st) == {429} and zz == 1, f"{zz} Zeile(n), {zb} B")
gesperrt = [z for z in zeilen() if z["ergebnis"] == "gesperrt"]
sage(
"N3 die eine Zeile traegt weg=schluessel und keinen Benutzer",
len(gesperrt) == 1 and gesperrt[0].get("weg") == "schluessel" and gesperrt[0]["benutzer"] == "",
gesperrt[:2],
)
n, st, zb, zz = feuer("beginn danach", BEGINN, falsch_beginn, sekunden=2.0)
sage("N4 beginn danach: keine weitere Zeile", set(st) == {429} and zz == 0, f"{zz} Zeile(n)")
text = PROTOKOLL.read_text(encoding="utf-8") if PROTOKOLL.exists() else ""
sage("Z1 kein Passwort im Protokoll", daten["passwort"] not in text and "falsch" not in text.replace('"ergebnis": "falsch"', ""), "geprueft")
print(f"{befunde} Befund(e).")
sys.exit(1 if befunde else 0)
@@ -0,0 +1,192 @@
"""Browserlauf zu Paket W: echter Chromium mit virtuellem Authenticator gegen die Probe-Instanz.
Koordinator-Pruefcode. Laeuft unter Windows mit Playwright. Ablauf: mit Passwort und Code anmelden, in
"Zugang" einen Schluessel registrieren, abmelden, mit dem Schluessel anmelden (der Zaehler des virtuellen
Authenticators muss steigen -- sonst waere er nie gefragt worden), abmelden, wieder mit Code anmelden,
Schluessel entfernen. Dazu die Gegenprobe unter 127.0.0.1 und ein Blick in Schluesseldatei und
Anmeldeprotokoll der Probe-Instanz.
Aufruf: python browser_w.py <port> <ausgabeordner-der-probe>
"""
import base64
import hashlib
import hmac
import json
import struct
import sys
import time
from pathlib import Path
from playwright.sync_api import sync_playwright
PORT = int(sys.argv[1])
ORDNER = Path(sys.argv[2])
ZUGANG = json.loads((ORDNER / "zugang.json").read_text(encoding="utf-8"))
KNOPF_SCHLUESSEL = "Mit Sicherheitsschlüssel anmelden"
befunde: list[str] = []
schritte: list[str] = []
def totp(geheimnis: str) -> tuple[str, int]:
schluessel = base64.b32decode(geheimnis)
zaehler = int(time.time() // 30)
mac = hmac.new(schluessel, struct.pack(">Q", zaehler), hashlib.sha1).digest()
versatz = mac[-1] & 0x0F
wert = (struct.unpack(">I", mac[versatz:versatz + 4])[0] & 0x7FFFFFFF) % 1_000_000
return f"{wert:06d}", zaehler
def schritt(text: str, ok: bool) -> None:
schritte.append(("OK " if ok else "BEFUND ") + text)
if not ok:
befunde.append(text)
def melde_mit_code_an(seite, letzter_zaehler: int) -> int:
# Ein Code gilt je Zeitschritt einmal; vor der zweiten Anmeldung den naechsten Schritt abwarten.
while int(time.time() // 30) <= letzter_zaehler:
time.sleep(1.0)
code, zaehler = totp(ZUGANG["geheimnis"])
seite.fill("#anmeldung-benutzer", ZUGANG["benutzer"])
seite.fill("#anmeldung-passwort", ZUGANG["passwort"])
seite.fill("#anmeldung-code", code)
seite.get_by_role("button", name="Anmelden", exact=True).click()
seite.get_by_role("button", name="Abmelden").wait_for(timeout=15000)
return zaehler
def main() -> int:
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
kontext = browser.new_context()
seite = kontext.new_page()
konsole: list[str] = []
api: list[str] = []
seite.on("console", lambda m: konsole.append(f"{m.type}: {m.text}") if m.type in ("error", "warning") else None)
seite.on("pageerror", lambda e: konsole.append(f"pageerror: {e}"))
seite.on("response", lambda r: api.append(f"{r.status} {r.request.method} {r.url.split(str(PORT), 1)[-1]}") if "/api/" in r.url else None)
cdp = kontext.new_cdp_session(seite)
cdp.send("WebAuthn.enable")
authenticator = cdp.send("WebAuthn.addVirtualAuthenticator", {"options": {
"protocol": "ctap2", "transport": "usb", "hasResidentKey": False,
"hasUserVerification": True, "isUserVerified": True, "automaticPresenceSimulation": True,
}})["authenticatorId"]
def zugaenge() -> list[dict]:
return cdp.send("WebAuthn.getCredentials", {"authenticatorId": authenticator})["credentials"]
seite.goto(f"http://localhost:{PORT}/")
seite.locator("#anmeldung-benutzer").wait_for(timeout=15000)
schritt("Anmeldemaske unter http://localhost erreichbar", True)
knopf = seite.get_by_role("button", name=KNOPF_SCHLUESSEL)
schritt("Knopf fuer den Schluessel ist da und bedienbar", knopf.count() == 1 and knopf.is_enabled())
zaehler = melde_mit_code_an(seite, -1)
schritt("Anmeldung mit Passwort und Code", True)
seite.get_by_text("Zugang", exact=True).first.click()
seite.locator("#zugang-bezeichnung").wait_for(timeout=10000)
seite.fill("#zugang-bezeichnung", "Virtueller Schluessel")
seite.fill("#zugang-passwort", ZUGANG["passwort"])
seite.get_by_role("button", name="Schlüssel hinzufügen").click()
liste = seite.get_by_role("list", name="Sicherheitsschlüssel")
liste.wait_for(timeout=15000)
schritt("Registrierung: der Schluessel steht in der Liste", "Virtueller Schluessel" in liste.inner_text())
vorher = zugaenge()
schritt("Der virtuelle Authenticator haelt genau einen Zugang", len(vorher) == 1)
zaehler_vorher = vorher[0]["signCount"] if vorher else -1
# Schluesseldatei ansehen, solange der Schluessel hinterlegt ist.
datei = ORDNER / "zugang-schluessel.json"
kandidaten = [pfad for pfad in ORDNER.glob("*schluessel*")]
schritt(f"Schluesseldatei liegt neben der Zugangsdatei ({[k.name for k in kandidaten]})", bool(kandidaten))
if kandidaten:
inhalt = kandidaten[0].read_text(encoding="utf-8")
schritt("Schluesseldatei traegt weder Passwort noch TOTP-Geheimnis",
ZUGANG["passwort"] not in inhalt and ZUGANG["geheimnis"] not in inhalt)
daten = json.loads(inhalt)
print("Schluesseldatei, Form:", json.dumps(
{k: (sorted(v[0]) if isinstance(v, list) and v and isinstance(v[0], dict) else type(v).__name__)
for k, v in (daten.items() if isinstance(daten, dict) else {"liste": daten}.items())},
ensure_ascii=False)[:600])
seite.get_by_role("button", name="Abmelden").click()
seite.locator("#anmeldung-benutzer").wait_for(timeout=10000)
seite.fill("#anmeldung-benutzer", ZUGANG["benutzer"])
seite.fill("#anmeldung-passwort", ZUGANG["passwort"])
seite.get_by_role("button", name=KNOPF_SCHLUESSEL).click()
seite.get_by_role("button", name="Abmelden").wait_for(timeout=15000)
schritt("Anmeldung mit Passwort und Sicherheitsschluessel, ohne Code", True)
nachher = zugaenge()
zaehler_nachher = nachher[0]["signCount"] if nachher else -1
schritt(f"Der Authenticator wurde wirklich gefragt (Zaehler {zaehler_vorher} -> {zaehler_nachher})",
zaehler_nachher > zaehler_vorher)
# Falsches Passwort auf dem Schluessel-Weg: Ablehnung, keine Sitzung.
seite.get_by_role("button", name="Abmelden").click()
seite.locator("#anmeldung-benutzer").wait_for(timeout=10000)
seite.fill("#anmeldung-benutzer", ZUGANG["benutzer"])
seite.fill("#anmeldung-passwort", "ganz-sicher-falsch-2026")
seite.get_by_role("button", name=KNOPF_SCHLUESSEL).click()
hinweis = seite.get_by_role("alert")
hinweis.wait_for(timeout=10000)
schritt("Falsches Passwort auf dem Schluessel-Weg wird abgelehnt", "stimmt nicht" in hinweis.inner_text())
zaehler = melde_mit_code_an(seite, zaehler)
schritt("Der Code-Weg geht danach unveraendert", True)
seite.get_by_text("Zugang", exact=True).first.click()
seite.locator("#zugang-passwort").wait_for(timeout=10000)
seite.fill("#zugang-passwort", ZUGANG["passwort"])
seite.get_by_role("button", name="Entfernen").click()
seite.get_by_text("Es ist kein Sicherheitsschlüssel hinterlegt").wait_for(timeout=10000)
schritt("Entfernen: die Liste ist wieder leer", True)
# Gegenprobe unter 127.0.0.1: beginn antwortet mit 409, kein 500, kein leerer Schirm.
zweite = kontext.new_page()
zweite.goto(f"http://127.0.0.1:{PORT}/")
zweite.locator("#anmeldung-benutzer").wait_for(timeout=15000)
antwort = zweite.evaluate(
"""async ([benutzer, passwort]) => {
const r = await fetch('/api/anmeldung/schluessel/beginn', {method: 'POST',
headers: {'content-type': 'application/json'},
body: JSON.stringify({benutzer, passwort})});
return [r.status, await r.text()];
}""", [ZUGANG["benutzer"], ZUGANG["passwort"]])
schritt(f"Unter 127.0.0.1 antwortet beginn mit 409 ({antwort[0]})", antwort[0] == 409)
print("Satz unter 127.0.0.1:", antwort[1][:260])
browser.close()
protokoll = ORDNER / "anmeldungen.jsonl"
zeilen = [json.loads(z) for z in protokoll.read_text(encoding="utf-8").splitlines() if z.strip()]
wege = [z.get("weg") for z in zeilen]
print(f"Anmeldeprotokoll: {len(zeilen)} Zeilen, Ausgaenge {[z.get('ausgang') for z in zeilen]}, Wege {wege}")
roh = protokoll.read_text(encoding="utf-8")
schritt("Anmeldeprotokoll traegt eine Zeile mit weg=schluessel", "schluessel" in wege)
schritt("Anmeldeprotokoll traegt weder Passwort noch Geheimnis",
ZUGANG["passwort"] not in roh and ZUGANG["geheimnis"] not in roh and "ganz-sicher-falsch" not in roh)
boese = [z for z in konsole if "Content Security Policy" in z or "Permissions" in z or "pageerror" in z]
schritt(f"Keine CSP-, Permissions-Policy- oder Seitenfehler in der Konsole ({len(boese)})", not boese)
fuenfhunderter = [z for z in api if z.startswith("5")]
schritt(f"Keine 5xx-Antwort der Schnittstelle ({len(fuenfhunderter)})", not fuenfhunderter)
print("--- Schritte")
for zeile in schritte:
print(zeile)
print("--- API-Antworten der Schluessel-Wege")
for zeile in api:
if "schluessel" in zeile or "anmeldung" in zeile:
print(" " + zeile)
print("--- Konsole (Fehler und Warnungen)")
for zeile in konsole[:12]:
print(" " + zeile[:220])
print(f"{len(befunde)} Befund(e): {befunde}")
return 2 if befunde else 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,343 @@
"""Eigener Nachtlauf der Generalprobe: _generalprobe aus scripts/paper.py mit echten Kerzen, ohne Netz.
Koordinator-Pruefcode, nicht Teil des Repos. Gerufen wird die echte Funktion _generalprobe -- dieselbe
Verdrahtung wie um 9:00 Uhr --, umgebogen sind nur die Stellen, die sonst ins Netz oder in den Betrieb
gingen:
- REGISTER zeigt auf eine Sicherungskopie des echten Registers in einem Wegwerf-Ordner (das echte wird
nur lesend geoeffnet), ARTEFAKTE auf den echten artifacts-Ordner (nur lesen: das Registermodell).
- Statusdatei und Protokollordner liegen im Wegwerf-Ordner.
- AlpacaKerzenquelle.from_secrets liefert die echten IEX-Kerzen des Tages aus einer JSON-Datei.
- AlpacaBroker.from_secrets liefert ein mitschreibendes Konto, das glatt ist und bei jedem schreibenden
Aufruf wirft -- damit steht im Bericht, was die Generalprobe am "echten" Broker ruft.
- requests.Session.request schreibt mit und wirft: Jeder Netzversuch des Laufs faellt auf.
- Der Melder ist ein echter Melder mit mitschreibendem Transport, in der Huelle GeneralprobeMelder.
"Pascal" klickt aus einem zweiten Faden -- wie das Dashboard mit einem frischen Killswitch-Objekt aus dem
Register, am mitschreibenden Konto: fortsetzen, spaeter anhalten und fortsetzen, glattstellen, freigeben.
Aufruf (WSL): python generalprobe_echtlauf.py <code-baum> <echter-projektordner> <kerzen.json> <wegwerf-ordner> [tempo]
"""
import importlib.util
import json
import os
import sqlite3
import sys
import threading
import time
from dataclasses import replace
from datetime import UTC
from datetime import datetime
from datetime import timedelta
from pathlib import Path
CODE = Path(sys.argv[1]).resolve()
ECHT = Path(sys.argv[2]).resolve()
KERZEN = Path(sys.argv[3]).resolve()
WEG = Path(sys.argv[4]).resolve()
TEMPO = float(sys.argv[5]) if len(sys.argv) > 5 else 1.0
os.chdir(CODE)
sys.path.insert(0, str(CODE))
import requests # noqa: E402
from trading.broker.schnittstelle import Marktuhr # noqa: E402
from trading.killswitch.killswitch import Killswitch # noqa: E402
from trading.killswitch.killswitch import lade_killswitch_config # noqa: E402
from trading.paper.config import lade_paper_config # noqa: E402
from trading.paper.kerzen import HANDELSZONE # noqa: E402
from trading.paper.kerzen import Kerze # noqa: E402
from trading.paper.melder import Melder # noqa: E402
from trading.paper.melder import MelderConfig # noqa: E402
from trading.paper.status import lies_status # noqa: E402
from trading.registry.registry import Registry # noqa: E402
NETZVERSUCHE: list[str] = []
def _kein_netz(self, method, url, *args, **kwargs):
NETZVERSUCHE.append(f"{method} {url.split('?')[0]}")
raise requests.ConnectionError("Nachtlauf: kein Netz")
requests.sessions.Session.request = _kein_netz
class Antwort:
status_code = 200
class Transport:
def __init__(self):
self.aufrufe = []
def __call__(self, url, *, data=None, headers=None, timeout=None, **rest):
rumpf = data.decode("utf-8") if isinstance(data, bytes) else str(data)
self.aufrufe.append(
{
"titel": headers.get("Title"),
"prioritaet": headers.get("Priority"),
"erste_zeile": rumpf.split("\n")[0],
"echt": datetime.now(UTC).isoformat(timespec="seconds"),
}
)
return Antwort()
class MitschreibendesKonto:
"""Steht fuer das echte Alpaca-Paper-Konto: glatt, keine Order, Boerse zu. Schreiben wirft."""
def __init__(self, name):
self.name = name
self.aufrufe = []
def positionen(self):
self.aufrufe.append("positionen")
return []
def offene_orders(self):
self.aufrufe.append("offene_orders")
return []
def marktuhr(self):
self.aufrufe.append("marktuhr")
jetzt = datetime.now(UTC)
return Marktuhr(is_open=False, next_open=jetzt + timedelta(hours=8),
next_close=jetzt + timedelta(hours=14))
def storniere_alle(self):
self.aufrufe.append("storniere_alle")
return []
def schliesse_alle(self):
self.aufrufe.append("schliesse_alle")
return []
def __getattr__(self, name):
def wirft(*a, **k):
self.aufrufe.append(f"VERBOTEN:{name}")
raise AssertionError(f"{self.name}: unerwarteter Aufruf {name}")
return wirft
def lade_kerzen(pfad):
roh = json.loads(Path(pfad).read_text(encoding="utf-8"))
return [
Kerze(zeit=datetime.fromisoformat(e["zeit"]), open=e["open"], high=e["high"],
low=e["low"], close=e["close"], volume=e["volume"])
for e in roh
]
def main():
kerzen = lade_kerzen(KERZEN)
tag = kerzen[0].zeit.astimezone(HANDELSZONE).date().isoformat()
WEG.mkdir(parents=True, exist_ok=False)
# Sicherungskopie des echten Registers; das echte wird nur lesend geoeffnet.
quelle = sqlite3.connect(f"file:{ECHT / 'data' / 'registry.db'}?mode=ro", uri=True)
ziel = sqlite3.connect(WEG / "registry.db")
quelle.backup(ziel)
quelle.close()
# Die Spalte artefakt_pfad ist relativ zum Projektordner. In der Kopie wird sie absolut, damit der
# Lauf aus dem Code-Baum heraus das Registermodell im echten artifacts-Ordner findet (nur lesen).
for zeile in ziel.execute("SELECT agent, version, artefakt_pfad FROM versionen").fetchall():
pfad = Path(zeile[2].replace("\\", "/"))
if not pfad.is_absolute():
ziel.execute("UPDATE versionen SET artefakt_pfad = ? WHERE agent = ? AND version = ?",
(str(ECHT / pfad), zeile[0], zeile[1]))
ziel.commit()
ziel.close()
spec = importlib.util.spec_from_file_location("paper_skript_nachtlauf", CODE / "scripts" / "paper.py")
paper = importlib.util.module_from_spec(spec)
spec.loader.exec_module(paper)
paper.REGISTER = str(WEG / "registry.db")
paper.ARTEFAKTE = str(ECHT / "artifacts")
konto_der_probe = MitschreibendesKonto("Konto der Generalprobe")
konto_des_dashboards = MitschreibendesKonto("Konto des Dashboards")
class Kerzenquelle:
@classmethod
def from_secrets(cls, *a, **k):
return cls()
def kerzen_des_tages(self, instrument_id, ende):
return list(kerzen)
class BrokerFabrik:
@classmethod
def from_secrets(cls, *a, **k):
return konto_der_probe
paper.AlpacaKerzenquelle = Kerzenquelle
paper.AlpacaBroker = BrokerFabrik
config = replace(
lade_paper_config("config/paper.toml"),
status_datei=str(WEG / "paper" / "status.json"),
protokoll_ordner=str(WEG / "protokolle"),
)
ks_config = lade_killswitch_config("config/killswitch.toml")
transport = Transport()
melder = Melder(
config=MelderConfig(aktiv=True, ntfy_url="https://ntfy.sh", thema="nachtlauf-geht-nie-ins-netz",
zeitlimit_s=3.0, stille_nach_s=180.0),
token="", transport=transport,
)
beobachtet = {"max_alter_s": 0.0, "modi": set(), "marktzeiten": [], "phasen": [], "klicks": [],
"rueckwaerts": 0, "position_nach_glatt": None, "fehler": []}
halt = threading.Event()
def dashboard_killswitch():
registry = Registry(db_pfad=WEG / "registry.db", artefakt_wurzel=ECHT / "artifacts")
gespeichert = registry.killswitch_zustand()
return Killswitch(broker=konto_des_dashboards, registry=registry, config=ks_config,
zustand=gespeichert.zustand if gespeichert is not None else "angehalten")
def klick(was, marktzeit, position):
ks = dashboard_killswitch()
vorher = ks.zustand
if was == "glattstellen":
ergebnis = ks.glattstellen("pascal", "Nachtlauf", ks_config.bestaetigungswort)
else:
ergebnis = getattr(ks, was)("pascal", "Nachtlauf")
beobachtet["klicks"].append(
f"{marktzeit.astimezone(HANDELSZONE):%H:%M} {was}: {vorher} -> {ergebnis.zustand}, "
f"erfolgreich={ergebnis.erfolgreich}, gespielte Position vorher {position}"
)
def pascal():
plan = [("fortsetzen", (9, 31)), ("anhalten", (10, 30)), ("fortsetzen", (10, 36)),
("glattstellen", (12, 0)), ("freigeben", (12, 10))]
letzte = None
glatt_um = None
while not halt.is_set():
status = lies_status(config.status_datei)
if isinstance(status, dict):
try:
zeit = datetime.fromisoformat(status["zeit"])
markt = datetime.fromisoformat(status["marktzeit"])
except (KeyError, ValueError, TypeError) as ausnahme:
beobachtet["fehler"].append(repr(ausnahme))
time.sleep(0.2)
continue
if status.get("phase") != "beendet":
alter = (datetime.now(UTC) - zeit).total_seconds()
beobachtet["max_alter_s"] = max(beobachtet["max_alter_s"], alter)
beobachtet["modi"].add(status.get("modus"))
if not beobachtet["phasen"] or beobachtet["phasen"][-1] != status.get("phase"):
beobachtet["phasen"].append(status.get("phase"))
if letzte is not None and markt < letzte:
beobachtet["rueckwaerts"] += 1
if letzte is None or markt != letzte:
beobachtet["marktzeiten"].append(markt)
letzte = markt
ort = markt.astimezone(HANDELSZONE)
if plan and (ort.hour, ort.minute) >= plan[0][1]:
was, _ = plan.pop(0)
try:
klick(was, markt, status.get("position"))
except Exception as ausnahme:
beobachtet["klicks"].append(f"{ort:%H:%M} {was}: AUSNAHME {ausnahme!r}")
if was == "glattstellen":
glatt_um = markt
if glatt_um is not None and beobachtet["position_nach_glatt"] is None \
and markt >= glatt_um + timedelta(minutes=2):
beobachtet["position_nach_glatt"] = status.get("position")
time.sleep(0.2)
faden = threading.Thread(target=pascal, name="pascal", daemon=True)
faden.start()
beginn = time.monotonic()
code = paper._generalprobe(config, tag, TEMPO, melder)
dauer = time.monotonic() - beginn
time.sleep(0.5)
halt.set()
faden.join(timeout=5)
print("=" * 70)
print(f"Tag {tag}, {len(kerzen)} Kerzen, Tempo {TEMPO:g} s je Marktminute, echte Dauer {dauer:.0f} s, Exit {code}")
schluss = lies_status(config.status_datei)
print(f"Statusdatei am Ende: modus={schluss.get('modus')}, phase={schluss.get('phase')}, "
f"position={schluss.get('position')}, schluessel={sorted(schluss)}")
print(f"Modi gesehen: {sorted(str(m) for m in beobachtet['modi'])}; Phasen: {beobachtet['phasen']}")
zeiten = beobachtet["marktzeiten"]
print(f"Marktzeit: {len(zeiten)} verschiedene Werte, von {zeiten[0].astimezone(HANDELSZONE):%Y-%m-%d %H:%M} "
f"bis {zeiten[-1].astimezone(HANDELSZONE):%Y-%m-%d %H:%M}, Ruecksprung {beobachtet['rueckwaerts']}")
print(f"Groesstes Alter des Herzschlags gegen die echte Uhr: {beobachtet['max_alter_s']:.1f} s")
print("Klicks:")
for zeile in beobachtet["klicks"]:
print(" " + zeile)
print(f"Gespielte Position zwei Marktminuten nach dem Glattstellen: {beobachtet['position_nach_glatt']}")
print(f"Lesefehler im Beobachter: {beobachtet['fehler'][:3]}")
protokolle = sorted((WEG / "protokolle").glob("*.jsonl"))
zeilen = []
for pfad in protokolle:
with pfad.open(encoding="utf-8") as datei:
zeilen += [json.loads(zeile) for zeile in datei if zeile.strip()]
takte = [z for z in zeilen if z.get("art") == "takt"]
arten = {}
for z in zeilen:
schluessel = z.get("art") or z.get("was") or "?"
arten[schluessel] = arten.get(schluessel, 0) + 1
print(f"Protokoll: {[p.name for p in protokolle]}, {len(zeilen)} Zeilen, Arten {arten}")
orders = [z for z in zeilen if z.get("art") == "order" or z.get("was") == "order"]
print(f"Takte {len(takte)}, Orderzeilen {len(orders)}")
if takte:
print(f"Erster Takt: {json.dumps(takte[0], ensure_ascii=False)[:400]}")
print(f"Letzter Takt: {json.dumps(takte[-1], ensure_ascii=False)[:400]}")
print(f"Meldungen ({len(transport.aufrufe)}):")
for m in transport.aufrufe:
print(f" {m['echt']} [{m['prioritaet']}] {m['titel']} | {m['erste_zeile']}")
print(f"Aufrufe am Konto der Generalprobe: {konto_der_probe.aufrufe}")
print(f"Aufrufe am Konto des Dashboards: {konto_des_dashboards.aufrufe}")
print(f"Netzversuche ueber requests: {NETZVERSUCHE}")
tagesstart = list(WEG.rglob("tagesstart.json"))
print(f"tagesstart.json im Wegwerf-Ordner: {tagesstart}")
registry = Registry(db_pfad=WEG / "registry.db", artefakt_wurzel=ECHT / "artifacts")
ereignisse = registry.killswitch_ereignisse()
frisch = [e for e in ereignisse if getattr(e, "grund", "") and "Nachtlauf" in e.grund
or getattr(e, "ausloeser", "") in ("neustart", "paper-runner")]
print(f"Killswitch-Ereignisse im Wegwerf-Register: {len(ereignisse)} gesamt; die letzten acht:")
for e in ereignisse[-8:] if len(ereignisse) >= 8 else ereignisse:
print(f" {getattr(e, 'zeit', '?')} {e.stufe} / {e.ausloeser} / {e.grund[:60]} / erfolgreich={e.erfolgreich}")
print(f"Killswitch am Ende: {registry.killswitch_zustand().zustand}")
befunde = []
if code != 0:
befunde.append(f"Exit {code} statt 0")
if beobachtet["modi"] != {"generalprobe"}:
befunde.append(f"Modi {beobachtet['modi']}")
if any(not str(m["titel"]).startswith("Generalprobe: ") for m in transport.aufrufe):
befunde.append("eine Meldung ohne Vorsatz")
if not transport.aufrufe:
befunde.append("keine Meldung")
if NETZVERSUCHE:
befunde.append("Netzversuch")
if sorted(konto_der_probe.aufrufe) != ["marktuhr", "offene_orders", "positionen"]:
befunde.append("Konto der Generalprobe: andere Aufrufe als die drei lesenden")
if any(a.startswith("VERBOTEN") for a in konto_des_dashboards.aufrufe):
befunde.append("Konto des Dashboards: verbotener Aufruf")
if beobachtet["position_nach_glatt"] != 0:
befunde.append(f"Position nach dem Glattstellen {beobachtet['position_nach_glatt']}")
if beobachtet["max_alter_s"] > 5.0:
befunde.append(f"Herzschlag bis {beobachtet['max_alter_s']:.1f} s alt")
if beobachtet["rueckwaerts"]:
befunde.append("Marktzeit lief rueckwaerts")
if tagesstart:
befunde.append("tagesstart.json entstanden")
if schluss.get("position") != 0:
befunde.append(f"Position am Ende {schluss.get('position')}")
print(f"{len(befunde)} Befund(e): {befunde}")
return 2 if befunde else 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,44 @@
"""Setzt [homeassistant] aktiv in config/secrets.toml auf true oder false. Gibt nie einen Wert der Datei aus.
Aufruf: python ha_aktiv_setzen.py <secrets.toml> <true|false>
Geaendert wird genau eine Zeile im Abschnitt [homeassistant]; alles andere bleibt Byte fuer Byte stehen.
"""
import sys
import tomllib
from pathlib import Path
pfad = Path(sys.argv[1])
ziel = sys.argv[2]
assert ziel in ("true", "false")
roh = pfad.read_bytes()
text = roh.decode("utf-8")
zeilenende = "\r\n" if "\r\n" in text else "\n"
zeilen = text.split(zeilenende)
im_abschnitt = False
treffer = []
for nummer, zeile in enumerate(zeilen):
kern = zeile.strip()
if kern.startswith("["):
im_abschnitt = kern == "[homeassistant]"
continue
if im_abschnitt and kern.split("=")[0].strip() == "aktiv":
treffer.append(nummer)
if len(treffer) != 1:
print(f"ABBRUCH: {len(treffer)} Zeilen 'aktiv' im Abschnitt [homeassistant] gefunden, erwartet genau eine")
sys.exit(2)
vorher = tomllib.loads(text)["homeassistant"]["aktiv"]
zeilen[treffer[0]] = f"aktiv = {ziel}"
neu = zeilenende.join(zeilen)
nachher = tomllib.loads(neu)
# Die Probe vor dem Schreiben: Die Datei bleibt gueltiges TOML, und nur dieser eine Wert aendert sich.
alt = tomllib.loads(text)
alt["homeassistant"]["aktiv"] = nachher["homeassistant"]["aktiv"]
assert alt == nachher, "mehr als ein Wert haette sich geaendert"
pfad.write_bytes(neu.encode("utf-8"))
print(f"[homeassistant] aktiv: {vorher} -> {nachher['homeassistant']['aktiv']}; Abschnitte {sorted(nachher)}; "
f"Laenge der Adresse {len(nachher['homeassistant'].get('webhook_url', ''))}")
@@ -2,7 +2,9 @@
Koordinator-Werkzeug zu Teil 10, nicht fuers Repo. Es misst und berichtet; ein Urteil faellt es nur an
drei Stellen, und dort gegen Werte aus dem Protokollkopf (Projektvorgabe, config/paper.toml):
- Fuellzeit einer Order (filled_at - submitted_at laut Alpaca) gegen order_timeout_s,
- Annahme bis Fuellung einer Order (filled_at - created_at laut Alpaca) gegen order_timeout_s -- nicht
filled_at - submitted_at: Am 2026-09-21 lagen zwischen Annahme und Weitergabe 8,7 s, und der Runner wartet
ab der Annahme. Beide Spannen werden ausgegeben; mehr als die Haelfte der Grenze ist ein Hinweis,
- Schlussposition ungleich null,
- eine Position beim Broker mit negativer Stueckzahl (Short): wird nur gemeldet, nicht bewertet.
@@ -175,6 +177,7 @@ def teil_broker(kopf: dict, tag_start: datetime, anmeldung: dict[str, str]) -> l
grenze = float(kopf.get("order_timeout_s") or 0)
print(f"orders beim broker am {tag_start.date()}: {len(orders)} (order_timeout_s laut protokoll: {grenze})")
for o in orders:
erstellt = zeitpunkt(o["created_at"]) if o.get("created_at") else None
eingereicht = zeitpunkt(o["submitted_at"]) if o.get("submitted_at") else None
gefuellt = zeitpunkt(o["filled_at"]) if o.get("filled_at") else None
dauer = (gefuellt - eingereicht).total_seconds() if eingereicht and gefuellt else None
@@ -183,8 +186,21 @@ def teil_broker(kopf: dict, tag_start: datetime, anmeldung: dict[str, str]) -> l
f" {o.get('side')} {o.get('qty')} {o.get('symbol')} typ {o.get('type')} status {o.get('status')} "
f"eingereicht {o.get('submitted_at')} fuellzeit {text} kurs {o.get('filled_avg_price')} id {str(o.get('client_order_id'))[:24]}"
)
if dauer is not None and grenze and dauer > grenze:
print(f" BEFUND: Fuellzeit {dauer:.1f} s ueber order_timeout_s {grenze}")
# Gemessen am 2026-09-21: Alpaca nahm eine Market-Order an (created_at) und gab sie erst 8,7 s
# spaeter weiter (submitted_at). filled_at - submitted_at allein (3,8 s) verschwieg das; der Runner
# wartet aber ab der Annahme. Gegen order_timeout_s zaehlt deshalb die ganze Spanne ab created_at.
ganz = (gefuellt - erstellt).total_seconds() if erstellt and gefuellt else None
if erstellt and eingereicht:
ganz_text = f"{ganz:.3f} s" if ganz is not None else "nicht gefuellt"
print(
f" angenommen {o.get('created_at')}: bis eingereicht {(eingereicht - erstellt).total_seconds():+.3f} s, "
f"angenommen bis gefuellt {ganz_text}"
)
massgeblich = ganz if ganz is not None else dauer
if massgeblich is not None and grenze and massgeblich > grenze:
print(f" BEFUND: Annahme bis Fuellung {massgeblich:.1f} s ueber order_timeout_s {grenze}")
elif massgeblich is not None and grenze and massgeblich > grenze / 2:
print(f" HINWEIS: Annahme bis Fuellung {massgeblich:.1f} s ist mehr als die Haelfte von order_timeout_s {grenze}")
positionen = requests.get(f"{HANDEL_URL}/v2/positions", headers=anmeldung, timeout=30)
if positionen.status_code == 200:
liste = positionen.json()
@@ -0,0 +1,224 @@
"""Eigener Lauf des Hausmelders gegen einen mitschreibenden Server auf 127.0.0.1 -- und wahlweise ein POST ans echte HA.
Koordinator-Pruefcode, nicht Teil des Repos. Teil 1 (immer): echter Hausmelder-Faden, echter requests-POST,
lokaler HTTP-Server, gestellte Statusdatei, Wegwerf-Register. Teil 2 (nur mit --echt <secrets.toml>): ein
einziger pruefe()-Durchlauf mit unterscheidbaren Werten an die Adresse aus der Geheimnisdatei; die Adresse
wird nie ausgegeben. Rueckgelesen wird danach unter Windows mit ha_einrichten.py.
Aufruf (WSL): python hausmelder_echtlauf.py <code-baum> <wegwerf-ordner> [--echt <secrets.toml>]
"""
import json
import os
import sys
import threading
import time
from datetime import UTC
from datetime import datetime
from http.server import BaseHTTPRequestHandler
from http.server import ThreadingHTTPServer
from pathlib import Path
CODE = Path(sys.argv[1]).resolve()
WEG = Path(sys.argv[2]).resolve()
ECHT_SECRETS = Path(sys.argv[4]).resolve() if len(sys.argv) > 4 and sys.argv[3] == "--echt" else None
os.chdir(CODE)
sys.path.insert(0, str(CODE))
from trading.dashboard import hausmelder as modul # noqa: E402
from trading.dashboard.hausmelder import Hausmelder # noqa: E402
from trading.dashboard.hausmelder import killswitch_leser # noqa: E402
from trading.dashboard.hausmelder import lies_hausmelder_config # noqa: E402
from trading.dashboard.hausmelder import starte_hausmelder # noqa: E402
from trading.paper.status import schreibe_status # noqa: E402
from trading.registry.registry import Registry # noqa: E402
EMPFANGEN = []
ANTWORT = {"code": 200, "location": None}
class Mitschreiber(BaseHTTPRequestHandler):
def do_POST(self):
laenge = int(self.headers.get("Content-Length", "0"))
rumpf = self.rfile.read(laenge)
EMPFANGEN.append({"pfad": self.path, "typ": self.headers.get("Content-Type"),
"daten": json.loads(rumpf.decode("utf-8")), "antwort": ANTWORT["code"]})
self.send_response(ANTWORT["code"])
if ANTWORT["location"]:
self.send_header("Location", ANTWORT["location"])
self.send_header("Content-Length", "0")
self.end_headers()
def do_GET(self):
EMPFANGEN.append({"pfad": self.path, "typ": "GET", "daten": None, "antwort": 200})
self.send_response(200)
self.send_header("Content-Length", "0")
self.end_headers()
def log_message(self, *a):
pass
def status(pfad, **felder):
grund = {"zeit": datetime.now(UTC).isoformat(), "modus": "paper", "phase": "handelt", "position": 0,
"kontowert": 100000.0, "startkapital": 100000.0}
grund.update(felder)
schreibe_status(pfad, grund)
def warte_bis(bedingung, sekunden=5.0):
ende = time.monotonic() + sekunden
while time.monotonic() < ende:
if bedingung():
return True
time.sleep(0.05)
return False
def main():
WEG.mkdir(parents=True, exist_ok=False)
befunde = []
server = ThreadingHTTPServer(("127.0.0.1", 0), Mitschreiber)
port = server.server_address[1]
threading.Thread(target=server.serve_forever, daemon=True).start()
kennung = "nachtlauf-" + "a" * 40
url = f"http://127.0.0.1:{port}/api/webhook/{kennung}"
registry = Registry(db_pfad=WEG / "registry.db", artefakt_wurzel=WEG / "artifacts")
registry.setze_killswitch_zustand("angehalten", ausloeser="neustart")
statusdatei = WEG / "paper" / "status.json"
statusdatei.parent.mkdir(parents=True)
# Teil 1a: der Faden selbst, mit kurzem Takt.
melder = Hausmelder(webhook_url=url, status_datei=statusdatei, killswitch=killswitch_leser(registry),
takt_s=0.1, lebenszeichen_s=1.5)
melder.starte()
if not warte_bis(lambda: len(EMPFANGEN) == 1):
befunde.append("kein erster POST ohne Statusdatei")
erster = EMPFANGEN[0]["daten"] if EMPFANGEN else {}
print("1. ohne Statusdatei:", json.dumps(erster, ensure_ascii=False))
if erster.get("phase") != "kein_status" or erster.get("killswitch") != "angehalten" or erster.get("herzschlag_alter_s") != -1:
befunde.append("Rueckfall ohne Statusdatei stimmt nicht")
if EMPFANGEN and (EMPFANGEN[0]["pfad"] != f"/api/webhook/{kennung}" or EMPFANGEN[0]["typ"] != "application/json"):
befunde.append("Pfad oder Content-Type falsch")
status(statusdatei, phase="handelt", position=20, kontowert=100012.345)
warte_bis(lambda: len(EMPFANGEN) == 2)
zweiter = EMPFANGEN[-1]["daten"]
print("2. nach Statusaenderung:", json.dumps(zweiter, ensure_ascii=False))
if zweiter.get("position") != 20 or zweiter.get("tagesergebnis_usd") != 12.35 or zweiter.get("phase") != "handelt":
befunde.append("Statuswerte kommen nicht an")
# Ohne Aenderung: In 0,8 s (acht Takte) darf nichts kommen; der Herzschlag wird dabei aelter.
stand = len(EMPFANGEN)
time.sleep(0.8)
if len(EMPFANGEN) != stand:
befunde.append(f"ohne Aenderung kamen {len(EMPFANGEN) - stand} POST(s)")
# Lebenszeichen nach 1,5 s.
if not warte_bis(lambda: len(EMPFANGEN) == stand + 1, 3.0):
befunde.append("kein Lebenszeichen")
print(f"3. ohne Aenderung still, Lebenszeichen nach der Frist: {len(EMPFANGEN)} POSTs bisher")
# Killswitch aus dem Register, ohne dass die Statusdatei sich aendert.
registry.setze_killswitch_zustand("gesperrt", ausloeser="pascal")
stand = len(EMPFANGEN)
warte_bis(lambda: len(EMPFANGEN) == stand + 1)
print("4. Killswitch:", EMPFANGEN[-1]["daten"].get("killswitch"))
if EMPFANGEN[-1]["daten"].get("killswitch") != "gesperrt":
befunde.append("Killswitch-Wechsel kommt nicht an")
# Fehlschlag: 500 kippt den Merker nicht, es wird wiederholt; danach 200.
ANTWORT["code"] = 500
status(statusdatei, phase="pausiert", position=0, kontowert=99990.0)
stand = len(EMPFANGEN)
warte_bis(lambda: len(EMPFANGEN) >= stand + 3)
wiederholt = len(EMPFANGEN) - stand
ANTWORT["code"] = 200
warte_bis(lambda: EMPFANGEN[-1]["antwort"] == 200)
stand = len(EMPFANGEN)
time.sleep(0.6)
print(f"5. bei HTTP 500 wiederholt: {wiederholt} Versuche; nach 200 still: {len(EMPFANGEN) - stand} weitere; "
f"Fehlschlaege: {sorted(set(melder.fehlschlaege))}")
if wiederholt < 3 or len(EMPFANGEN) != stand:
befunde.append("Wiederholung nach Fehlschlag stimmt nicht")
if EMPFANGEN[-1]["daten"].get("tagesergebnis_usd") != -10.0:
befunde.append("negatives Tagesergebnis stimmt nicht")
# Weiterleitung: wird nicht gefolgt (kein GET am Server), gilt als Fehlschlag.
ANTWORT["code"] = 302
ANTWORT["location"] = f"http://127.0.0.1:{port}/woanders"
status(statusdatei, phase="handelt", position=10)
stand = len(EMPFANGEN)
warte_bis(lambda: len(EMPFANGEN) >= stand + 2)
gets = [e for e in EMPFANGEN if e["typ"] == "GET" or e["pfad"] == "/woanders"]
ANTWORT["code"] = 200
ANTWORT["location"] = None
warte_bis(lambda: EMPFANGEN[-1]["antwort"] == 200)
print(f"6. Weiterleitung: gefolgt {len(gets)}-mal; Fehlschlaege: {sorted(set(melder.fehlschlaege))}")
if gets:
befunde.append("einer Weiterleitung wurde gefolgt")
# Veraltet: Herzschlag 200 s alt, Phase nicht beendet.
status(statusdatei, phase="handelt", position=10,
zeit=datetime.fromtimestamp(time.time() - 200, UTC).isoformat())
stand = len(EMPFANGEN)
warte_bis(lambda: len(EMPFANGEN) >= stand + 1)
print("7. veraltet:", EMPFANGEN[-1]["daten"].get("veraltet"), EMPFANGEN[-1]["daten"].get("herzschlag_alter_s"))
if EMPFANGEN[-1]["daten"].get("veraltet") is not True:
befunde.append("veraltet wird nicht gemeldet")
melder.stoppe()
if melder.laeuft:
befunde.append("Faden laeuft nach stoppe() weiter")
geheim = [text for text in melder.fehlschlaege if kennung in text or "127.0.0.1" in text]
if geheim or kennung in repr(melder):
befunde.append("Adresse in Fehlschlag oder repr")
schluessel = sorted(EMPFANGEN[-1]["daten"])
print("8. Schluessel der Nutzlast:", schluessel)
if schluessel != sorted(["phase", "modus", "killswitch", "position", "tagesergebnis_usd",
"herzschlag_alter_s", "veraltet", "gesendet"]):
befunde.append("Nutzlast traegt andere Schluessel")
# Teil 1b: Verdrahtung ueber die Geheimnisdatei.
secrets = WEG / "secrets.toml"
secrets.write_text(f'[homeassistant]\nwebhook_url = "{url}"\naktiv = true\n', encoding="utf-8")
stand = len(EMPFANGEN)
zweiter_melder = starte_hausmelder(status_datei=statusdatei, killswitch=killswitch_leser(registry),
secrets_pfad=secrets)
ok = zweiter_melder is not None and warte_bis(lambda: len(EMPFANGEN) > stand)
if zweiter_melder is not None:
zweiter_melder.stoppe()
print(f"9. starte_hausmelder mit aktiv = true: Faden {'lief und hat gesendet' if ok else 'FEHLT'}")
if not ok:
befunde.append("starte_hausmelder sendet nicht")
secrets.write_text(f'[homeassistant]\nwebhook_url = "{url}"\naktiv = false\n', encoding="utf-8")
if starte_hausmelder(status_datei=statusdatei, killswitch=killswitch_leser(registry), secrets_pfad=secrets) is not None:
befunde.append("aktiv = false startet einen Faden")
secrets.write_text('[homeassistant]\nwebhook_url = "ftp://falsch"\naktiv = true\n', encoding="utf-8")
if starte_hausmelder(status_datei=statusdatei, killswitch=killswitch_leser(registry), secrets_pfad=secrets) is not None:
befunde.append("eine falsche Adresse startet einen Faden")
server.shutdown()
# Teil 2: ein POST ans echte HA, mit Werten, die sich von der neutralen Lage unterscheiden.
if ECHT_SECRETS is not None:
config = lies_hausmelder_config(ECHT_SECRETS)
if config is None or not config.webhook_url:
befunde.append("echte Geheimnisdatei ohne [homeassistant]")
else:
status(statusdatei, modus="generalprobe", phase="handelt", position=7, kontowert=100004.56)
registry.setze_killswitch_zustand("aktiv", ausloeser="pascal")
echter = Hausmelder(webhook_url=config.webhook_url, status_datei=statusdatei,
killswitch=killswitch_leser(registry))
echter.pruefe()
print(f"10. echtes HA: zugestellt={echter._zuletzt is not None}, Fehlschlaege={echter.fehlschlaege}, "
f"Laenge der Adresse {len(config.webhook_url)}")
if echter._zuletzt is None:
befunde.append("echtes HA hat nicht mit 200 geantwortet")
print(f"{len(befunde)} Befund(e): {befunde}")
return 2 if befunde else 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,509 @@
"""Killswitch-Simulation: die echte Funktion _generalprobe aus scripts/paper.py, ohne Netz und ohne das echte Konto.
Koordinator-Pruefcode zur Trading Plattform, nicht Teil des Projekt-Repos. Entstanden am 2026-09-21 auf Pascals Auftrag
"fuehre selbststaendig eine Killswitch-Simulation durch". Die Killswitch-UEBUNG des Projekts (scripts/killswitch_uebung.py)
ist dafuer das falsche Werkzeug, sobald der Paper-Runner als Dienst laeuft: Sie stellt echte Orders und schreibt ins
gemeinsame Register. Diese Simulation fasst weder das Konto noch das echte Register an.
VERDRAHTUNG (wie generalprobe_echtlauf.py): Gerufen wird die echte Funktion _generalprobe -- dieselbe Schleife, derselbe
Agent aus dem Register, dieselben Grenzen wie im Betrieb. Umgebogen sind nur die Stellen, die sonst ins Netz oder in den
Betrieb gingen: REGISTER zeigt auf eine Sicherungskopie im Wegwerf-Ordner (das echte wird nur lesend geoeffnet),
Statusdatei und Protokoll liegen im Wegwerf-Ordner, die Kerzen kommen aus einer JSON-Datei (hole_kerzen.py),
AlpacaBroker.from_secrets liefert ein mitschreibendes Konto, das glatt ist und bei jedem schreibenden Aufruf wirft,
requests.Session.request wirft, und der Melder schreibt in einen mitschreibenden Transport. "Pascal" klickt aus einem
zweiten Faden -- wie das Dashboard mit einem frischen Killswitch-Objekt aus dem Register.
SZENARIEN
hand Pascal klickt alle fuenf Stufen: fortsetzen, anhalten, fortsetzen, glattstellen, freigeben.
sturz Ab 11:00 Uhr New Yorker Zeit stehen alle Kerzen zehn Prozent tiefer: Der Runner muss die Tagesverlustgrenze
selbst reissen, glattstellen und sperren -- und nach Pascals "Freigeben" gesperrt bleiben, solange der
Verlust steht, ohne neue Order.
neustart Das Dashboard startet mitten am Tag neu (Killswitch.laden, Ausloeser neustart): Der Runner pausiert mit
stehender Position, meldet die Pause, und die Latenzanzeige zaehlt den Neustart nicht als Klick.
WAS GEPRUEFT WIRD, UND WOHER DIE ERWARTUNG STAMMT
- Eine Order entsteht nur in einem Takt, der im Zustand aktiv endet; kein Takt ausserhalb von aktiv traegt die Phase
handelt oder schliesst. Projektvorgabe: docs/code/killswitch.md, docs/code/paper.md
- Anhalten laesst die Position stehen; Glattstellen schliesst sie, sperrt, und gesperrt bleibt es bis zum Freigeben.
Projektvorgabe: Administrator-Anleitung, "Killswitch ueben"
- Tagesverlust: Der Runner stellt selbst glatt (Ausloeser paper-runner), meldet "Grenze verletzt" mit
"Killswitch: gesperrt", und sperrt nach einem Freigeben im selben Takt wieder.
Quelle: trading/paper/sitzung.py, _grenzverletzung und der Takt
- Ein Neustart ist kein Klick; die Pause wird gemeldet, wenn nicht Pascal gedrueckt hat.
Quelle: trading/paper/sitzung.py, KLICK_AUSLOESER und
_melde_pause_von_aussen
- Kein Netzversuch, am "echten" Konto nur die drei lesenden Aufrufe der Startbedingung, jede Meldung mit dem Vorsatz
der Generalprobe, keine tagesstart.json, Position am Ende null, Exit 0.
Projektvorgabe: Spec der Generalprobe (Paket G)
- Reaktion auf einen Klick hoechstens zwei Takte plus eine Sekunde. Setzung dieses Werkzeugs, keine Projektvorgabe
GRENZEN -- ein Fangnetz, kein Nachweis
Gefuellt wird gespielt: Der Weg einer echten Order zu Alpaca (Annahme, Teilausfuehrung, Zeitlimit) kommt nicht vor.
Der Klick geht nicht ueber HTTP durch das Dashboard, sondern ueber dasselbe Killswitch-Objekt, das das Dashboard baut.
Der Watchdog laeuft mit, wird aber nicht gezielt ausgeloest. Eine Short-Position kommt mit dem Agenten im Betrieb
(konstante Politik, Stufe +1) nicht vor.
REGRESSIONSNACHWEIS: killswitch_simulation_mutanten.py macht den Betriebscode an sechs Stellen kaputt; die Simulation
muss jeden Mutanten mit Exit 2 melden. Ergebnis und Datum stehen in der LIESMICH.
Exit 0: kein Befund. Exit 2: Befunde (auch ein Absturz ist ein Befund und steht in der Liste).
Aufruf (WSL, Venv des Projekts):
python killswitch_simulation.py <hand|sturz|neustart> <code-baum> <echter-projektordner> <kerzen.json> <wegwerf-ordner> [tempo]
"""
import importlib.util
import json
import os
import sqlite3
import sys
import threading
import time
import traceback
from dataclasses import replace
from datetime import UTC
from datetime import datetime
from datetime import timedelta
from pathlib import Path
STURZ_AB = (11, 0)
STURZ_FAKTOR = 0.90
PLAENE = {
"hand": [("fortsetzen", (9, 31)), ("anhalten", (10, 30)), ("fortsetzen", (10, 36)),
("glattstellen", (12, 0)), ("freigeben", (12, 10))],
"sturz": [("fortsetzen", (9, 31)), ("freigeben", (13, 0))],
"neustart": [("fortsetzen", (9, 31)), ("NEUSTART", (10, 30)), ("fortsetzen", (10, 40))],
}
BEFUNDE: list[str] = []
NETZVERSUCHE: list[str] = []
def pruefe(bedingung, satz):
if not bedingung:
BEFUNDE.append(satz)
def main() -> None:
if len(sys.argv) < 6 or sys.argv[1] not in PLAENE:
BEFUNDE.append("Aufruf: killswitch_simulation.py <hand|sturz|neustart> <code-baum> <echter-projektordner> "
"<kerzen.json> <wegwerf-ordner> [tempo]")
return
szenario = sys.argv[1]
code_baum = Path(sys.argv[2]).resolve()
echt = Path(sys.argv[3]).resolve()
kerzen_datei = Path(sys.argv[4]).resolve()
weg = Path(sys.argv[5]).resolve()
tempo = float(sys.argv[6]) if len(sys.argv) > 6 else 1.0
# Eine fehlende Voraussetzung ist ein Befund, kein Grund zu schweigen.
for pfad, was in ((code_baum / "scripts" / "paper.py", "Code-Baum ohne scripts/paper.py"),
(echt / "data" / "registry.db", "echter Projektordner ohne data/registry.db"),
(kerzen_datei, "Kerzendatei fehlt")):
if not pfad.exists():
BEFUNDE.append(f"Voraussetzung fehlt: {was}")
if weg.exists():
BEFUNDE.append("Voraussetzung fehlt: Der Wegwerf-Ordner existiert schon; die Simulation schreibt nur in einen neuen")
if BEFUNDE:
return
os.chdir(code_baum)
sys.path.insert(0, str(code_baum))
import requests
from trading.broker.schnittstelle import Marktuhr
from trading.killswitch.killswitch import Killswitch
from trading.killswitch.killswitch import lade_killswitch_config
from trading.paper.config import lade_paper_config
from trading.paper.kerzen import HANDELSZONE
from trading.paper.kerzen import Kerze
from trading.paper.melder import Melder
from trading.paper.melder import MelderConfig
from trading.paper.status import lies_status
from trading.registry.registry import Registry
def kein_netz(self, method, url, *args, **kwargs):
NETZVERSUCHE.append(f"{method} {url.split('?')[0]}")
raise requests.ConnectionError("Simulation: kein Netz")
requests.sessions.Session.request = kein_netz
class Antwort:
status_code = 200
class Transport:
def __init__(self):
self.aufrufe = []
def __call__(self, url, *, data=None, headers=None, timeout=None, **rest):
rumpf = data.decode("utf-8") if isinstance(data, bytes) else str(data)
self.aufrufe.append({"titel": headers.get("Title"), "prioritaet": headers.get("Priority"),
"zeilen": rumpf.split("\n")})
return Antwort()
class MitschreibendesKonto:
"""Steht fuer das echte Alpaca-Paper-Konto: glatt, keine Order, Boerse zu. Schreiben wirft."""
def __init__(self, name):
self.name = name
self.aufrufe = []
def positionen(self):
self.aufrufe.append("positionen")
return []
def offene_orders(self):
self.aufrufe.append("offene_orders")
return []
def marktuhr(self):
self.aufrufe.append("marktuhr")
jetzt = datetime.now(UTC)
return Marktuhr(is_open=False, next_open=jetzt + timedelta(hours=8), next_close=jetzt + timedelta(hours=14))
def storniere_alle(self):
self.aufrufe.append("storniere_alle")
return []
def schliesse_alle(self):
self.aufrufe.append("schliesse_alle")
return []
def __getattr__(self, name):
def wirft(*a, **k):
self.aufrufe.append(f"VERBOTEN:{name}")
raise AssertionError(f"{self.name}: unerwarteter Aufruf {name}")
return wirft
def ortszeit(text):
return datetime.fromisoformat(text).astimezone(HANDELSZONE)
roh = json.loads(kerzen_datei.read_text(encoding="utf-8"))
if not isinstance(roh, list) or not roh:
BEFUNDE.append("Voraussetzung fehlt: Die Kerzendatei ist keine nicht leere Liste")
return
kerzen = [Kerze(zeit=datetime.fromisoformat(e["zeit"]), open=e["open"], high=e["high"], low=e["low"],
close=e["close"], volume=e["volume"]) for e in roh]
if szenario == "sturz":
gestuerzt = []
for kerze in kerzen:
ort = kerze.zeit.astimezone(HANDELSZONE)
if (ort.hour, ort.minute) >= STURZ_AB:
kerze = replace(kerze, open=kerze.open * STURZ_FAKTOR, high=kerze.high * STURZ_FAKTOR,
low=kerze.low * STURZ_FAKTOR, close=kerze.close * STURZ_FAKTOR)
gestuerzt.append(kerze)
kerzen = gestuerzt
tag = kerzen[0].zeit.astimezone(HANDELSZONE).date().isoformat()
weg.mkdir(parents=True, exist_ok=False)
# Sicherungskopie des echten Registers; das echte wird nur lesend geoeffnet.
quelle = sqlite3.connect(f"file:{echt / 'data' / 'registry.db'}?mode=ro", uri=True)
ziel = sqlite3.connect(weg / "registry.db")
quelle.backup(ziel)
quelle.close()
# artefakt_pfad ist relativ zum Projektordner; in der Kopie wird er absolut, damit der Lauf aus dem Code-Baum
# heraus das Registermodell im echten artifacts-Ordner findet (nur lesen).
for zeile in ziel.execute("SELECT agent, version, artefakt_pfad FROM versionen").fetchall():
pfad = Path(zeile[2].replace("\\", "/"))
if not pfad.is_absolute():
ziel.execute("UPDATE versionen SET artefakt_pfad = ? WHERE agent = ? AND version = ?",
(str(echt / pfad), zeile[0], zeile[1]))
ziel.commit()
ziel.close()
ereignisse_vorher = len(Registry(db_pfad=weg / "registry.db", artefakt_wurzel=echt / "artifacts").killswitch_ereignisse())
spec = importlib.util.spec_from_file_location("paper_skript_simulation", code_baum / "scripts" / "paper.py")
paper = importlib.util.module_from_spec(spec)
spec.loader.exec_module(paper)
paper.REGISTER = str(weg / "registry.db")
paper.ARTEFAKTE = str(echt / "artifacts")
konto_der_probe = MitschreibendesKonto("Konto der Generalprobe")
konto_des_dashboards = MitschreibendesKonto("Konto des Dashboards")
class Kerzenquelle:
@classmethod
def from_secrets(cls, *a, **k):
return cls()
def kerzen_des_tages(self, instrument_id, ende):
return list(kerzen)
class BrokerFabrik:
@classmethod
def from_secrets(cls, *a, **k):
return konto_der_probe
paper.AlpacaKerzenquelle = Kerzenquelle
paper.AlpacaBroker = BrokerFabrik
config = replace(lade_paper_config("config/paper.toml"), status_datei=str(weg / "paper" / "status.json"),
protokoll_ordner=str(weg / "protokolle"))
ks_config = lade_killswitch_config("config/killswitch.toml")
transport = Transport()
melder = Melder(config=MelderConfig(aktiv=True, ntfy_url="https://ntfy.sh", thema="simulation-geht-nie-ins-netz",
zeitlimit_s=3.0, stille_nach_s=180.0), token="", transport=transport)
klicks = []
modi = set()
halt = threading.Event()
def dashboard_killswitch():
registry = Registry(db_pfad=weg / "registry.db", artefakt_wurzel=echt / "artifacts")
gespeichert = registry.killswitch_zustand()
return Killswitch(broker=konto_des_dashboards, registry=registry, config=ks_config,
zustand=gespeichert.zustand if gespeichert is not None else "angehalten")
def pascal():
plan = list(PLAENE[szenario])
while not halt.is_set():
status = lies_status(config.status_datei)
if isinstance(status, dict) and status.get("marktzeit"):
modi.add(status.get("modus"))
try:
ort = ortszeit(status["marktzeit"])
except (ValueError, TypeError):
time.sleep(0.2)
continue
if plan and ort.date().isoformat() == tag and (ort.hour, ort.minute) >= plan[0][1]:
was, _ = plan.pop(0)
eintrag = {"was": was, "marktzeit": f"{ort:%H:%M}", "position_vorher": status.get("position")}
try:
if was == "NEUSTART":
registry = Registry(db_pfad=weg / "registry.db", artefakt_wurzel=echt / "artifacts")
eintrag["vorher"] = registry.killswitch_zustand().zustand
neu_geladen = Killswitch.laden(broker=konto_des_dashboards, registry=registry, config=ks_config)
eintrag["nachher"] = neu_geladen.zustand
eintrag["erfolgreich"] = True
else:
ks = dashboard_killswitch()
eintrag["vorher"] = ks.zustand
if was == "glattstellen":
ergebnis = ks.glattstellen("pascal", "Killswitch-Simulation", ks_config.bestaetigungswort)
else:
ergebnis = getattr(ks, was)("pascal", "Killswitch-Simulation")
eintrag["nachher"] = ergebnis.zustand
eintrag["erfolgreich"] = ergebnis.erfolgreich
except Exception as ausnahme: # noqa: BLE001
eintrag["ausnahme"] = repr(ausnahme)
klicks.append(eintrag)
time.sleep(0.2)
faden = threading.Thread(target=pascal, name="pascal", daemon=True)
faden.start()
beginn = time.monotonic()
code = paper._generalprobe(config, tag, tempo, melder)
dauer = time.monotonic() - beginn
time.sleep(0.5)
halt.set()
faden.join(timeout=5)
# ---------------------------------------------------------------- Auswertung
zeilen = []
for pfad in sorted((weg / "protokolle").glob("*.jsonl")):
with pfad.open(encoding="utf-8") as datei:
zeilen += [json.loads(zeile) for zeile in datei if zeile.strip()]
takte = [z for z in zeilen if z.get("art") == "takt"]
ende = next((z for z in zeilen if z.get("art") == "ende"), {})
klickzeilen = [z for z in zeilen if z.get("art") == "latenz" and z.get("was") == "klick_reaktion"]
grenzereignisse = [z for z in zeilen if z.get("art") == "ereignis" and z.get("was") == "grenzwert"]
# Jeder Orderzeile den Takt zuordnen, in dem sie entstand: Die Taktzeile schliesst den Takt ab, steht also danach.
order_takte = []
for nummer, zeile in enumerate(zeilen):
if zeile.get("art") == "order":
order_takte.append((zeile, next((z for z in zeilen[nummer + 1:] if z.get("art") == "takt"), None)))
registry = Registry(db_pfad=weg / "registry.db", artefakt_wurzel=echt / "artifacts")
ereignisse = registry.killswitch_ereignisse()[ereignisse_vorher:]
schluss = lies_status(config.status_datei)
print("=" * 78)
print(f"Szenario {szenario}; Code {code_baum.name}; Tag {tag}, {len(kerzen)} Kerzen, {tempo:g} s je Marktminute, "
f"echte Dauer {dauer:.0f} s, Exit {code}")
print(f"Ende laut Protokoll: grund={ende.get('grund')} killswitch={ende.get('killswitch')} position={ende.get('position')} "
f"orders={ende.get('orders')} schritte={ende.get('schritte')} "
f"ergebnis={None if ende.get('kontowert') is None else round(ende['kontowert'] - ende['startkapital'], 2)} (gespielt)")
print("Klicks (Marktzeit New York):")
for k in klicks:
print(f" {k}")
print("Reaktion des Runners je Wechsel (klick_reaktion, echte Sekunden):")
for z in klickzeilen:
print(f" {z.get('sekunden')} s -> zustand {z.get('zustand')} (ausloeser {z.get('ausloeser')}, phase {z.get('phase')})")
print("Zustandsverlauf in den Taktzeilen (Marktzeit der letzten Kerze, Killswitch, Phase, Position):")
letzter = None
for t in takte:
schluessel = (t.get("killswitch"), t.get("phase"), t.get("position"))
if schluessel != letzter:
print(f" Takt {t.get('schritt'):>3} {ortszeit(t['letzte_kerze']):%H:%M} {str(t.get('killswitch')):<11} "
f"{str(t.get('phase')):<15} Position {t.get('position'):>3} gespieltes Ergebnis "
f"{round(t['kontowert'] - t['startkapital'], 2)}")
letzter = schluessel
print("Orders (mit dem Killswitch-Zustand ihres Takts):")
for zeile, folgetakt in order_takte:
print(f" {zeile.get('id')} {zeile.get('seite')} {zeile.get('menge')} -> {zeile.get('status')}, Position danach "
f"{zeile.get('position_danach')}; Takt {None if folgetakt is None else folgetakt.get('schritt')} "
f"killswitch={None if folgetakt is None else folgetakt.get('killswitch')} "
f"phase={None if folgetakt is None else folgetakt.get('phase')}")
print("Grenzereignisse im Protokoll:")
for z in grenzereignisse:
print(f" {z.get('zeit')} {z.get('text')} (phase {z.get('phase')})")
print("Killswitch-Ereignisse dieses Laufs im Wegwerf-Register:")
for e in ereignisse:
print(f" {e.stufe:<13} {e.ausloeser:<13} erfolgreich={int(e.erfolgreich)} {e.grund[:70]}")
print(f"Killswitch am Ende im Register: {registry.killswitch_zustand().zustand}")
print(f"Meldungen ({len(transport.aufrufe)}):")
for m in transport.aufrufe:
kern = [zeile for zeile in m["zeilen"] if zeile.startswith(("Grund:", "Killswitch:", "Position:"))]
print(f" [{m['prioritaet']}] {m['titel']} | {' | '.join(kern)}")
print(f"Aufrufe am Konto der Generalprobe: {konto_der_probe.aufrufe}")
print(f"Aufrufe am Konto des Dashboards: {konto_des_dashboards.aufrufe}")
print(f"Netzversuche: {NETZVERSUCHE}")
# ---------------------------------------------------------------- Pruefungen
pruefe(bool(takte), "kein einziger Takt im Protokoll")
pruefe(code == 0, f"Exit {code} statt 0")
pruefe(modi == {"generalprobe"}, f"Modi {modi}")
pruefe(not NETZVERSUCHE, "Netzversuch")
pruefe(sorted(konto_der_probe.aufrufe) == ["marktuhr", "offene_orders", "positionen"],
"Konto der Generalprobe: andere Aufrufe als die drei lesenden")
pruefe(not any(a.startswith("VERBOTEN") for a in konto_des_dashboards.aufrufe), "Konto des Dashboards: verbotener Aufruf")
pruefe(bool(transport.aufrufe) and all(str(m["titel"]).startswith("Generalprobe: ") for m in transport.aufrufe),
"keine Meldung, oder eine ohne den Vorsatz der Generalprobe")
pruefe(not list(weg.rglob("tagesstart.json")), "tagesstart.json entstanden")
pruefe(ende.get("position") == 0 and isinstance(schluss, dict) and schluss.get("position") == 0, "Position am Ende nicht null")
pruefe(len(klicks) == len(PLAENE[szenario]) and all("ausnahme" not in k for k in klicks), "nicht jeder Klick kam an")
pruefe(all(k.get("erfolgreich") for k in klicks), "ein Klick meldet erfolgreich=False")
pruefe(bool(order_takte), "keine einzige Order -- ohne Handel misst die Simulation den Killswitch nicht")
# Ein Klick kann mitten in einen Takt fallen. Dann traegt dessen Taktzeile schon den neuen Zustand, aber noch die
# Phase und den lokalen Bestand von vorher: Die Sitzung liest den Zustand mehrmals je Takt, und die Zeile entsteht
# am Taktende. Gemessen am 2026-09-21 im Schnelltempo: Takt 152 "gesperrt / handelt / Position 10" -- ohne
# Entscheidung und ohne Order, im naechsten Takt "pausiert / Position 0". Verbindlich ist deshalb alles AB DEM
# ZWEITEN Takt eines nicht aktiven Abschnitts; der erste wird ausgewiesen, nicht beurteilt. Eine Order in diesem
# ersten Takt ist ein Wettlauf zwischen Klick und Order (die Order ging unter 'aktiv' hinaus) und wird als Hinweis
# gedruckt -- im echten Betrieb faengt 'Anhalten' sie mit dem Stornieren, 'Glattstellen' mit dem Schliessen.
hinweise = []
erste_takte = set()
vorher_aktiv = True
for t in takte:
nicht_aktiv = t.get("killswitch") != "aktiv"
if nicht_aktiv and vorher_aktiv:
erste_takte.add(t.get("schritt"))
vorher_aktiv = not nicht_aktiv
for zeile, folgetakt in order_takte:
if folgetakt is not None and folgetakt.get("killswitch") != "aktiv" and folgetakt.get("schritt") in erste_takte:
hinweise.append(f"Wettlauf: Order {zeile.get('id')} und Zustandswechsel im selben Takt {folgetakt.get('schritt')}")
continue
pruefe(folgetakt is not None and folgetakt.get("killswitch") == "aktiv",
f"Order {zeile.get('id')} in einem Takt mit Killswitch {None if folgetakt is None else folgetakt.get('killswitch')}")
for t in takte:
if t.get("killswitch") == "aktiv":
continue
if t.get("schritt") in erste_takte:
if t.get("phase") in ("handelt", "schliesst"):
hinweise.append(f"Takt {t.get('schritt')}: Klick mitten im Takt -- die Zeile traegt schon "
f"'{t.get('killswitch')}', aber noch die Phase '{t.get('phase')}'")
continue
pruefe(t.get("orders") == 0 and t.get("phase") not in ("handelt", "schliesst"),
f"Takt {t.get('schritt')} handelt im Zustand {t.get('killswitch')}")
print(f"Hinweise ({len(hinweise)}): {hinweise}")
echte_klicks = [p for p in PLAENE[szenario] if p[0] != "NEUSTART"]
pruefe(len(klickzeilen) == len(echte_klicks) and all(z.get("ausloeser") == "pascal" for z in klickzeilen),
f"{len(klickzeilen)} Klick-Messungen statt {len(echte_klicks)} (ein Neustart ist kein Klick)")
pruefe(all(isinstance(z.get("sekunden"), (int, float)) and z["sekunden"] <= 2.0 * tempo + 1.0 for z in klickzeilen),
"eine Klick-Reaktion dauerte laenger als zwei Takte plus eine Sekunde")
if szenario == "hand":
erwartet = [("fortsetzen", "angehalten", "aktiv"), ("anhalten", "aktiv", "angehalten"),
("fortsetzen", "angehalten", "aktiv"), ("glattstellen", "aktiv", "gesperrt"),
("freigeben", "gesperrt", "aktiv")]
pruefe([(k["was"], k.get("vorher"), k.get("nachher")) for k in klicks] == erwartet, "Uebergaenge anders als geplant")
erster_handel = next((t for t in takte if t.get("position") != 0), None)
angehalten = [t for t in takte if t.get("killswitch") == "angehalten" and erster_handel is not None
and t.get("schritt", 0) > erster_handel.get("schritt", 0)]
pruefe(bool(angehalten), "kein Takt im Zustand angehalten nach dem ersten Handel")
pruefe(bool(angehalten) and len({t.get("position") for t in angehalten}) == 1
and angehalten[0].get("position") != 0, "Anhalten: Position blieb nicht stehen oder war null")
gesperrt = [t for t in takte if t.get("killswitch") == "gesperrt"]
pruefe(bool(gesperrt), "kein Takt im Zustand gesperrt")
pruefe(bool(gesperrt) and all(t.get("position") == 0 for t in gesperrt[1:]), "gesperrt: Position nicht null ab dem zweiten Takt")
if gesperrt:
danach = [t for t in takte if t.get("schritt", 0) > gesperrt[-1].get("schritt", 0)]
pruefe(any(t.get("killswitch") == "aktiv" and t.get("position") != 0 for t in danach),
"nach dem Freigeben wurde nicht wieder gehandelt")
pruefe(konto_des_dashboards.aufrufe.count("schliesse_alle") == 1, "Dashboard-Konto: schliesse_alle nicht genau einmal")
pruefe(konto_des_dashboards.aufrufe.count("storniere_alle") == 2, "Dashboard-Konto: storniere_alle nicht genau zweimal")
pruefe(ende.get("killswitch") == "aktiv", f"Killswitch am Ende {ende.get('killswitch')}")
elif szenario == "neustart":
erwartet = [("fortsetzen", "angehalten", "aktiv"), ("NEUSTART", "aktiv", "angehalten"),
("fortsetzen", "angehalten", "aktiv")]
pruefe([(k["was"], k.get("vorher"), k.get("nachher")) for k in klicks] == erwartet, "Uebergaenge anders als geplant")
erster_handel = next((t for t in takte if t.get("position") != 0), None)
pause = [t for t in takte if t.get("killswitch") == "angehalten" and erster_handel is not None
and t.get("schritt", 0) > erster_handel.get("schritt", 0)]
pruefe(bool(pause), "kein Takt im Zustand angehalten nach dem ersten Handel")
# Die Phase gilt ab dem zweiten Takt der Pause (siehe oben: ein Klick mitten im Takt), die Position ab dem ersten.
pruefe(bool(pause) and all(t.get("position") == erster_handel.get("position") for t in pause)
and all(t.get("phase") == "pausiert" for t in pause[1:]),
"Neustart: Position blieb nicht stehen oder Phase nicht pausiert")
if pause:
danach = [t for t in takte if t.get("schritt", 0) > pause[-1].get("schritt", 0)]
pruefe(any(t.get("killswitch") == "aktiv" and t.get("phase") == "handelt" for t in danach),
"nach dem Fortsetzen wurde nicht wieder gehandelt")
pausenmeldungen = [m for m in transport.aufrufe if "pausiert" in str(m["titel"])]
pruefe(len(pausenmeldungen) == 1, f"{len(pausenmeldungen)} Pausenmeldung(en); erwartet genau eine zum Neustart")
pruefe(all(any("neustart" in zeile for zeile in m["zeilen"]) for m in pausenmeldungen),
"die Pausenmeldung nennt den Ausloeser neustart nicht")
neustarts = [e for e in ereignisse if e.ausloeser == "neustart"]
pruefe(len(neustarts) == 2, f"{len(neustarts)} Neustart-Ereignisse; erwartet eines beim Start und eines mitten am Tag")
pruefe(ende.get("killswitch") == "aktiv", f"Killswitch am Ende {ende.get('killswitch')}")
else:
pruefe(bool(grenzereignisse) and all("tagesverlust" in str(z.get("text")) for z in grenzereignisse),
"kein Grenzereignis zum Tagesverlust")
vom_runner = [e for e in ereignisse if e.ausloeser == "paper-runner"]
pruefe(bool(vom_runner) and all(e.stufe == "glattstellen" and e.erfolgreich for e in vom_runner),
"kein erfolgreiches Glattstellen durch den Runner im Register")
sturz_takt = next((t for t in takte if (ortszeit(t["letzte_kerze"]).hour, ortszeit(t["letzte_kerze"]).minute) >= STURZ_AB), None)
sperr_takt = next((t for t in takte if t.get("killswitch") == "gesperrt"), None)
# Hoechstens ein Takt Verzug: Der Kontowert wird im Takt vor der Entscheidung gelesen.
pruefe(sturz_takt is not None and sperr_takt is not None
and 0 <= sperr_takt.get("schritt", 0) - sturz_takt.get("schritt", 0) <= 1 and sperr_takt.get("position") == 0,
"der Runner sperrt nicht spaetestens einen Takt nach der ersten gestuerzten Kerze, oder nicht glatt")
if sperr_takt is not None:
print(f"Sturz: erste gestuerzte Kerze in Takt {None if sturz_takt is None else sturz_takt.get('schritt')}, "
f"gesperrt und glatt ab Takt {sperr_takt.get('schritt')}")
danach = [t for t in takte if t.get("schritt", 0) >= sperr_takt.get("schritt", 0)]
pruefe(all(t.get("position") == 0 for t in danach), "nach dem Sturz noch einmal eine Position")
# Faellt das "Freigeben" mitten in einen Takt, kann genau dessen Zeile 'aktiv' tragen; der naechste Takt
# prueft die Grenze und sperrt wieder. Mehr als ein solcher Takt, oder einer mit Order oder Position, ist
# ein Befund.
offen = [t for t in danach if t.get("killswitch") != "gesperrt"]
pruefe(len(offen) <= 1 and all(t.get("orders") == 0 and t.get("position") == 0 for t in offen),
f"nach dem Sturz enden {len(offen)} Takte nicht gesperrt, oder einer davon mit Order oder Position")
grenze = danach[0]["startkapital"] * -config_grenze(code_baum)
pruefe(all(t["kontowert"] - t["startkapital"] < grenze for t in danach), "Kontowert nach dem Sturz nicht unter der Grenze")
spaete = [zeile for zeile, folgetakt in order_takte if folgetakt is not None
and folgetakt.get("schritt", 0) > sperr_takt.get("schritt", 0)]
pruefe(not spaete, f"{len(spaete)} Order(s) nach dem Sturz")
grenzmeldungen = [m for m in transport.aufrufe if "Grenze verletzt" in str(m["titel"])]
pruefe(len(grenzmeldungen) >= 2, f"{len(grenzmeldungen)} Grenzmeldung(en); erwartet eine beim Sturz und eine nach dem Freigeben")
pruefe(all(any(zeile == "Killswitch: gesperrt" for zeile in m["zeilen"]) for m in grenzmeldungen),
"eine Grenzmeldung nennt nicht 'Killswitch: gesperrt'")
pruefe(len(vom_runner) >= 2, f"{len(vom_runner)} Glattstellen durch den Runner; erwartet eines beim Sturz und eines nach dem Freigeben")
def config_grenze(code_baum: Path) -> float:
"""max_tagesverlust_anteil aus config/paper.toml des Code-Baums -- gelesen, nicht abgeschrieben."""
import tomllib
daten = tomllib.loads((code_baum / "config" / "paper.toml").read_text(encoding="utf-8"))
return float(daten["grenzen"]["max_tagesverlust_anteil"])
if __name__ == "__main__":
try:
main()
except BaseException as ausnahme: # noqa: BLE001 -- ein Absturz ist ein Befund, kein Ergebnis
BEFUNDE.append(f"Absturz: {ausnahme!r}")
traceback.print_exc()
finally:
print(f"{len(BEFUNDE)} Befund(e): {BEFUNDE}")
sys.exit(2 if BEFUNDE else 0)
@@ -0,0 +1,123 @@
"""Regressionsnachweis fuer killswitch_simulation.py: Sieht die Simulation einen kaputten Killswitch?
"0 Befunde" sagt nur etwas, wenn die Pruefungen anschlagen koennen. Dieser Treiber kopiert den Code-Baum in einen
Arbeitsordner, faehrt zuerst die Grundlinie (alle drei Szenarien muessen ohne Befund sein -- sonst misst der Rest nichts),
macht die Kopie dann je Mutant an EINER Stelle kaputt, faehrt das passende Szenario im Schnelltempo und erwartet Exit 2.
Nach jedem Mutanten wird die Datei zurueckgesetzt; am Ende laeuft jedes Szenario noch einmal unveraendert.
Die Mutanten sind Textersetzungen im Betriebscode. Trifft ein Muster nicht genau einmal -- weil sich der Code geaendert
hat --, ist das ein Befund dieses Treibers ("nicht anwendbar") und kein stiller Durchlass: Dann gehoert das Muster
nachgezogen.
Zwei Teile, weil ein Werkzeugaufruf der Sitzung hoechstens zehn Minuten laufen darf: Teil 1 faehrt die Grundlinie und die
ersten drei Mutanten, Teil 2 die letzten drei und die Gegenprobe nach dem Zuruecksetzen. Ein Lauf im Schnelltempo dauert
rund eine halbe Minute.
Kein Netz, kein echtes Konto: Gerufen wird nur killswitch_simulation.py.
Aufruf (WSL, Venv des Projekts):
python killswitch_simulation_mutanten.py <code-baum> <echter-projektordner> <kerzen.json> <arbeitsordner> <1|2>
Exit 0: jeder Mutant gesehen, Grundlinie und Gegenprobe sauber. Exit 1: sonst.
"""
import shutil
import subprocess
import sys
from pathlib import Path
TEMPO = "0.05"
SIMULATION = Path(__file__).resolve().parent / "killswitch_simulation.py"
MUTANTEN = [
("K1 der Runner uebergeht den Killswitch", "hand", "trading/paper/sitzung.py",
' self._zustand_text = (\n gespeichert.zustand if gespeichert is not None else "angehalten"\n )\n',
' self._zustand_text = "aktiv"\n'),
("K2 Glattstellen schliesst die gespielte Position nicht", "hand", "trading/paper/generalprobe.py",
" self._broker.storniere_alle()\n self._broker.schliesse_alle()\n",
" self._broker.storniere_alle()\n"),
# 0.02 ist der Wert aus config/risk.toml: Die engere Grenze darf nur verschaerfen; ein groesserer Wert braeche schon
# beim Laden ab und waere dann ein Abbruch statt eines uebersehenen Sturzes.
("K3 die Tagesverlustgrenze greift nicht", "sturz", "config/paper.toml",
"max_tagesverlust_anteil = 0.002", "max_tagesverlust_anteil = 0.02"),
("K4 jeder Ausloeser zaehlt als Klick", "neustart", "trading/paper/sitzung.py",
" if gespeichert.ausloeser != KLICK_AUSLOESER:\n", " if False:\n"),
("K5 keine Pausenmeldung beim Neustart", "neustart", "trading/paper/sitzung.py",
" if self._ausloeser_text == MENSCH:\n return\n", " return\n"),
("K6 der Tagesverlust haelt nur an, statt glattzustellen", "sturz", "trading/paper/sitzung.py",
" if grund == TAGESVERLUST:\n self._glattstellen(grund)\n else:\n self._anhalten(grund)\n",
" self._anhalten(grund)\n"),
]
def main() -> int:
if len(sys.argv) != 6 or sys.argv[5] not in ("1", "2"):
print(__doc__)
return 1
quelle = Path(sys.argv[1]).resolve()
echt = Path(sys.argv[2]).resolve()
kerzen = Path(sys.argv[3]).resolve()
arbeit = Path(sys.argv[4]).resolve()
teil = sys.argv[5]
kopie = arbeit / "kopie"
def fahre(szenario: str, lauf: str) -> tuple[int, str]:
ziel = arbeit / lauf
if ziel.exists():
shutil.rmtree(ziel)
ergebnis = subprocess.run(
[sys.executable, str(SIMULATION), szenario, str(kopie), str(echt), str(kerzen), str(ziel), TEMPO],
capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=600,
env={**__import__("os").environ, "PYTHONDONTWRITEBYTECODE": "1"},
)
(arbeit / f"{lauf}.txt").write_text(ergebnis.stdout + ergebnis.stderr, encoding="utf-8")
zeilen = (ergebnis.stdout + ergebnis.stderr).strip().splitlines()
befundzeile = next((z for z in reversed(zeilen) if "Befund(e)" in z), zeilen[-1] if zeilen else "")
return ergebnis.returncode, befundzeile[:500]
arbeit.mkdir(parents=True, exist_ok=True)
if kopie.exists():
shutil.rmtree(kopie)
shutil.copytree(quelle, kopie, ignore=shutil.ignore_patterns(".git", "node_modules", "dist", "__pycache__", ".pytest_cache"))
gestoert = []
if teil == "1":
for szenario in ("hand", "sturz", "neustart"):
code, befundzeile = fahre(szenario, f"grundlinie-{szenario}")
print(f"Grundlinie [{szenario}]: exit={code} -- {befundzeile}")
if code != 0:
print("Abbruch: Die Grundlinie ist nicht sauber; die Mutanten wuerden nichts messen.")
return 1
for nummer, (name, szenario, datei, alt, neu) in enumerate(MUTANTEN, start=1):
if (nummer <= 3) != (teil == "1"):
continue
pfad = kopie / datei
original = pfad.read_text(encoding="utf-8")
if original.count(alt) != 1:
print(f"{name}: Muster steht {original.count(alt)}-mal -- NICHT ANWENDBAR, Muster nachziehen")
gestoert.append(name + " (nicht anwendbar)")
continue
with open(pfad, "w", encoding="utf-8", newline="\n") as strom:
strom.write(original.replace(alt, neu))
try:
code, befundzeile = fahre(szenario, f"mutant-{nummer}")
finally:
with open(pfad, "w", encoding="utf-8", newline="\n") as strom:
strom.write(original)
urteil = "gesehen" if code == 2 else "UEBERSEHEN" if code == 0 else f"ABGEBROCHEN (exit {code})"
print(f"{name} [{szenario}]: {urteil} -- {befundzeile}")
if code != 2:
gestoert.append(name)
if teil == "2":
for szenario in ("hand", "sturz", "neustart"):
code, befundzeile = fahre(szenario, f"gegenprobe-{szenario}")
print(f"Gegenprobe nach dem Zuruecksetzen [{szenario}]: exit={code} -- {befundzeile}")
if code != 0:
gestoert.append(f"Gegenprobe {szenario}")
print(f"Teil {teil}: nicht gesehen oder gestoert: {gestoert}")
return 1 if gestoert else 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,131 @@
"""Rechnet Reaktionszeiten aus dem, was ein Lauf des Paper-Runners ohnehin hinterlaesst -- nur lesend.
Koordinator-Werkzeug fuer Tage, an denen der Runner seine Latenzen noch nicht selbst misst (vor Paket L).
Quellen: das JSONL-Protokoll des Laufs (jede Zeile traegt echte Zeit, auch in der Generalprobe) und die
Tabelle killswitch_ereignisse im Register (Zeitpunkt jedes Klicks). Das Register wird mit mode=ro geoeffnet.
Gerechnet wird:
Klick -> Runner reagiert Registerereignis (Ausloeser pascal, erfolgreich) bis zur ersten Taktzeile
danach, die den neuen Zustand traegt. Die Taktzeile entsteht am ENDE des
Takts -- der Wert ist eine obere Schranke fuer "gelesen" und sagt, ab wann
der neue Zustand im Protokoll wirkt.
Kerze -> Taktende nur im Modus paper: Taktzeile mit frischer Kerze und Aktion, Zeit der Zeile
minus (juengste Kerze + 60 s). Obere Schranke fuer Kerze -> Entscheidung,
weil die Zeile erst nach den Orders geschrieben wird. In der Generalprobe
sinnlos (Kerzen von einem anderen Tag) und deshalb ausgelassen.
Taktabstand Abstand aufeinanderfolgender Taktzeilen -- der Rahmen fuer alles andere.
Nicht gerechnet: Fuellzeit (dafuer handelstag_auswertung.py --broker) und Meldungen (ntfy_lesen.py).
Gibt nie einen Kontowert aus.
Aufruf (WSL, aus dem Projektordner):
python latenz_auswertung.py <protokoll.jsonl> [<register.db>]
"""
import json
import sqlite3
import statistics
import sys
from datetime import datetime
from datetime import timedelta
from pathlib import Path
NEUER_ZUSTAND = {"anhalten": "angehalten", "glattstellen": "gesperrt", "fortsetzen": "aktiv", "freigeben": "aktiv"}
KLICKER = "pascal"
def zeit(text: str) -> datetime:
return datetime.fromisoformat(text)
def kennzahlen(name: str, werte: list[float], einheit: str = "s") -> None:
if not werte:
print(f" {name}: keine Messung")
return
sortiert = sorted(werte)
p90 = sortiert[min(len(sortiert) - 1, int(round(0.9 * (len(sortiert) - 1))))]
print(
f" {name}: n={len(werte)} Median {statistics.median(werte):.1f} {einheit} "
f"Mittel {statistics.fmean(werte):.1f} p90 {p90:.1f} Min {sortiert[0]:.1f} Max {sortiert[-1]:.1f}"
)
def main() -> int:
protokoll = Path(sys.argv[1])
register = Path(sys.argv[2]) if len(sys.argv) > 2 else Path("data/registry.db")
zeilen = []
with open(protokoll, encoding="utf-8") as strom: # Dateiiteration, nie splitlines (U+2028)
for roh in strom:
if roh.strip():
zeilen.append(json.loads(roh))
kopf = next((z for z in zeilen if z.get("art") == "kopf"), {})
takte = [z for z in zeilen if z.get("art") == "takt"]
modus = kopf.get("modus", "unbekannt")
if not takte:
print("Keine Taktzeile im Protokoll.")
return 1
start, ende = zeit(zeilen[0]["zeit"]), zeit(zeilen[-1]["zeit"])
print(f"Protokoll {protokoll.name}: Modus {modus}, {len(takte)} Takte, {start:%H:%M:%S} bis {ende:%H:%M:%S} UTC")
abstaende = [
(zeit(b["zeit"]) - zeit(a["zeit"])).total_seconds() for a, b in zip(takte, takte[1:])
]
print("Rahmen:")
kennzahlen("Taktabstand", abstaende)
print("Klick -> Runner reagiert (Register gegen erste Taktzeile mit dem neuen Zustand):")
if not register.exists():
print(f" Register {register} fehlt -- ausgelassen")
else:
verbindung = sqlite3.connect(f"file:{register.as_posix()}?mode=ro", uri=True)
try:
ereignisse = verbindung.execute(
"SELECT zeit, stufe, ausloeser, erfolgreich FROM killswitch_ereignisse ORDER BY id"
).fetchall()
finally:
verbindung.close()
reaktionen = []
for ereignis_zeit, stufe, ausloeser, erfolgreich in ereignisse:
wann = zeit(ereignis_zeit)
if not (start <= wann <= ende):
continue
if ausloeser != KLICKER:
print(f" {wann:%H:%M:%S} {stufe} von {ausloeser}: kein Klick, ausgelassen")
continue
if not erfolgreich:
print(f" {wann:%H:%M:%S} {stufe}: nicht erfolgreich, ausgelassen")
continue
neu = NEUER_ZUSTAND.get(stufe)
treffer = next(
(t for t in takte if zeit(t["zeit"]) > wann and t.get("killswitch") == neu), None
)
if treffer is None:
print(f" {wann:%H:%M:%S} {stufe}: keine Taktzeile mit Zustand {neu} danach")
continue
sekunden = (zeit(treffer["zeit"]) - wann).total_seconds()
reaktionen.append(sekunden)
print(f" {wann:%H:%M:%S} {stufe:<13} -> Takt {treffer.get('schritt')} ({treffer.get('phase')}) nach {sekunden:5.1f} s")
kennzahlen("Klick -> Reaktion", reaktionen)
if modus == "paper":
print("Kerze -> Taktende (obere Schranke fuer Kerze -> Entscheidung):")
werte = []
for takt in takte:
if takt.get("frische_kerze") and takt.get("aktion") is not None and takt.get("letzte_kerze"):
schluss = zeit(takt["letzte_kerze"]) + timedelta(seconds=60)
werte.append((zeit(takt["zeit"]) - schluss).total_seconds())
kennzahlen("Kerze -> Taktende", werte)
ohne = sum(1 for takt in takte if takt.get("markt_offen") and not takt.get("frische_kerze"))
print(f" Takte bei offenem Markt ohne frische Kerze: {ohne} von {len(takte)}")
else:
print("Kerze -> Entscheidung: im Modus", modus, "nicht gerechnet (Kerzen eines anderen Tages).")
orders = [z for z in zeilen if z.get("art") == "order"]
meldefehler = [z for z in zeilen if z.get("art") == "ereignis" and z.get("was") == "benachrichtigung_fehlgeschlagen"]
print(f"Dazu: {len(orders)} Orderzeile(n), {len(meldefehler)} fehlgeschlagene Meldung(en).")
return 0
if __name__ == "__main__":
sys.exit(main())
@@ -0,0 +1,65 @@
"""Liest mit dem Alpaca-Klienten eines Code-Baums die echten Antworten des Paper-Kontos -- NUR LESEND, ohne Betraege auszugeben.
Koordinator-Pruefcode zu Paket S1 (strengerer Parser in trading/broker/alpaca.py). Frage: Liest der neue Parser
jede echte Antwort, die der alte las -- Konto, Positionen, offene Orders, die Orderhistorie (gefuellt, storniert),
Marktuhr, letzter Kurs? Ein Parser, der eine echte Order als unlesbar wertet, waere am Handelstag eine Stoerung.
Aufruf (WSL): python s1_alpaca_lesen.py <code-baum> <echter-projektordner>
"""
import collections
import dataclasses
import os
import sys
from pathlib import Path
CODE = Path(sys.argv[1]).resolve()
ECHT = Path(sys.argv[2]).resolve()
sys.path.insert(0, str(CODE))
os.chdir(ECHT)
from trading.broker.alpaca import AlpacaBroker # noqa: E402
from trading.risk.limits import lade_limits # noqa: E402
limits = lade_limits("config/risk.toml")
broker = AlpacaBroker.from_secrets("config/secrets.toml", universum=limits.universum, feed="iex", timeout=10.0)
def form(objekt) -> str:
"""Feldnamen und Typen eines Dataclass-Objekts, nie die Werte."""
if dataclasses.is_dataclass(objekt):
return ", ".join(f"{f.name}:{type(getattr(objekt, f.name)).__name__}" for f in dataclasses.fields(objekt))
return type(objekt).__name__
schritte = [
("konto", lambda: broker.konto()),
("positionen", lambda: broker.positionen()),
("offene_orders", lambda: broker.offene_orders()),
("alle_orders(200)", lambda: broker.alle_orders(limit=200)),
("marktuhr", lambda: broker.marktuhr()),
("letzter_kurs", lambda: broker.letzter_kurs("AAPL.XNAS")),
("verbindung_ok", lambda: broker.verbindung_ok()),
]
befunde = 0
for name, aufruf in schritte:
try:
ergebnis = aufruf()
except Exception as ausnahme: # noqa: BLE001
befunde += 1
# Der Text eines BrokerFehler kann Antwortteile tragen; nur Typ und die ersten Zeichen ohne Ziffernfolgen.
print(f"{name}: AUSNAHME {type(ausnahme).__name__}: {str(ausnahme)[:160]}")
continue
if isinstance(ergebnis, list):
print(f"{name}: {len(ergebnis)} Eintraege" + (f"; Form: {form(ergebnis[0])}" if ergebnis else ""))
if ergebnis and hasattr(ergebnis[0], "status"):
print(" Status:", dict(collections.Counter(getattr(e, "status", "?") for e in ergebnis)))
leer = collections.Counter()
for eintrag in ergebnis:
for feld in dataclasses.fields(eintrag):
if getattr(eintrag, feld.name) is None:
leer[feld.name] += 1
print(" Felder mit None:", dict(leer))
else:
print(f"{name}: {form(ergebnis)}")
print(f"{befunde} Aufruf(e) mit Ausnahme.")
sys.exit(2 if befunde else 0)
@@ -0,0 +1,216 @@
"""Gegenmessung zu Paket S5 gegen eine laufende Probe-Instanz: dieselben Angriffe, die der Sicherheitspruefer gemessen hat.
Koordinator-Pruefcode (WSL). Gemessen wird gegen 127.0.0.1:<port>; die Instanz laeuft aus einer Kopie mit
Broker-Attrappe, Wegwerf-Register und Wegwerf-Zugang. Nichts davon beruehrt den Betrieb.
Aufruf: python s5_echtlauf.py <port> <ablage-der-probe> <code-baum>
"""
import base64
import hashlib
import hmac
import http.client
import json
import os
import struct
import subprocess
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import requests
PORT = int(sys.argv[1])
ABLAGE = Path(sys.argv[2])
CODE = Path(sys.argv[3])
BASIS = f"http://127.0.0.1:{PORT}"
KOPF = {"Origin": BASIS}
z = json.loads((ABLAGE / "zugang.json").read_text(encoding="utf-8"))
BEN, PW, GEH = z["benutzer"], z["passwort"], z["geheimnis"]
befunde = []
def code(versatz=0):
s = base64.b32decode(GEH)
mac = hmac.new(s, struct.pack(">Q", int(time.time()) // 30 + versatz), hashlib.sha1).digest()
v = mac[-1] & 0x0F
return f"{(struct.unpack('>I', mac[v:v + 4])[0] & 0x7FFFFFFF) % 1_000_000:06d}"
def melde_an():
s = requests.Session()
s.trust_env = False
a = s.post(f"{BASIS}/api/anmeldung", json={"benutzer": BEN, "passwort": PW, "code": code()}, headers=KOPF, timeout=60)
return s, a.status_code
def naechster_schritt():
time.sleep(30.5 - (time.time() % 30))
def server_pid():
aus = subprocess.run(["pgrep", "-f", "start_probe_w.py"], capture_output=True, text=True).stdout.split()
return int(aus[0]) if aus else None
def speicher_mib(pid):
werte = {}
for zeile in Path(f"/proc/{pid}/status").read_text().splitlines():
if zeile.startswith(("VmRSS", "VmHWM")):
name, wert = zeile.split(":")
werte[name] = int(wert.split()[0]) / 1024
return werte
def pruefe(text, ok):
print((" OK " if ok else " BEFUND ") + text)
if not ok:
befunde.append(text)
pid = server_pid()
print(f"Server-PID {pid}, Speicher zu Beginn {speicher_mib(pid)}")
print("== 1: 422 spiegelt die Eingabe nicht mehr")
lang = "P" * 300
a = requests.post(f"{BASIS}/api/anmeldung", json={"benutzer": BEN, "passwort": lang, "code": "123456"}, headers=KOPF, timeout=10)
pruefe(f"HTTP {a.status_code}, Passwort im Koerper: {lang in a.text}", a.status_code == 422 and lang not in a.text)
print(f" Koerper: {a.text[:200]}")
print("== 2: Rumpf 512 MiB mit content-length, unangemeldet")
vor = speicher_mib(pid)
t0 = time.perf_counter()
verbindung = http.client.HTTPConnection("127.0.0.1", PORT, timeout=60)
verbindung.putrequest("POST", "/api/anmeldung")
verbindung.putheader("Origin", BASIS)
verbindung.putheader("Content-Type", "application/json")
verbindung.putheader("Content-Length", str(512 * 1024 * 1024))
verbindung.endheaders()
gesendet = 0
status = None
try:
block = b"x" * (1024 * 1024)
for _ in range(512):
verbindung.send(block)
gesendet += len(block)
except OSError as ausnahme:
print(f" Senden brach nach {gesendet / 1024 / 1024:.0f} MiB ab: {type(ausnahme).__name__}")
try:
antwort = verbindung.getresponse()
status = antwort.status
antwort.read()
except Exception as ausnahme: # noqa: BLE001
print(f" Antwort nicht lesbar: {type(ausnahme).__name__}")
verbindung.close()
nach = speicher_mib(pid)
print(f" Status {status}, {time.perf_counter() - t0:.2f} s, gesendet {gesendet / 1024 / 1024:.0f} MiB, Speicher {vor} -> {nach}")
pruefe("413 und Spitzenspeicher waechst um weniger als 50 MiB", status == 413 and nach["VmHWM"] - vor["VmHWM"] < 50)
print("== 3: Rumpf ohne content-length (chunked), unangemeldet")
vor = speicher_mib(pid)
t0 = time.perf_counter()
verbindung = http.client.HTTPConnection("127.0.0.1", PORT, timeout=60)
verbindung.putrequest("POST", "/api/anmeldung")
verbindung.putheader("Origin", BASIS)
verbindung.putheader("Content-Type", "application/json")
verbindung.putheader("Transfer-Encoding", "chunked")
verbindung.endheaders()
gesendet = 0
status = None
try:
block = b"y" * (256 * 1024)
for _ in range(800):
verbindung.send(hex(len(block))[2:].encode() + b"\r\n" + block + b"\r\n")
gesendet += len(block)
verbindung.send(b"0\r\n\r\n")
except OSError as ausnahme:
print(f" Senden brach nach {gesendet / 1024 / 1024:.1f} MiB ab: {type(ausnahme).__name__}")
try:
antwort = verbindung.getresponse()
status = antwort.status
antwort.read()
except Exception as ausnahme: # noqa: BLE001
print(f" Antwort nicht lesbar: {type(ausnahme).__name__}")
verbindung.close()
nach = speicher_mib(pid)
print(f" Status {status}, {time.perf_counter() - t0:.2f} s, gesendet {gesendet / 1024 / 1024:.1f} MiB, Speicher {vor} -> {nach}")
pruefe("413 (oder Abbruch der Verbindung) und Spitzenspeicher waechst um weniger als 50 MiB",
status in (413, None) and nach["VmHWM"] - vor["VmHWM"] < 50)
print("== 4: Killswitch unter 40 parallelen teuren Kerzenabrufen")
sitz, st = melde_an()
print(f" Anmeldung HTTP {st}")
cookie = "; ".join(f"{c.name}={c.value}" for c in sitz.cookies)
ds = sitz.get(f"{BASIS}/api/markt/datensaetze", timeout=120).json().get("datensaetze", [])
intraday = [d for d in ds if d.get("intraday")]
if not intraday:
print(" kein Intraday-Datensatz im Katalog -- Lastmessung nicht moeglich")
befunde.append("Lastmessung nicht moeglich")
else:
bt = intraday[0]["bar_type"]
teuer = f"{BASIS}/api/markt/kerzen?bar_type={bt}&aufloesung_min=60&anzahl=3000"
t0 = time.perf_counter()
einzel = requests.get(teuer, headers={"Cookie": cookie}, timeout=600)
print(f" ein einzelner teurer Abruf: HTTP {einzel.status_code}, {time.perf_counter() - t0:.1f} s")
t0 = time.perf_counter()
sitz.post(f"{BASIS}/api/killswitch/anhalten", json={"grund": "grundlinie"}, headers=KOPF, timeout=60)
print(f" Grundlinie POST /api/killswitch/anhalten ohne Last: {time.perf_counter() - t0:.2f} s")
laeuft = [True]
messung = {"gesund": [], "anhalten": []}
def miss():
while laeuft[0]:
t = time.perf_counter()
try:
requests.get(f"{BASIS}/api/gesund", timeout=300)
messung["gesund"].append(time.perf_counter() - t)
except Exception: # noqa: BLE001
messung["gesund"].append(-1.0)
t = time.perf_counter()
try:
r = sitz.post(f"{BASIS}/api/killswitch/anhalten", json={"grund": "lastprobe"}, headers=KOPF, timeout=300)
messung["anhalten"].append((time.perf_counter() - t, r.status_code))
except Exception: # noqa: BLE001
messung["anhalten"].append((-1.0, 0))
time.sleep(0.5)
faden = threading.Thread(target=miss, daemon=True)
faden.start()
t0 = time.perf_counter()
with ThreadPoolExecutor(max_workers=40) as pool:
codes = list(pool.map(lambda _: requests.get(teuer, headers={"Cookie": cookie}, timeout=900).status_code, range(40)))
dauer = time.perf_counter() - t0
laeuft[0] = False
faden.join(timeout=310)
zeiten = [w for w, _ in messung["anhalten"]]
print(f" 40 Kerzenabrufe: {dauer:.1f} s, Status {sorted(set(codes))}")
print(f" anhalten waehrend der Last: n={len(zeiten)}, max {max(zeiten):.2f} s, Status {sorted({s for _, s in messung['anhalten']})}")
print(f" gesund waehrend der Last: n={len(messung['gesund'])}, max {max(messung['gesund']):.2f} s")
pruefe("Killswitch antwortet unter Last in unter 5 s (vorher gemessen: 118 s)", 0 <= max(zeiten) < 5.0)
print("== 5: ueberlanger Grund landet nicht im Register")
db = ABLAGE / "registry.db"
vorher = db.stat().st_size
a = sitz.post(f"{BASIS}/api/killswitch/anhalten", json={"grund": "G" * (1024 * 1024)}, headers=KOPF, timeout=120)
print(f" grund mit 1 MiB -> HTTP {a.status_code}, registry.db {vorher} -> {db.stat().st_size} Bytes")
pruefe("abgelehnt (413 oder 422), Register waechst nicht", a.status_code in (413, 422) and db.stat().st_size - vorher < 65536)
print("== 6: Passwortwechsel beendet die Sitzung")
pruefe(f"Sitzung gilt vor dem Wechsel (HTTP {sitz.get(f'{BASIS}/api/uebersicht', timeout=60).status_code})", True)
sys.path.insert(0, str(CODE))
os.chdir(CODE)
from trading.dashboard.zugang import Benutzer, lies_benutzer, schreibe_benutzer, setze_passwort # noqa: E402
bestand = lies_benutzer(ABLAGE / "zugang.toml")
bestand[BEN] = Benutzer(name=BEN, passwort=setze_passwort("ein-neues-passwort-2026"), totp_geheimnis=bestand[BEN].totp_geheimnis)
schreibe_benutzer(ABLAGE / "zugang.toml", bestand)
time.sleep(1.0)
danach = sitz.get(f"{BASIS}/api/uebersicht", timeout=60).status_code
pruefe(f"alte Sitzung nach dem Passwortwechsel: HTTP {danach}", danach == 401)
halt = sitz.post(f"{BASIS}/api/killswitch/anhalten", json={"grund": "nach-wechsel"}, headers=KOPF, timeout=60).status_code
pruefe(f"Killswitch mit der alten Sitzung: HTTP {halt}", halt == 401)
print(f"Speicher am Ende {speicher_mib(pid)}")
print(f"{len(befunde)} Befund(e): {befunde}")
sys.exit(2 if befunde else 0)
@@ -0,0 +1,74 @@
"""Probe-Backend fuer den Browserlauf zu Paket W: die echte App mit Broker-Attrappe, Wegwerf-Register und Wegwerf-Zugang.
Koordinator-Pruefcode, nicht fuers Repo. Wie .vault-tools/trading-plattform-pruefung/start_probe_zugang.py,
dazu die Konfiguration fuer Sicherheitsschluessel: RP-ID localhost, Origin http://localhost:<port>.
Gestartet wird aus einer KOPIE des Baums (dort gibt es keine Geheimnisdatei) -- kein Broker-Aufruf, kein
Ereignis im echten Register, keine echte Zugangsdatei, kein Melder.
Aufruf (WSL, aus der Kopie): python start_probe_w.py <port> <ausgabeordner>
"""
import dataclasses
import json
import sys
from pathlib import Path
sys.path.insert(0, str(Path.cwd()))
import uvicorn # noqa: E402
from trading.broker.attrappe import BrokerAttrappe # noqa: E402
from trading.dashboard.app import erzeuge_app # noqa: E402
from trading.dashboard.config import lade_dashboard_config # noqa: E402
from trading.dashboard.zugang import Benutzer # noqa: E402
from trading.dashboard.zugang import schreibe_benutzer # noqa: E402
from trading.dashboard.zugang import setze_passwort # noqa: E402
from trading.killswitch.killswitch import lade_killswitch_config # noqa: E402
from trading.registry.registry import Registry # noqa: E402
BENUTZER = "probe"
PASSWORT = "nur-fuer-die-probe-2026"
GEHEIMNIS = "JBSWY3DPEHPK3PXP"
port = int(sys.argv[1])
ordner = Path(sys.argv[2])
# Optional: ein Marktkatalog, der nur gelesen wird (fuer die Lastmessung der Kerzenabrufe).
katalog = sys.argv[3] if len(sys.argv) > 3 else None
ordner.mkdir(parents=True, exist_ok=True)
assert not Path("config/secrets.toml").exists(), "Die Probe startet nur aus einem Baum ohne Geheimnisdatei"
zugang_datei = ordner / "zugang.toml"
schreibe_benutzer(
zugang_datei,
{BENUTZER: Benutzer(name=BENUTZER, passwort=setze_passwort(PASSWORT), totp_geheimnis=GEHEIMNIS)},
)
(ordner / "zugang.json").write_text(
json.dumps({"benutzer": BENUTZER, "passwort": PASSWORT, "geheimnis": GEHEIMNIS}), encoding="utf-8"
)
(ordner / "paper").mkdir(exist_ok=True)
config = dataclasses.replace(
lade_dashboard_config("config/dashboard.toml"),
port=port,
broker_wahl="attrappe",
sendeintervall_s=2.0,
erlaubte_hosts=(f"127.0.0.1:{port}", f"localhost:{port}"),
anmeldung_pflicht=True,
mfa_pflicht=True,
zugang_datei=str(zugang_datei),
zugang_protokoll=str(ordner / "anmeldungen.jsonl"),
paper_status_datei=str(ordner / "status.json"),
paper_protokoll_ordner=str(ordner / "paper"),
schluessel_rp_id="localhost",
**({} if katalog is None else {"markt_katalog": katalog}),
schluessel_origins=(f"http://localhost:{port}",),
)
app = erzeuge_app(
broker=BrokerAttrappe(),
registry=Registry(db_pfad=ordner / "registry.db", artefakt_wurzel=ordner / "artifacts"),
config=config,
killswitch_config=lade_killswitch_config("config/killswitch.toml"),
dist=Path("dashboard/dist"),
)
print(f"Probe-Backend W auf 127.0.0.1:{port}, Ablage {ordner}", flush=True)
uvicorn.run(app, host="127.0.0.1", port=port, log_level="warning", server_header=False, proxy_headers=False)
@@ -36,6 +36,15 @@ GEHEIMNIS = "JBSWY3DPEHPK3PXP"
port = int(sys.argv[1])
ordner = Path(sys.argv[2])
# Befund der Sicherheitspruefung vom 2026-09-20: erzeuge_app startete (bis Paket S5) immer den echten
# Stille-Waechter und las dafuer config/paper.toml und config/secrets.toml aus dem Arbeitsordner. Eine
# Probe-Instanz im echten Projektordner haette nach 180 s auf das Telefon geklingelt. Deshalb nur aus
# einer Kopie oder einem Arbeitsbaum ohne Geheimnisdatei.
if Path("config/secrets.toml").exists():
sys.exit(
"Abbruch: Im Arbeitsordner liegt config/secrets.toml. Die Probe startet nur aus einer Kopie "
"ohne Geheimnisdatei, sonst spraeche sie mit dem echten ntfy und dem echten Home Assistant."
)
ordner.mkdir(parents=True, exist_ok=True)
mit_status = "mit-paperstatus" in sys.argv[3:]
zwei_benutzer = "zwei-benutzer" in sys.argv[3:]