claude-regelwerk/.vault-tools/trading-plattform-pruefung/probe_neu_einfrieren.py

"""Friert die T0/T1-Probe der Tagesmessung nur dann neu ein, wenn der Diff genau der vorhergesagte ist.

Koordinator-Werkzeug, nicht fuers Repo. Frage (Herkunft: Plan Paket B vom 2026-09-18, Task 154, und die Messung
der Plankritik): Aendert der Umbau der gepoolten Prognose an der eingefrorenen Probe
tests/fixtures/tage_t0_t1_referenz.json genau das, was er aendern darf -- und sonst nichts?
Erlaubt ist:
  - in jeder Zeile der Art zelle_pool der neue Schluessel fenster_ausgelassen_pool,
  - in der Zeile zelle_pool mit merkmale T1 geaenderte Werte in den Feldern aus ERLAUBT_T1_POOL.
Alles andere ist ein Befund: jede geaenderte Zahl in einer Zeile zelle_instrument, in einer T0-Zeile oder in
einem Feld ausserhalb der Liste, jeder fehlende Schluessel, jede andere Zeilenzahl.
Die neuen Zeilen entstehen ueber den Helfer des Tests (_t0_t1_protokollzeilen), also mit dem Code des Baums,
aus dem das Werkzeug gestartet wird. Geschrieben wird nur mit --schreiben und nur ohne Befund, im Format des
Rezepts aus Teil 9 (indent 2, ensure_ascii aus, Zeilenende LF -- deshalb in WSL starten).
Was es nicht sieht: ob die neuen Zahlen richtig sind. Es belegt nur, dass sich nichts ausserhalb des
vorhergesagten Bereichs bewegt hat.
Regressionsnachweis: --sabotage verschiebt in den neuen Zeilen eine Zahl einer T0-Zeile um ein ULP; das muss
einen Befund geben, und --schreiben darf dann nichts schreiben.

Aufruf (WSL, aus der Projektwurzel): python probe_neu_einfrieren.py [--schreiben] [--sabotage]
Exit 0 ohne Befund, 2 mit Befund, 3 wenn eine Voraussetzung fehlt.
"""

import json
import math
import sys
import tempfile
from pathlib import Path

FIXTURE = Path("tests/fixtures/tage_t0_t1_referenz.json")
NEUER_SCHLUESSEL = "fenster_ausgelassen_pool"
# Gemessen von der Plankritik am 2026-09-18 (Feld-fuer-Feld-Vergleich auf einer Kopie mit angewendetem Plan).
ERLAUBT_T1_POOL = (
    "K3",
    "anzahl_fenster",
    "ic_haelfte_1",
    "ic_haelfte_2",
    "ic_pool",
    "n_pool",
    "perioden",
    "sharpe_",
    "t_pool_naiv",
    "treffer_pool",
)


def flach(knoten, praefix: str = "") -> dict:
    werte: dict = {}
    if isinstance(knoten, dict):
        for name, inhalt in knoten.items():
            werte.update(flach(inhalt, f"{praefix}.{name}"))
    elif isinstance(knoten, list):
        for nummer, inhalt in enumerate(knoten):
            werte.update(flach(inhalt, f"{praefix}[{nummer}]"))
    else:
        werte[praefix] = knoten
    return werte


def gleich(a, b) -> bool:
    if isinstance(a, float) and isinstance(b, float):
        return a == b or (math.isnan(a) and math.isnan(b))
    return type(a) is type(b) and a == b


def erlaubt(zeile: dict, pfad: str) -> bool:
    if zeile.get("art") != "zelle_pool" or zeile.get("merkmale") != "T1":
        return False
    feld = pfad.lstrip(".").split(".")[0].split("[")[0]
    return any(feld == name or (name.endswith("_") and feld.startswith(name)) for name in ERLAUBT_T1_POOL)


