Pascal MichelandClaude Opus 5
2c9a4475fc
Erster Stand des öffentlichen Ausschnitts aus dem Vault. Die .gitignore arbeitet als Positivliste: ausgeschlossen ist alles, eingeschlossen nur, was ausdrücklich wieder aufgenommen wird. Notizen, Projekte, Finanzen, das Sitzungsgedächtnis und settings.local.json bleiben damit außen vor. Der Kundenname in AI/Regelwerk/seo.md ist durch eine neutrale Bezeichnung ersetzt; die Regeln selbst sind unverändert. Co-Authored-By: Claude Opus 5 <[email protected]>
395 lines
20 KiB
Python
395 lines
20 KiB
Python
"""Eigener Lauf mit echten Daten zu Teil 3b: Liste und Einzelansicht der Laeufe, Kerzen.
|
|
|
|
Koordinator-Pruefskript, nicht fuers Repo. Laeuft in WSL aus dem Projektordner (PYTHONPATH=.).
|
|
|
|
Geltungsbereich und Herkunft jeder Antwort:
|
|
F1 Die Liste nennt jede *.jsonl der konfigurierten Ordner genau einmal. Quelle: eigenes glob
|
|
F2 art und zeit je Eintrag gleich der ersten Zeile der Datei. Quelle: eigener Vollparse
|
|
F3 vollstaendig genau dann, wenn die letzte Zeile ergebnis/zusammenfassung. Projektvorgabe (Spec)
|
|
F4 Jede Zahl der Abschlusszeilen gleich dem eigenen Vollparse, ausser den
|
|
ausdruecklich weggelassenen Schluesseln beste_zelle, beste_zelle_je_gruppe. Projektvorgabe (Spec)
|
|
F5 Einzelansicht: Auslassungen (kurven, zeitstempel, startpunkte, training),
|
|
Zellen je Symbol und Zaehlungen je Symbol gleich eigener Zaehlung. Projektvorgabe (Spec)
|
|
F6 Kerzen: Zusammenfassung ab 09:30 New York gleich eigener pandas-Rechnung
|
|
aus demselben Rohabruf; Achsenwert = Berliner Wanduhr; Tagesstrings. Projektvorgabe (Spec)
|
|
Was es nicht sieht: die Oberflaeche, den WebSocket, die Skripte. Es ist ein Fangnetz, kein Nachweis.
|
|
Regressionsnachweis: --sabotage verfaelscht je einen Wert der Antwort vor dem Vergleich; der Lauf
|
|
muss dann mit Exit 2 und der passenden Meldung enden.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import math
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import traceback
|
|
from pathlib import Path
|
|
|
|
BEFUNDE: list[str] = []
|
|
|
|
|
|
def befund(text: str) -> None:
|
|
BEFUNDE.append(text)
|
|
print("BEFUND:", text)
|
|
|
|
|
|
def gleich(a, b) -> bool:
|
|
# FastAPI 0.141.1 serialisiert ueber pydantic und macht aus NaN und Infinity ein null
|
|
# (gemessen 2026-09-17); der eigene Vollparse liefert dafuer float('nan') oder inf.
|
|
if a is None and isinstance(b, float) and not math.isfinite(b):
|
|
return True
|
|
if b is None and isinstance(a, float) and not math.isfinite(a):
|
|
return True
|
|
if isinstance(a, float) and isinstance(b, float):
|
|
if math.isnan(a) and math.isnan(b):
|
|
return True
|
|
return a == b
|
|
if isinstance(a, dict) and isinstance(b, dict):
|
|
return a.keys() == b.keys() and all(gleich(a[k], b[k]) for k in a)
|
|
if isinstance(a, list) and isinstance(b, list):
|
|
return len(a) == len(b) and all(gleich(x, y) for x, y in zip(a, b))
|
|
return a == b
|
|
|
|
|
|
def vollparse(pfad: Path) -> list[dict]:
|
|
zeilen = []
|
|
with pfad.open("rb") as strom:
|
|
for roh in strom:
|
|
roh = roh.rstrip(b"\n")
|
|
if roh.strip():
|
|
zeilen.append(json.loads(roh.decode("utf-8")))
|
|
return zeilen
|
|
|
|
|
|
def baue_client():
|
|
from fastapi.testclient import TestClient
|
|
|
|
from trading.broker.attrappe import BrokerAttrappe
|
|
from trading.dashboard.app import erzeuge_app
|
|
from trading.dashboard.config import lade_dashboard_config
|
|
from trading.killswitch.killswitch import lade_killswitch_config
|
|
from trading.registry.registry import Registry
|
|
|
|
config = lade_dashboard_config("config/dashboard.toml")
|
|
temp = Path(tempfile.mkdtemp(prefix="pruef3b-"))
|
|
registry = Registry(db_pfad=temp / "registry.db", artefakt_wurzel=temp / "artifacts")
|
|
app = erzeuge_app(
|
|
broker=BrokerAttrappe(),
|
|
registry=registry,
|
|
config=config,
|
|
killswitch_config=lade_killswitch_config("config/killswitch.toml"),
|
|
dist=Path("dashboard/dist"),
|
|
)
|
|
return TestClient(app), config
|
|
|
|
|
|
ABSCHLUSS = {
|
|
"kampagne": ("walk_forward", "ergebnis"),
|
|
"signal": ("zusammenfassung",),
|
|
"signal_tage": ("zusammenfassung",),
|
|
}
|
|
LETZTE = {"kampagne": "ergebnis", "signal": "zusammenfassung", "signal_tage": "zusammenfassung"}
|
|
WEGGELASSEN = ("beste_zelle", "beste_zelle_je_gruppe")
|
|
|
|
|
|
def pruefe_laeufe(client, config, sabotage: str | None) -> None:
|
|
t0 = time.perf_counter()
|
|
antwort = client.get("/api/laeufe")
|
|
dauer = time.perf_counter() - t0
|
|
if antwort.status_code != 200:
|
|
befund(f"/api/laeufe antwortet {antwort.status_code}")
|
|
return
|
|
daten = antwort.json()
|
|
liste = daten["laeufe"]
|
|
print(f"/api/laeufe: {len(liste)} Eintraege in {dauer:.3f} s, hinweise={daten.get('hinweise')}")
|
|
|
|
if sabotage == "zahl":
|
|
for eintrag in liste:
|
|
if eintrag["art"] == "kampagne" and eintrag["ergebnis"]:
|
|
eintrag["ergebnis"]["walk_forward"]["sharpe"] += 1e-9
|
|
break
|
|
if sabotage == "fehlt":
|
|
liste.pop()
|
|
if sabotage == "art":
|
|
liste[0]["art"] = "signal" if liste[0]["art"] != "signal" else "kampagne"
|
|
|
|
dateien = {}
|
|
for ordner in config.laeufe_ordner:
|
|
for pfad in sorted(Path(ordner).glob("*.jsonl")):
|
|
dateien[(Path(ordner).name, pfad.name)] = pfad
|
|
gelistet = [(e["quelle"], e["datei"]) for e in liste]
|
|
if sorted(gelistet) != sorted(dateien):
|
|
befund(f"F1: Liste und Ordner weichen ab: nur Liste {set(gelistet) - set(dateien)}, nur Ordner {set(dateien) - set(gelistet)}")
|
|
if len(gelistet) != len(set(gelistet)):
|
|
befund("F1: doppelte Eintraege in der Liste")
|
|
|
|
zeiten = [e["zeit"] for e in liste if e["zeit"] is not None]
|
|
if zeiten != sorted(zeiten, reverse=True):
|
|
befund("Sortierung: zeit ist nicht absteigend")
|
|
|
|
arten: dict[str, int] = {}
|
|
for eintrag in liste:
|
|
schluessel = (eintrag["quelle"], eintrag["datei"])
|
|
pfad = dateien.get(schluessel)
|
|
if pfad is None:
|
|
continue
|
|
zeilen = vollparse(pfad)
|
|
kopf = zeilen[0]
|
|
arten[eintrag["art"]] = arten.get(eintrag["art"], 0) + 1
|
|
if eintrag["art"] != kopf.get("art"):
|
|
befund(f"F2 {pfad.name}: art {eintrag['art']!r} gegen {kopf.get('art')!r}")
|
|
continue
|
|
if eintrag["zeit"] != kopf.get("zeit"):
|
|
befund(f"F2 {pfad.name}: zeit {eintrag['zeit']!r} gegen {kopf.get('zeit')!r}")
|
|
if eintrag["groesse_bytes"] != pfad.stat().st_size:
|
|
befund(f"{pfad.name}: groesse_bytes {eintrag['groesse_bytes']} gegen {pfad.stat().st_size}")
|
|
art = kopf.get("art")
|
|
erwartet_voll = zeilen[-1].get("art") == LETZTE.get(art)
|
|
if eintrag["vollstaendig"] != erwartet_voll:
|
|
befund(f"F3 {pfad.name}: vollstaendig {eintrag['vollstaendig']} gegen {erwartet_voll}")
|
|
for name in ABSCHLUSS.get(art, ()):
|
|
eigene = [z for z in zeilen if z.get("art") == name]
|
|
if not eigene:
|
|
continue
|
|
soll = {k: v for k, v in eigene[-1].items() if k not in WEGGELASSEN}
|
|
ist = (eintrag["ergebnis"] or {}).get(name)
|
|
if ist is None:
|
|
befund(f"F4 {pfad.name}: Abschlusszeile {name} fehlt in der Liste")
|
|
elif not gleich(ist, soll):
|
|
abweichend = [k for k in soll if k not in ist or not gleich(ist[k], soll[k])]
|
|
zusaetzlich = [k for k in ist if k not in soll]
|
|
befund(f"F4 {pfad.name}: {name} weicht ab in {abweichend}, zusaetzlich {zusaetzlich}")
|
|
kopf_ist = eintrag["kopf"]
|
|
kopf_soll = json.loads(json.dumps(kopf))
|
|
if art == "signal_tage":
|
|
kopf_soll.get("daten", {}).pop("instrumente", None)
|
|
if not gleich(kopf_ist, kopf_soll):
|
|
befund(f"{pfad.name}: kopf weicht ab in {[k for k in kopf_soll if not gleich(kopf_ist.get(k), kopf_soll[k])]}")
|
|
print("Arten:", arten)
|
|
|
|
|
|
def pruefe_einzelansicht(client, config, sabotage: str | None) -> None:
|
|
kandidaten = [
|
|
("kampagne", "laeufe", "aapl-minute-ppo-v20260914-132555.jsonl", None),
|
|
# Traegt 30 Zeilen mit "gewinnfaktor": NaN (Protokollschreiber vor dem Fix vom 2026-09-14).
|
|
("kampagne", "laeufe", "aapl-minute-ppo-v20260914-085048.jsonl", None),
|
|
("signal", "signal", "signal-20260914-213950.jsonl", None),
|
|
("signal_tage", "signal", "tage-teil9-20260917-030800.jsonl", "MSFT"),
|
|
]
|
|
ordner = {Path(o).name: Path(o) for o in config.laeufe_ordner}
|
|
for art, quelle, datei, symbol in kandidaten:
|
|
pfad = ordner[quelle] / datei
|
|
if not pfad.exists():
|
|
befund(f"Voraussetzung fehlt: {pfad}")
|
|
continue
|
|
adresse = f"/api/laeufe/{quelle}/{datei}" + (f"?symbol={symbol}" if symbol else "")
|
|
t0 = time.perf_counter()
|
|
antwort = client.get(adresse)
|
|
dauer = time.perf_counter() - t0
|
|
if antwort.status_code != 200:
|
|
befund(f"{adresse} antwortet {antwort.status_code}")
|
|
continue
|
|
daten = antwort.json()
|
|
print(f"{adresse}: {len(antwort.content):,} Bytes in {dauer:.3f} s, Zeilenarten { {k: len(v) for k, v in daten['zeilen'].items()} }, unlesbar {daten.get('unlesbare_zeilen')}")
|
|
zeilen = vollparse(pfad)
|
|
if daten["art"] != art:
|
|
befund(f"F5 {datei}: art {daten['art']!r}")
|
|
if not gleich(daten["kopf"], zeilen[0]):
|
|
befund(f"F5 {datei}: kopf der Einzelansicht ist nicht die ganze Kopfzeile")
|
|
if sabotage == "zelle" and art == "signal_tage":
|
|
daten["zeilen"]["zelle_instrument"].pop()
|
|
if sabotage == "zaehlung" and art == "signal_tage":
|
|
daten["symbole"][0]["k1"] += 1
|
|
|
|
if art == "kampagne":
|
|
if "training" in daten["zeilen"] and daten["zeilen"]["training"]:
|
|
befund("F5: training-Zeilen sind in der Einzelansicht")
|
|
eigene = [z for z in zeilen if z.get("art") == "fenster"]
|
|
ist = daten["zeilen"].get("fenster", [])
|
|
if len(ist) != len(eigene):
|
|
befund(f"F5: {len(ist)} fenster gegen {len(eigene)}")
|
|
for a, b in zip(ist, eigene):
|
|
soll = {k: v for k, v in b.items() if k not in ("kurven", "zeitstempel", "startpunkte")}
|
|
if not gleich(a, soll):
|
|
befund(f"F5: fenster {b.get('nummer')} weicht ab oder traegt Masse: {sorted(set(a) ^ set(soll))}")
|
|
break
|
|
for name in ("walk_forward", "ergebnis"):
|
|
if not gleich(daten["zeilen"].get(name, []), [z for z in zeilen if z.get("art") == name]):
|
|
befund(f"F5: {name} weicht ab")
|
|
if art == "signal":
|
|
for name in ("zelle", "permutation", "zusammenfassung"):
|
|
if not gleich(daten["zeilen"].get(name, []), [z for z in zeilen if z.get("art") == name]):
|
|
befund(f"F5 signal: {name} weicht ab")
|
|
if art == "signal_tage":
|
|
for name in ("diagnose", "zelle_pool", "jahr", "permutation", "zusammenfassung"):
|
|
if not gleich(daten["zeilen"].get(name, []), [z for z in zeilen if z.get("art") == name]):
|
|
befund(f"F5 signal_tage: {name} weicht ab")
|
|
eigene = [z for z in zeilen if z.get("art") == "zelle_instrument" and z.get("symbol") == symbol]
|
|
if not gleich(daten["zeilen"].get("zelle_instrument", []), eigene):
|
|
befund(f"F5 signal_tage: Zellen fuer {symbol}: {len(daten['zeilen'].get('zelle_instrument', []))} gegen {len(eigene)}")
|
|
print(f" Zellen {symbol}: {len(eigene)}")
|
|
zaehlung: dict[str, dict] = {}
|
|
for z in zeilen:
|
|
if z.get("art") != "zelle_instrument":
|
|
continue
|
|
s = zaehlung.setdefault(z["symbol"], {"symbol": z["symbol"], "gruppe": z.get("gruppe"), "zellen": 0, "t_ab_2": 0, "k1": 0, "k2": 0, "signale": 0})
|
|
s["zellen"] += 1
|
|
t = z.get("t_ic_massgeblich")
|
|
if isinstance(t, (int, float)) and not math.isnan(t) and abs(t) >= 2:
|
|
s["t_ab_2"] += 1
|
|
s["k1"] += 1 if z.get("K1") is True else 0
|
|
s["k2"] += 1 if z.get("K2") is True else 0
|
|
s["signale"] += 1 if z.get("signal") is True else 0
|
|
ist = {s["symbol"]: s for s in daten.get("symbole", [])}
|
|
if ist.keys() != zaehlung.keys():
|
|
befund(f"F5 symbole: Menge weicht ab: {set(ist) ^ set(zaehlung)}")
|
|
for name, soll in zaehlung.items():
|
|
if name in ist and not gleich({k: ist[name].get(k) for k in soll}, soll):
|
|
befund(f"F5 symbole {name}: {ist[name]} gegen {soll}")
|
|
print(f" Symbole: {len(zaehlung)}, Summe Signale {sum(s['signale'] for s in zaehlung.values())}, Summe K1 {sum(s['k1'] for s in zaehlung.values())}")
|
|
ohne = client.get(f"/api/laeufe/{quelle}/{datei}").json()
|
|
if ohne["zeilen"].get("zelle_instrument"):
|
|
befund("F5: ohne symbol kommen Instrumentenzellen")
|
|
|
|
for adresse in ("/api/laeufe/signal/gibt-es-nicht.jsonl", "/api/laeufe/fremd/signal-20260914-213950.jsonl", "/api/laeufe/signal/..%2F..%2Fconfig%2Fsecrets.toml"):
|
|
status = client.get(adresse).status_code
|
|
print(f"{adresse}: {status}")
|
|
if status != 404:
|
|
befund(f"Pfadsicherheit: {adresse} antwortet {status} statt 404")
|
|
|
|
|
|
def pruefe_kerzen(client, config, sabotage: str | None) -> None:
|
|
import numpy as np
|
|
import pandas as pd
|
|
|
|
from trading.data.catalog import Catalog
|
|
|
|
antwort = client.get("/api/markt/datensaetze")
|
|
if antwort.status_code != 200:
|
|
befund(f"/api/markt/datensaetze antwortet {antwort.status_code}")
|
|
return
|
|
saetze = antwort.json()["datensaetze"]
|
|
print(f"/api/markt/datensaetze: {len(saetze)} Datensaetze")
|
|
katalog = Catalog(config.markt_katalog)
|
|
if sorted(s["bar_type"] for s in saetze) != katalog.bar_types():
|
|
befund("F6: Datensaetze weichen von Catalog.bar_types() ab")
|
|
minute = [s for s in saetze if s["bar_type"] == "AAPL.XNAS-1-MINUTE-LAST-EXTERNAL"]
|
|
if not minute or minute[0].get("aufloesungen_min") != [1, 5, 15, 60]:
|
|
befund(f"F6: aufloesungen_min des Minutendatensatzes: {minute[0].get('aufloesungen_min') if minute else 'fehlt'}")
|
|
|
|
tag = "MSFT.XNAS-1-DAY-LAST-EXTERNAL"
|
|
t0 = time.perf_counter()
|
|
antwort = client.get("/api/markt/kerzen", params={"bar_type": tag, "aufloesung_min": 1, "anzahl": 3000})
|
|
dauer = time.perf_counter() - t0
|
|
if antwort.status_code != 200:
|
|
befund(f"Tageskerzen antworten {antwort.status_code}: {antwort.text[:300]}")
|
|
else:
|
|
kerzen = antwort.json()["kerzen"]
|
|
roh = katalog.read(tag)
|
|
print(f"Tageskerzen {tag}: {len(kerzen)} in {dauer:.3f} s (Katalog {len(roh)})")
|
|
if len(kerzen) != len(roh):
|
|
befund(f"F6 Tage: {len(kerzen)} gegen {len(roh)}")
|
|
for k, b in list(zip(kerzen, roh))[:: max(1, len(roh) // 200)]:
|
|
soll_tag = pd.Timestamp(b.ts_event, unit="ns", tz="UTC").tz_convert("America/New_York").strftime("%Y-%m-%d")
|
|
if k["achse"] != soll_tag or not gleich([k["open"], k["high"], k["low"], k["close"], k["volumen"]], [float(b.open), float(b.high), float(b.low), float(b.close), float(b.volume)]):
|
|
befund(f"F6 Tage: Kerze {k} gegen {b}")
|
|
break
|
|
|
|
minute_typ = "AAPL.XNAS-1-MINUTE-LAST-EXTERNAL"
|
|
for aufloesung, ende, anzahl in [(1, "2026-09-10", 500), (5, "2026-09-10", 500), (15, "2026-03-13", 400), (60, "2025-11-05", 300), (60, "2026-09-10", 3000)]:
|
|
t0 = time.perf_counter()
|
|
antwort = client.get("/api/markt/kerzen", params={"bar_type": minute_typ, "aufloesung_min": aufloesung, "ende": ende, "anzahl": anzahl})
|
|
dauer = time.perf_counter() - t0
|
|
if antwort.status_code != 200:
|
|
befund(f"Minuten {aufloesung}/{ende}: {antwort.status_code} {antwort.text[:300]}")
|
|
continue
|
|
kerzen = antwort.json()["kerzen"]
|
|
if sabotage == "kerze" and aufloesung == 5:
|
|
kerzen[3]["high"] += 0.01
|
|
if sabotage == "achse" and aufloesung == 15:
|
|
kerzen[5]["achse"] += 3600
|
|
|
|
bis = pd.Timestamp(ende, tz="UTC") + pd.Timedelta(days=1)
|
|
von = bis - pd.Timedelta(days=int(anzahl * aufloesung / 390 * 1.6) + 30)
|
|
roh = katalog.read(minute_typ, von, bis)
|
|
rahmen = pd.DataFrame(
|
|
{
|
|
"ts": [b.ts_event for b in roh],
|
|
"open": [float(b.open) for b in roh],
|
|
"high": [float(b.high) for b in roh],
|
|
"low": [float(b.low) for b in roh],
|
|
"close": [float(b.close) for b in roh],
|
|
"volumen": [float(b.volume) for b in roh],
|
|
}
|
|
)
|
|
ny = pd.to_datetime(rahmen["ts"], unit="ns", utc=True).dt.tz_convert("America/New_York")
|
|
rahmen = rahmen[ny.dt.strftime("%Y-%m-%d") <= ende].copy()
|
|
ny = ny[rahmen.index]
|
|
minuten = ny.dt.hour * 60 + ny.dt.minute - (9 * 60 + 30)
|
|
rahmen["tag"] = ny.dt.strftime("%Y-%m-%d")
|
|
rahmen["eimer"] = (minuten // aufloesung).astype(int)
|
|
gruppen = rahmen.groupby(["tag", "eimer"], sort=True)
|
|
eigen = gruppen.agg(open=("open", "first"), high=("high", "max"), low=("low", "min"), close=("close", "last"), volumen=("volumen", "sum")).reset_index()
|
|
anfang = pd.to_datetime(eigen["tag"]).dt.tz_localize("America/New_York") + pd.to_timedelta(9 * 60 + 30 + eigen["eimer"] * aufloesung, unit="m")
|
|
eigen["zeit_utc"] = anfang.dt.tz_convert("UTC")
|
|
berlin = anfang.dt.tz_convert("Europe/Berlin").dt.tz_localize(None)
|
|
eigen["achse"] = (berlin - pd.Timestamp("1970-01-01")) // pd.Timedelta(seconds=1)
|
|
eigen = eigen.tail(anzahl).reset_index(drop=True)
|
|
print(f"Minuten {aufloesung:>2} min bis {ende}: {len(kerzen)} Kerzen in {dauer:.3f} s (eigene Rechnung {len(eigen)}, Rohkerzen {len(roh)})")
|
|
if len(kerzen) != len(eigen):
|
|
befund(f"F6 {aufloesung} min bis {ende}: {len(kerzen)} Kerzen gegen {len(eigen)}")
|
|
continue
|
|
for i, k in enumerate(kerzen):
|
|
e = eigen.iloc[i]
|
|
zeit_ist = pd.Timestamp(k["zeit"])
|
|
werte_ist = np.array([k["open"], k["high"], k["low"], k["close"], k["volumen"]], dtype=float)
|
|
werte_soll = np.array([e["open"], e["high"], e["low"], e["close"], e["volumen"]], dtype=float)
|
|
if zeit_ist != e["zeit_utc"] or int(k["achse"]) != int(e["achse"]) or not np.allclose(werte_ist, werte_soll, rtol=0, atol=1e-9):
|
|
befund(f"F6 {aufloesung} min bis {ende}, Kerze {i}: ist {k} soll zeit {e['zeit_utc']} achse {int(e['achse'])} werte {werte_soll.tolist()}")
|
|
break
|
|
achsen = [k["achse"] for k in kerzen]
|
|
if achsen != sorted(set(achsen)):
|
|
befund(f"F6 {aufloesung} min bis {ende}: Achsenwerte nicht streng steigend")
|
|
|
|
for params, erwartet in [
|
|
({"bar_type": "GIBT.ES-NICHT", "aufloesung_min": 1, "anzahl": 10}, 404),
|
|
({"bar_type": minute_typ, "aufloesung_min": 7, "anzahl": 10}, 422),
|
|
({"bar_type": minute_typ, "aufloesung_min": 5, "anzahl": 0}, 422),
|
|
({"bar_type": minute_typ, "aufloesung_min": 5, "anzahl": 999999}, 422),
|
|
({"bar_type": minute_typ, "aufloesung_min": 5, "anzahl": 10, "ende": "kein-datum"}, 422),
|
|
({"bar_type": tag, "aufloesung_min": 5, "anzahl": 10}, 422),
|
|
]:
|
|
status = client.get("/api/markt/kerzen", params=params).status_code
|
|
print(f"{params}: {status}")
|
|
if status != erwartet:
|
|
befund(f"F6 Fehlerfall {params}: {status} statt {erwartet}")
|
|
|
|
|
|
def main() -> int:
|
|
zerleger = argparse.ArgumentParser()
|
|
zerleger.add_argument("teil", choices=["laeufe", "einzel", "kerzen", "alles"])
|
|
zerleger.add_argument("--sabotage", choices=["zahl", "fehlt", "art", "zelle", "zaehlung", "kerze", "achse"])
|
|
argumente = zerleger.parse_args()
|
|
try:
|
|
client, config = baue_client()
|
|
with client:
|
|
if argumente.teil in ("laeufe", "alles"):
|
|
pruefe_laeufe(client, config, argumente.sabotage)
|
|
if argumente.teil in ("einzel", "alles"):
|
|
pruefe_einzelansicht(client, config, argumente.sabotage)
|
|
if argumente.teil in ("kerzen", "alles"):
|
|
pruefe_kerzen(client, config, argumente.sabotage)
|
|
except Exception:
|
|
befund("unerwarteter Abbruch:\n" + traceback.format_exc())
|
|
print()
|
|
if BEFUNDE:
|
|
print(f"{len(BEFUNDE)} Befund(e).")
|
|
return 2
|
|
print("Kein Befund.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|