Pascal MichelandClaude Opus 5
593acdd122
- .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]>
510 lines
30 KiB
Python
510 lines
30 KiB
Python
"""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)
|