def main() -> int:
    argumente = sys.argv[1:]
    if not FIXTURE.is_file() or not Path("tests/test_training_signal_tage.py").is_file():
        print("VORAUSSETZUNG FEHLT: aus der Projektwurzel starten (tests/fixtures und tests/ muessen da sein)")
        return 3
    sys.path.insert(0, "tests")
    sys.path.insert(0, ".")
    from test_training_signal_tage import _t0_t1_protokollzeilen

    alt = json.loads(FIXTURE.read_text(encoding="utf-8"))
    with tempfile.TemporaryDirectory() as ordner:
        neu = _t0_t1_protokollzeilen(Path(ordner))
    # Dieselbe Reise wie die Fixture: einmal durch JSON, damit nan/inf und Typen gleich ankommen.
    neu = json.loads(json.dumps(neu, ensure_ascii=False))

    if "--sabotage" in argumente:
        for zeile in neu:
            if zeile.get("merkmale") == "T0":
                zahlen = [k for k, v in zeile.items() if isinstance(v, float) and math.isfinite(v) and v != 0.0]
                if zahlen:
                    zeile[zahlen[0]] = math.nextafter(zeile[zahlen[0]], math.inf)
                    print(f"sabotage: {zahlen[0]} in einer T0-Zeile der Art {zeile.get('art')} um ein ULP verschoben")
                    break

    befunde: list[str] = []
    if len(alt) != len(neu):
        befunde.append(f"Zeilenzahl: alt {len(alt)}, neu {len(neu)}")
    geaendert: dict[str, list[str]] = {}
    dazu: dict[str, list[str]] = {}
    for nummer, (a, n) in enumerate(zip(alt, neu), start=1):
        kennung = f"Zeile {nummer} ({a.get('art')}/{a.get('symbol', '-')}/{a.get('merkmale')})"
        if (a.get("art"), a.get("symbol"), a.get("merkmale")) != (n.get("art"), n.get("symbol"), n.get("merkmale")):
            befunde.append(f"{kennung}: die neue Zeile ist eine andere ({n.get('art')}/{n.get('symbol', '-')}/{n.get('merkmale')})")
            continue
        fa, fn = flach(a), flach(n)
        for pfad in sorted(fa.keys() - fn.keys()):
            befunde.append(f"{kennung}: NUR ALT {pfad}")
        for pfad in sorted(fn.keys() - fa.keys()):
            if pfad == f".{NEUER_SCHLUESSEL}" and a.get("art") == "zelle_pool":
                dazu.setdefault(kennung, []).append(f"{pfad} = {fn[pfad]!r}")
            else:
                befunde.append(f"{kennung}: NUR NEU {pfad}")
        for pfad in sorted(fa.keys() & fn.keys()):
            if gleich(fa[pfad], fn[pfad]):
                continue
            if erlaubt(a, pfad):
                geaendert.setdefault(kennung, []).append(pfad)
            else:
                befunde.append(f"{kennung}: ABWEICHUNG ausserhalb des Erlaubten {pfad}: alt {fa[pfad]!r}, neu {fn[pfad]!r}")
    pool_zeilen = [z for z in alt if z.get("art") == "zelle_pool"]
    if len(dazu) != len(pool_zeilen):
        befunde.append(f"der neue Schluessel steht in {len(dazu)} von {len(pool_zeilen)} Zeilen der Art zelle_pool")

    for kennung, pfade in dazu.items():
        print(f"NEU   {kennung}: {', '.join(pfade)}")
    for kennung, pfade in geaendert.items():
        print(f"ERLAUBT GEAENDERT {kennung}: {len(pfade)} Werte: {', '.join(pfade)}")
    for befund in befunde[:30]:
        print("BEFUND:", befund)
    print(f"{len(befunde)} Befund(e).")
    if befunde:
        return 2
    if "--schreiben" in argumente:
        with open(FIXTURE, "w", encoding="utf-8", newline="\n") as datei:
            datei.write(json.dumps(neu, ensure_ascii=False, indent=2) + "\n")
        print(f"geschrieben: {FIXTURE}")
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except Exception as ausnahme:  # noqa: BLE001 - ein Absturz des Werkzeugs ist selbst ein Befund
        print(f"BEFUND: das Werkzeug ist abgebrochen: {ausnahme!r}")
        sys.exit(2)