"""Erhebung 2026, Schritt 3: die sieben Kennzahlen aus Abschnitt 6 der Methodik berechnen und die Veroeffentlichungsdateien nach Abschnitt 10 erzeugen. Eingabe: erhebung-2026/auswertung/klassifikation.csv erhebung-2026/auswertung/stichprobe-handpruefung.csv erhebung-2026/rohdaten/abruf-protokoll.json Ausgabe: erhebung-2026/veroeffentlichung/kennzahlen.json erhebung-2026/veroeffentlichung/meldungen-ausgewertet.csv erhebung-2026/veroeffentlichung/ausschluesse.csv erhebung-2026/veroeffentlichung/handpruefung.csv Aufruf aus dem Ordner backend: python scripts/erhebung_kennzahlen.py """ from __future__ import annotations import csv import json import re import shutil import statistics from collections import Counter, defaultdict from datetime import date from pathlib import Path REPO = Path(__file__).resolve().parents[2] AUSWERTUNG = REPO / "erhebung-2026" / "auswertung" ROHDATEN = REPO / "erhebung-2026" / "rohdaten" ZIEL = REPO / "erhebung-2026" / "veroeffentlichung" KANTONE = [ "ZH", "BE", "LU", "UR", "SZ", "OW", "NW", "GL", "ZG", "SO", "BS", "BL", "SH", "AR", "AI", "SG", "GR", "AG", "TG", ] KK_PRAEFIXE = r"^(Konkurspublikation/Schuldenruf|Kollokationsplan und Inventar|Vorläufige Konkursanzeige|Konkurseröffnung|Löschung|Mutation|Neueintragung)\s+" def wahr(v: str) -> bool: return v == "True" def betrieb_schluessel(m: dict) -> str: """Ein Konkurs erzeugt mehrere Meldungen (Schuldenruf, Kollokationsplan ...). Fuer die Zaehlung je Betrieb: UID, sonst bereinigter Name.""" if m["uid"]: return m["uid"] name = m["firma"] or re.sub(KK_PRAEFIXE, "", m["titel"]) name = re.sub(r"\s+in Liquidation", "", name) name = name.split(",")[0] return re.sub(r"[^a-z0-9]", "", name.lower()) def main() -> None: ZIEL.mkdir(parents=True, exist_ok=True) meldungen = list(csv.DictReader((AUSWERTUNG / "klassifikation.csv").open(encoding="utf-8"))) probe = list(csv.DictReader((AUSWERTUNG / "stichprobe-handpruefung.csv").open(encoding="utf-8"))) protokoll = json.loads((ROHDATEN / "abruf-protokoll.json").read_text(encoding="utf-8")) for m in meldungen: m["kanton"] = m["kantone"].split("|")[0] praxen = [m for m in meldungen if wahr(m["zaehlt_praxis"])] abgang_meldungen = [m for m in meldungen if wahr(m["zaehlt_laborabgang"])] # Laborabgaenge je Betrieb: erste Meldung je Betrieb behalten betriebe: dict[str, dict] = {} for m in sorted(abgang_meldungen, key=lambda x: x["publikationsdatum"]): betriebe.setdefault(betrieb_schluessel(m), m) abgaenge = list(betriebe.values()) praxen_je_kanton = Counter(m["kanton"] for m in praxen) abgaenge_je_kanton = Counter(m["kanton"] for m in abgaenge) art_je_kanton: dict[str, Counter] = defaultdict(Counter) for m in praxen: art_je_kanton[m["kanton"]][m["ereignis"]] += 1 kantone_tabelle = [] for k in KANTONE: p, a = praxen_je_kanton.get(k, 0), abgaenge_je_kanton.get(k, 0) kantone_tabelle.append({ "kanton": k, "ansprechbare_praxen": p, "neueintragung": art_je_kanton[k]["neueintragung"], "personenwechsel": art_je_kanton[k]["personenwechsel"], "laborabgaenge": a, "verhaeltnis_praxen_zu_abgaengen": round(p / a, 1) if a else None, }) # Publikationsverzug: Tage zwischen Erstellung und Publikation der Meldung verzug = [] for m in meldungen: if m["erstellungsdatum"] and m["publikationsdatum"]: e, p = date.fromisoformat(m["erstellungsdatum"]), date.fromisoformat(m["publikationsdatum"]) verzug.append((p - e).days) # Handpruefung nach 4.5: Fehlzuordnungen in beide Richtungen zaehlend = [r for r in probe if wahr(r["zaehlt_praxis"]) or wahr(r["zaehlt_laborabgang"])] nicht = [r for r in probe if not (wahr(r["zaehlt_praxis"]) or wahr(r["zaehlt_laborabgang"]))] fehler_zaehlend = sum(1 for r in zaehlend if r["handpruefung_korrekt"].strip().lower() == "nein") fehler_nicht = sum(1 for r in nicht if r["handpruefung_korrekt"].strip().lower() == "nein") ausschluesse = [ m for m in meldungen if m["typ"] == "praxis" and m["unterrubrik"] in ("HR01", "HR02") and not wahr(m["zaehlt_praxis"]) ] kennzahlen = { "stand": date.today().isoformat(), "zeitfenster_publikationsdatum": protokoll["fenster"], "abruf": protokoll["abruf_begonnen"], "meldungen_gesamt": len(meldungen), "meldungen_nach_rubrik": protokoll["nach_rubrik"], "ansprechbare_praxen_gesamt": len(praxen), "aufteilung_nach_art": dict(Counter(m["ereignis"] for m in praxen)), "laborabgaenge_meldungen": len(abgang_meldungen), "laborabgaenge_betriebe": len(abgaenge), "laborabgaenge_nach_art": dict(Counter("konkurs" if m["rubrik"] == "KK" else "loeschung" for m in abgaenge)), "je_kanton": kantone_tabelle, "publikationsverzug_tage_median": statistics.median(verzug) if verzug else None, "publikationsverzug_basis": len(verzug), "handpruefung": { "stichprobe": len(probe), "startwert": 20260921, "zaehlend_geprueft": len(zaehlend), "zaehlend_fehlzuordnungen": fehler_zaehlend, "nicht_zaehlend_geprueft": len(nicht), "nicht_zaehlend_fehlzuordnungen": fehler_nicht, "hinweis": "Die Fehlzuordnungen wurden nach der Pruefung korrigiert (Handkorrekturen), " "die Regel wurde enger gefasst; die Quote beschreibt den Stand vor der Korrektur.", }, "entscheid_durch": dict(Counter(m["entscheid_durch"] for m in meldungen)), "praxis_ausschluesse_nach_grund": dict(Counter(m["ausschluss"] for m in ausschluesse)), } (ZIEL / "kennzahlen.json").write_text(json.dumps(kennzahlen, ensure_ascii=False, indent=2), encoding="utf-8") # Veroeffentlichungsdateien nach Abschnitt 10 oeffentlich = [ "publikationsdatum", "erstellungsdatum", "publikationsnummer", "rubrik", "unterrubrik", "kanton", "titel", "typ", "ereignis", "ausschluss", "zaehlt_praxis", "zaehlt_laborabgang", "entscheid_durch", "begruendung", "url", ] with (ZIEL / "meldungen-ausgewertet.csv").open("w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=oeffentlich, extrasaction="ignore") w.writeheader() w.writerows(sorted(meldungen, key=lambda m: (m["kanton"], m["publikationsdatum"]))) with (ZIEL / "ausschluesse.csv").open("w", newline="", encoding="utf-8") as f: w = csv.DictWriter(f, fieldnames=oeffentlich, extrasaction="ignore") w.writeheader() w.writerows(sorted(ausschluesse, key=lambda m: (m["ausschluss"], m["kanton"]))) shutil.copy(AUSWERTUNG / "stichprobe-handpruefung.csv", ZIEL / "handpruefung.csv") print(json.dumps({k: v for k, v in kennzahlen.items() if k != "je_kanton"}, ensure_ascii=False, indent=1)) print(f"{'Kanton':<8}{'Praxen':>7}{'Neu':>5}{'Wechsel':>9}{'Abgaenge':>10}") for z in kantone_tabelle: print(f"{z['kanton']:<8}{z['ansprechbare_praxen']:>7}{z['neueintragung']:>5}" f"{z['personenwechsel']:>9}{z['laborabgaenge']:>10}") print(f"\nDateien in {ZIEL}") if __name__ == "__main__": main()