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

"""Vergleicht zwei JSONL-Protokolle der Messwerkzeuge Wert fuer Wert (Bitgleichheit nach einem Umbau).

Koordinator-Werkzeug, nicht fuers Repo. Frage (Herkunft: Projektvorgabe, Anforderung Paket B vom
2026-09-18, "Die eine Regel ueber allem"): Liefert der umgebaute Code auf denselben echten Daten dieselben
Zahlen wie das Referenzprotokoll -- jede Gleitkommazahl exakt gleich, nan gleich nan?
Es unterscheidet drei Dinge und vermischt sie nicht:
  ABWEICHUNG  ein Blattwert unter demselben Pfad ist anders            -> Befund (Exit 2)
  NUR NEU     ein Pfad steht nur im neuen Protokoll (neuer Schluessel) -> gemeldet, kein Befund, ausser mit --streng
  NUR ALT     ein Pfad steht nur im alten Protokoll                    -> Befund (Exit 2)
Zeilen werden in ihrer Reihenfolge gepaart; verschiedene Zeilenzahlen oder Zeilenarten sind ein Befund.
Uebergangen werden die Schluessel aus --ohne (Vorgabe: zeit, dauer_s, protokoll, name), weil sie vom Lauf
abhaengen und nicht von der Rechnung.
Was es nicht sieht: ob das Referenzprotokoll selbst stimmt, und Zeilen, die in beiden Protokollen fehlen.
Regressionsnachweis: --sabotage veraendert im neuen Protokoll eine Zahl in der letzten Stelle und muss rot enden.

Aufruf: python protokoll_vergleich.py <alt.jsonl> <neu.jsonl> [--ohne k1 k2 ...] [--streng] [--sabotage]
Exit 0 ohne Befund, 2 mit Befund, 3 wenn eine Datei fehlt oder leer ist.
"""

import json
import math
import sys
from pathlib import Path


def lies(pfad: Path) -> list[dict]:
    zeilen = []
    with open(pfad, encoding="utf-8") as datei:
        for zeile in datei:
            if zeile.strip():
                zeilen.append(json.loads(zeile, parse_constant=float))
    return zeilen


def flach(knoten, ohne: set[str], praefix: str = "") -> dict:
    werte: dict = {}
    if isinstance(knoten, dict):
        for name, inhalt in knoten.items():
            if name in ohne:
                continue
            werte.update(flach(inhalt, ohne, f"{praefix}.{name}"))
    elif isinstance(knoten, list):
        for nummer, inhalt in enumerate(knoten):
            werte.update(flach(inhalt, ohne, 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 main() -> int:
    argumente = sys.argv[1:]
    streng = "--streng" in argumente
    sabotage = "--sabotage" in argumente
    ohne = {"zeit", "dauer_s", "protokoll", "name"}
    if "--ohne" in argumente:
        stelle = argumente.index("--ohne")
        ohne = set()
        for wort in argumente[stelle + 1 :]:
            if wort.startswith("--"):
                break
            ohne.add(wort)
    pfade = [Path(a) for a in argumente if not a.startswith("--") and a not in ohne]
    if len(pfade) != 2 or not all(p.is_file() for p in pfade):
        print(__doc__)
        return 3
    alt, neu = lies(pfade[0]), lies(pfade[1])
    if not alt or not neu:
        print("VORAUSSETZUNG FEHLT: ein Protokoll ist leer")
        return 3
    if sabotage:
        for zeile in reversed(neu):
            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 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)}")
    verglichen = 0
    nur_neu: dict[str, int] = {}
    for nummer, (a, n) in enumerate(zip(alt, neu), start=1):
        if a.get("art") != n.get("art"):
            befunde.append(f"Zeile {nummer}: Art alt {a.get('art')!r}, neu {n.get('art')!r}")
            continue
        fa, fn = flach(a, ohne), flach(n, ohne)
        for pfad in fa.keys() - fn.keys():
            befunde.append(f"Zeile {nummer} ({a.get('art')}): NUR ALT {pfad}")
        for pfad in fn.keys() - fa.keys():
            kurz = pfad.split("[")[0]
            nur_neu[kurz] = nur_neu.get(kurz, 0) + 1
            if streng:
                befunde.append(f"Zeile {nummer} ({a.get('art')}): NUR NEU {pfad}")
        for pfad in fa.keys() & fn.keys():
            verglichen += 1
            if not gleich(fa[pfad], fn[pfad]):
                befunde.append(f"Zeile {nummer} ({a.get('art')}): ABWEICHUNG {pfad}: alt {fa[pfad]!r}, neu {fn[pfad]!r}")

    print(f"{min(len(alt), len(neu))} Zeilen gepaart, {verglichen} Werte verglichen, uebergangene Schluessel: {sorted(ohne)}")
    for pfad, zahl in sorted(nur_neu.items()):
        print(f"NUR NEU: {pfad} in {zahl} Zeile(n)")
    for befund in befunde[:25]:
        print("BEFUND:", befund)
    if len(befunde) > 25:
        print(f"... und {len(befunde) - 25} weitere")
    print(f"{len(befunde)} Befund(e).")
    return 2 if befunde else 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)