- .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]>
124 lines
6.2 KiB
Python
124 lines
6.2 KiB
Python
"""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())
|