Nexhorizons-Public/claude-regelwerk
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Trading-Plattform-Pruefung: Abendkontrolle prueft die Sicherung des vorigen Handelstags (Paket E9, Task 726)

Browse Source 
abendkontrolle_wsl.py nimmt --sicherung (Stand der Ablage vom Host) und meldet per ntfy, wenn die
Sicherung des vorigen Handelstags fehlt, kein Archiv hat, nicht bestaetigt ist oder die Bestaetigung
nicht passt; vor erster-tag still, ohne erster-tag ein Befund. abendkontrolle_angriffe.py mit
--nur-sicherung (60 Faelle), abendkontrolle_mutanten.py mit --nur und 30 Sicherungs-Mutanten.
Gegengeprueft: byte-gleich mit der Plankopie, Lauf gegen echte Archive ohne Abweichung.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
Claude-Session: https://claude.ai/code/session_0126RDfXiECscXipzf67uypJ
This commit is contained in:
Pascal MichelandClaude Opus 5.5 committed 2026-10-11 17:09:14 +02:00
1 parent c5db92c6de
commit 3e55bb915d
3 files changed
+514 -3
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.vault-tools/trading-plattform-pruefung/abendkontrolle_angriffe.py
+258
View File
@@ -5,6 +5,7 @@ Grundlinie; ohne Netz (--ohne-broker/--ohne-kerzen, Kalender und Journal aus Dat
Auswertung fuer die Brokerzeile) und ohne Meldung (--ohne-meldung). Faelle an zwei Listenpositionen,
wo die Reihenfolge zaehlen kann (AI/Regelwerk/pruefwerkzeuge.md, Abschnitt 3).
Aufruf in WSL (Venv) und unter Windows: python abendkontrolle_angriffe.py
Nur die Faelle zur Sicherung (Paket E9, Task 726), in WSL: python abendkontrolle_angriffe.py --nur-sicherung
Exit 0, wenn jeder Fall so ausgeht wie erwartet; sonst 1 mit der Liste der Abweichungen.
"""
@@ -811,12 +812,269 @@ def ibkr_windows_faelle() -> None:
f"ibkr {ibkr} alpaca {alpaca_tage}")
# --- Sicherung des vorigen Handelstags (Paket E9, Task 726), WSL-Teil ------------------------------
# TAG ist ein Mittwoch, S_VORHER der Dienstag davor, S_DAVOR der Montag. Die Summen sind erfunden.
S_VORHER = "2026-09-22"
S_DAVOR = "2026-09-21"
S_ERSTER = "2026-09-14"
S_SUMME = "ab" * 32
S_ANDERE = "cd" * 32
S_KALENDER = [
{"date": S_DAVOR, "open": "09:30", "close": "16:00"},
{"date": S_VORHER, "open": "09:30", "close": "16:00"},
] + KALENDER_TAG
def s_eintraege(anders: dict | None = None, ohne: tuple = ()) -> dict:
"""Beide Instanzen an S_DAVOR und S_VORHER mit Archiv und bestaetigt; anders ersetzt, ohne loescht."""
eintraege = {
f"{tag}-{instanz}": {"summe": S_SUMME, "bestaetigt": S_SUMME, "archiv": True}
for tag in (S_DAVOR, S_VORHER)
for instanz in ("alpaca", "ibkr")
}
for name in ohne:
del eintraege[name]
eintraege.update(anders or {})
return eintraege
def s_stand(eintraege: dict | None = None, **oben) -> dict:
"""Ein Stand wie von abendkontrolle-host.sh; oben ersetzt oder ergaenzt Schluessel der obersten Ebene."""
stand = {"erster_tag": S_ERSTER, "eintraege": s_eintraege() if eintraege is None else eintraege}
stand.update(oben)
return stand
def fahre_sicherung(wurzel: Path, ziel: Path, teil: str, *, schalter: bool = True, tag: str = TAG,
meldung: bool = False, env: dict | None = None) -> dict:
"""Ein Lauf eines Teils mit --sicherung; ziel ist der Projektordner (alpaca) oder der Betriebsordner (ibkr)."""
befehl = [
sys.executable, str(HIER / "abendkontrolle_wsl.py"), "--tag", tag,
"--kalender", str(wurzel / "kalender.json"), "--journal", str(wurzel / "journal.txt"),
"--dienste", str(wurzel / "dienste.json"),
]
if teil == "ibkr":
befehl += ["--teil", "ibkr", "--ibkr-ordner", str(ziel), "--projekt", str(wurzel),
"--ibkr-konto", str(wurzel / "konto.json"), "--jetzt", NACH_SCHLUSS]
else:
befehl += ["--projekt", str(ziel), "--ohne-kerzen", "--ohne-broker"]
if schalter:
befehl += ["--sicherung", str(wurzel / "sicherung.json")]
if not meldung:
befehl.append("--ohne-meldung")
umgebung = {k: v for k, v in os.environ.items() if k != "ABENDKONTROLLE_AUSWERTUNG"}
umgebung.update(env or {})
lauf = subprocess.run(befehl, capture_output=True, text=True, env=umgebung, timeout=300)
try:
ergebnis = json.loads(lauf.stdout.strip().splitlines()[-1])
except (IndexError, json.JSONDecodeError):
return {"befunde": [f"KEINE JSON-AUSGABE exit {lauf.returncode}: {lauf.stderr[-300:]}"], "exit": lauf.returncode}
ergebnis["_exit"] = lauf.returncode
return ergebnis
def sicherung_faelle() -> None:
def fall(name, stand, *, befund, teil="alpaca", kalender=S_KALENDER, anzahl=None, exit_code=0,
schalter=True, tag=TAG, auch=(), nicht=(), fakt=None):
"""stand: None heisst, die Datei fehlt; eine Zeichenkette wird roh geschrieben, sonst als JSON."""
with tempfile.TemporaryDirectory() as tmp:
wurzel = Path(tmp)
ziel = baue_ibkr(wurzel) if teil == "ibkr" else baue_projekt(wurzel)
(wurzel / "kalender.json").write_text(json.dumps(kalender), encoding="utf-8")
if isinstance(stand, str):
(wurzel / "sicherung.json").write_text(stand, encoding="utf-8")
elif stand is not None:
(wurzel / "sicherung.json").write_text(json.dumps(stand), encoding="utf-8")
erwarte("Sicherung " + name, fahre_sicherung(wurzel, ziel, teil, schalter=schalter, tag=tag),
befund=befund, anzahl=anzahl, exit_code=exit_code, auch=auch, nicht=nicht, fakt=fakt)
a = f"{S_VORHER}-alpaca"
i = f"{S_VORHER}-ibkr"
offen = {"summe": S_SUMME, "bestaetigt": None, "archiv": True}
# Grundlinie in beiden Teilen; ohne Schalter bleibt das Werkzeug, wie es war.
fall("Grundlinie alpaca", s_stand(), befund=None, anzahl=0,
fakt=lambda f, e: f.get("sicherung") == {"name": a, "erster_tag": S_ERSTER, "geprueft": True})
fall("Grundlinie ibkr", s_stand(), teil="ibkr", befund=None, anzahl=0,
fakt=lambda f, e: f.get("sicherung") == {"name": i, "erster_tag": S_ERSTER, "geprueft": True})
fall("ohne Schalter", s_stand(s_eintraege({a: offen})), schalter=False, befund=None, anzahl=0,
fakt=lambda f, e: "sicherung" not in f)
# Umgebung und Form der Datei
fall("Datei fehlt", None, befund="Sicherung nicht geprueft (FileNotFoundError", anzahl=1)
fall("Datei kein JSON", "<html>", befund="Sicherung nicht geprueft (JSONDecodeError", anzahl=1)
fall("Datei leer", "", befund="Sicherung nicht geprueft (JSONDecodeError", anzahl=1)
fall("Stand null", "null", befund="Sicherung: Stand ist kein Objekt", anzahl=1)
fall("Stand als Liste", [], befund="Sicherung: Stand ist kein Objekt", anzahl=1)
fall("eintraege als Liste", s_stand([]), befund="Sicherung: Stand in unerwarteter Form", anzahl=1)
fall("eintraege fehlt", {"erster_tag": S_ERSTER}, befund="Sicherung: Stand in unerwarteter Form", anzahl=1)
fall("erster_tag fehlt", {"eintraege": s_eintraege()}, befund="Sicherung: Stand in unerwarteter Form", anzahl=1)
fall("erster_tag als Zahl", s_stand(erster_tag=20260914), befund="Sicherung: erster_tag ist kein Datum", anzahl=1)
fall("erster_tag kein Datum", s_stand(erster_tag="morgen"), befund="Sicherung: erster_tag ist kein Datum", anzahl=1)
# Fehler, die abendkontrolle-host.sh in den Stand schreibt
fall("Host meldet fehler", {"fehler": "Ablage der Sicherung fehlt"},
befund="Sicherung nicht geprueft: Ablage der Sicherung fehlt", anzahl=1)
fall("fehler neben vollem Stand", s_stand(fehler="Ablage der Sicherung nicht lesbar"),
befund="Sicherung nicht geprueft: Ablage der Sicherung nicht lesbar", anzahl=1)
# Erster Sicherungstag (seit der Plankritik Runde 1): null und ein Tag nach TAG sind Befunde, sonst
# schwiege die Pruefung auf Dauer. Still bleibt nur ein voriger Handelstag vor erster_tag; TAG selbst
# als erster_tag ist erlaubt, und der erste Sicherungstag selbst wird geprueft.
fall("erster_tag null", s_stand({}, erster_tag=None),
befund="Sicherung nicht geprueft: auf dem Docker-Host fehlt erster-tag der Sicherung", anzahl=1,
fakt=lambda f, e: "sicherung" not in f
and not any("erster-tag" in h for h in e.get("hinweise", [])))
fall("erster_tag nach dem geprueften Tag", s_stand({}, erster_tag="2026-09-24"),
befund=f"Sicherung nicht geprueft: erster_tag 2026-09-24 liegt nach dem geprueften Tag {TAG}", anzahl=1,
fakt=lambda f, e: "sicherung" not in f)
fall("voriger Handelstag vor erster_tag", s_stand({}, erster_tag=TAG), befund=None, anzahl=0,
fakt=lambda f, e: f.get("sicherung", {}).get("geprueft") is False)
fall("erster_tag ist der vorige Handelstag", s_stand({}, erster_tag=S_VORHER),
befund=f"Sicherung {a} fehlt auf dem Docker-Host", anzahl=1)
# Eintrag des vorigen Handelstags
fall("Eintrag fehlt", s_stand(s_eintraege(ohne=(a,))), befund=f"Sicherung {a} fehlt auf dem Docker-Host", anzahl=1)
fall("Eintrag null", s_stand(s_eintraege({a: None})), befund=f"Sicherung {a} fehlt auf dem Docker-Host", anzahl=1)
fall("Eintrag als Liste", s_stand(s_eintraege({a: [S_SUMME, S_SUMME]})),
befund=f"Sicherung {a} fehlt auf dem Docker-Host", anzahl=1)
# summe
for titel, eintrag in (
("summe null", {"summe": None, "bestaetigt": S_SUMME}),
("summe fehlt", {"bestaetigt": S_SUMME}),
("summe leer", {"summe": "", "bestaetigt": S_SUMME}),
("summe in Grossbuchstaben", {"summe": S_SUMME.upper(), "bestaetigt": S_SUMME.upper()}),
("summe mit 63 Zeichen", {"summe": S_SUMME[:-1], "bestaetigt": S_SUMME[:-1]}),
("summe mit 65 Zeichen", {"summe": S_SUMME + "a", "bestaetigt": S_SUMME + "a"}),
("summe als Zahl", {"summe": 5, "bestaetigt": S_SUMME}),
):
fall(titel, s_stand(s_eintraege({a: eintrag})), befund=f"Sicherung {a}: keine gueltige Summe", anzahl=1)
# bestaetigt
for titel, eintrag, text in (
("bestaetigt null", {"summe": S_SUMME, "bestaetigt": None}, "nicht bestaetigt"),
("bestaetigt fehlt", {"summe": S_SUMME}, "nicht bestaetigt"),
("bestaetigt leer", {"summe": S_SUMME, "bestaetigt": ""}, "nicht bestaetigt"),
("bestaetigt in Grossbuchstaben", {"summe": S_SUMME, "bestaetigt": S_SUMME.upper()}, "nicht bestaetigt"),
("bestaetigt erstes Zeichen anders", {"summe": S_SUMME, "bestaetigt": "0" + S_SUMME[1:]},
"die Bestaetigung passt nicht zur Summe"),
("bestaetigt letztes Zeichen anders", {"summe": S_SUMME, "bestaetigt": S_SUMME[:-1] + "0"},
"die Bestaetigung passt nicht zur Summe"),
("Bestaetigung einer anderen Sicherung", {"summe": S_SUMME, "bestaetigt": S_ANDERE},
"die Bestaetigung passt nicht zur Summe"),
):
fall(titel, s_stand(s_eintraege({a: {**eintrag, "archiv": True}})), befund=f"Sicherung {a}: {text}", anzahl=1)
# archiv (seit der Plankritik Runde 1): Bei gueltiger Summe zaehlt nur true, sonst fehlt das Archiv.
for titel, eintrag in (
("archiv false", {"summe": S_SUMME, "bestaetigt": S_SUMME, "archiv": False}),
("archiv fehlt", {"summe": S_SUMME, "bestaetigt": S_SUMME}),
("archiv null", {"summe": S_SUMME, "bestaetigt": S_SUMME, "archiv": None}),
("archiv als Text", {"summe": S_SUMME, "bestaetigt": S_SUMME, "archiv": "true"}),
("archiv false, nicht bestaetigt", {"summe": S_SUMME, "bestaetigt": None, "archiv": False}),
):
fall(titel, s_stand(s_eintraege({a: eintrag})), befund=f"Sicherung {a}: Archiv fehlt auf dem Docker-Host",
anzahl=1)
# Instanz nach --teil
fall("Teil ibkr, nur Alpaca bestaetigt", s_stand(s_eintraege({i: offen})), teil="ibkr",
befund=f"Sicherung {i}: nicht bestaetigt", anzahl=1, nicht=("alpaca",))
fall("Teil alpaca, nur IBKR bestaetigt", s_stand(s_eintraege({a: offen})),
befund=f"Sicherung {a}: nicht bestaetigt", anzahl=1, nicht=("ibkr",))
fall("Teil ibkr, Eintrag fehlt", s_stand(s_eintraege(ohne=(i,))), teil="ibkr",
befund=f"Sicherung {i} fehlt auf dem Docker-Host", anzahl=1)
# Reihenfolge: derselbe Schaden am ersten und am letzten Eintrag
andere = {name: wert for name, wert in s_eintraege().items() if name != a}
fall("Schaden am ersten Eintrag", s_stand({a: offen, **andere}), befund=f"Sicherung {a}: nicht bestaetigt", anzahl=1)
fall("Schaden am letzten Eintrag", s_stand({**andere, a: offen}), befund=f"Sicherung {a}: nicht bestaetigt", anzahl=1)
# Kalender: der juengste Handelstag vor TAG, nicht der Vortag und nicht der letzte Listeneintrag
fall("Kalender unsortiert", s_stand(s_eintraege({a: offen})), kalender=list(reversed(S_KALENDER)),
befund=f"Sicherung {a}: nicht bestaetigt", anzahl=1)
feiertag = [S_KALENDER[0]] + KALENDER_TAG
fall("Feiertag am Vortag, der Tag davor bestaetigt", s_stand(s_eintraege({a: offen})), kalender=feiertag,
befund=None, anzahl=0, fakt=lambda f, e: f.get("sicherung", {}).get("name") == f"{S_DAVOR}-alpaca")
fall("Feiertag am Vortag, der Tag davor offen", s_stand(s_eintraege({f"{S_DAVOR}-alpaca": offen})),
kalender=feiertag, befund=f"Sicherung {S_DAVOR}-alpaca: nicht bestaetigt", anzahl=1)
fall("Kalender ohne Tag davor", s_stand(), kalender=KALENDER_TAG, befund="Sicherung: kein Handelstag in den", anzahl=1)
fall("Kalender als Objekt", s_stand(), kalender={}, befund="Sicherung nicht geprueft (ValueError", anzahl=2,
auch=("Kalender in unerwarteter Form",))
fall("Kalender Eintrag ohne date, erste Position", s_stand(), kalender=[{"open": "09:30"}] + S_KALENDER,
befund="Sicherung nicht geprueft (ValueError", anzahl=2)
fall("Kalender Eintrag ohne date, letzte Position", s_stand(), kalender=S_KALENDER + [{"date": 5}],
befund="Sicherung nicht geprueft (ValueError", anzahl=2)
fall("Kalender mit Datum in anderer Form", s_stand(), kalender=[{"date": "22.09.2026"}] + S_KALENDER,
befund="Sicherung nicht geprueft (ValueError", anzahl=1)
# Fenster (seit der Plankritik Runde 1): SICHERUNG_FENSTER_TAGE Kalendertage vor TAG, der Rand gehoert
# dazu, TAG selbst nie. Das Werkzeug prueft keinen Wochentag; Rand und Tag davor sind Wochenendtage.
rand = "2026-09-13"
fall("Kalender voriger Handelstag am Rand des Fensters",
s_stand(s_eintraege({f"{rand}-alpaca": {"summe": S_SUMME, "bestaetigt": S_SUMME, "archiv": True}}),
erster_tag="2026-09-01"),
kalender=[{"date": rand, "open": "09:30", "close": "16:00"}] + KALENDER_TAG, befund=None, anzahl=0,
fakt=lambda f, e: f.get("sicherung", {}).get("name") == f"{rand}-alpaca")
fall("Kalender voriger Handelstag ausserhalb des Fensters", s_stand(),
kalender=[{"date": "2026-09-12", "open": "09:30", "close": "16:00"}] + KALENDER_TAG,
befund=f"Sicherung: kein Handelstag in den 10 Tagen vor {TAG} im Kalender", anzahl=1)
fall("Kalender Tag selbst nie der vorige", s_stand(s_eintraege({f"{TAG}-alpaca": offen})),
kalender=KALENDER_TAG + [S_KALENDER[1]], befund=None, anzahl=0,
fakt=lambda f, e: f.get("sicherung", {}).get("name") == a)
# Zusaetzlich: unbekannte Schluessel und fremde Namen stoeren nicht
fall("unbekannte Schluessel und fremde Namen",
s_stand(s_eintraege({"notiz": {"summe": None}, f"{S_VORHER}-paper": None}), neu=1), befund=None, anzahl=0)
# Bedienfehler: Ein falscher Tag bricht den Teil ab, und die Sicherung meldet sich trotzdem.
fall("Tag falsch", s_stand(), tag="2026-13-40", befund="abgebrochen", exit_code=2, anzahl=2,
auch=("Sicherung nicht geprueft (ValueError",))
# Meldeweg: Ein Befund der Sicherung allein loest die Meldung aus (lokaler Probeserver, kein Netz).
with tempfile.TemporaryDirectory() as tmp_server:
mitschrift = Path(tmp_server) / "mitschrift.jsonl"
port = 18767
server = subprocess.Popen([sys.executable, str(HIER / "ntfy_probe_server.py"), str(port), str(mitschrift)],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
try:
import socket
import time
for _ in range(50):
try:
socket.create_connection(("127.0.0.1", port), timeout=0.2).close()
break
except OSError:
time.sleep(0.1)
with tempfile.TemporaryDirectory() as tmp:
wurzel = Path(tmp)
projekt = baue_projekt(wurzel)
(wurzel / "kalender.json").write_text(json.dumps(S_KALENDER), encoding="utf-8")
(wurzel / "sicherung.json").write_text(json.dumps(s_stand(s_eintraege({a: offen}))), encoding="utf-8")
text = (ECHTES_PROJEKT / "config" / "paper.toml").read_text(encoding="utf-8")
anker = 'ntfy_url = "https://ntfy.sh"'
if text.count(anker) != 1:
raise RuntimeError("Anker ntfy_url fehlt in config/paper.toml")
(projekt / "config" / "paper.toml").write_text(
text.replace(anker, f'ntfy_url = "http://127.0.0.1:{port}"\nklartext_lokal = true'), encoding="utf-8")
(wurzel / "secrets.toml").write_text('[ntfy]\naktiv = true\nthema = "abendkontrolle-angriff"\n',
encoding="utf-8")
ergebnis = fahre_sicherung(wurzel, projekt, "alpaca", meldung=True,
env={"TRADING_GEHEIMNISSE": str(wurzel / "secrets.toml")})
zeilen = ([json.loads(z) for z in mitschrift.read_text(encoding="utf-8").split("\n") if z.strip()]
if mitschrift.exists() else [])
titel = [str(z.get("title") or "") for z in zeilen]
befunde = ergebnis.get("befunde", [])
gesendet = (ergebnis.get("fakten") or {}).get("meldung")
ok = (gesendet == "gesendet" and len(zeilen) == 1 and f"Abendkontrolle {TAG}: 1 Befund(e)" in titel[0]
and len(befunde) == 1 and befunde[0].startswith(f"Sicherung {a}: nicht bestaetigt"))
zeile = (f"{'OK ' if ok else 'FEHL'} Sicherung Befund geht in die Meldung: "
f"befunde={befunde[:2]} titel={titel} meldung={gesendet}")
print(zeile)
if not ok:
abweichungen.append(zeile)
finally:
server.terminate()
server.wait(timeout=10)
if __name__ == "__main__":
if sys.platform == "win32":
windows_faelle()
ibkr_windows_faelle()
elif "--nur-sicherung" in sys.argv[1:]:
sicherung_faelle()
else:
wsl_faelle()
ibkr_wsl_faelle()
sicherung_faelle()
print(f"ERGEBNIS: {len(abweichungen)} Abweichung(en)")
sys.exit(1 if abweichungen else 0)