"""Studie #2: Bacs-Zahlungsvolumen nach Zahlungszweck, 2020-2025. C:\\Python313\\python.exe scripts/studies/payment_rails.py Pay.UK veroeffentlicht die Bacs-Jahresstatistik nach Zahlungszweck - aber je Jahr als eigenes PDF, ohne Zeitreihe und ohne maschinenlesbare Fassung. Wer wissen will, wie sich eine Kategorie ueber sechs Jahre entwickelt hat, muss sechs PDFs nebeneinanderlegen und die Kategorienamen von Hand abgleichen. Ablauf: 1. PDFs aus data/raw/bacs/ lesen 2. Werte je Zahlungszweck extrahieren, GETRENNT nach Rail (Direct Credit / Direct Debit) - mehrere Bezeichnungen kommen auf beiden Rails vor ("Contra", "Loan advances"), und ohne Trennung ueberschreibt die eine die andere 3. Kategorienamen unabhaengig von Gross-/Kleinschreibung und Zeichensetzung zusammenfuehren (Pay.UK schreibt sie je Ausgabe anders) 4. Sprungtest und Verdraengungstest anwenden 5. data/payment-rails.json schreiben - die Seite rendert ausschliesslich das Quelle: Pay.UK, "Bacs processing statistics" 2020-2025, https://www.wearepay.uk/what-we-do/payment-systems/bacs-payment-system/bacs-payment-system-statistics/ Alle Werte sind Transaktionszahlen in Tausend, so wie Pay.UK sie ausweist. """ import io import json import re import statistics import sys from pathlib import Path sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8") REPO = Path(__file__).resolve().parent.parent.parent RAW = REPO / "data" / "raw" / "bacs" # Bewusst NICHT unter data/: Pay.UKs Nutzungsbedingungen untersagen die # Weiterverbreitung ihrer Veroeffentlichungen ("Redistribution of the # publications via another website is not permitted without the Company's # prior written consent"). Die Seite liest die Datei serverseitig; die # Root-.htaccess sperrt .json unterhalb von includes/, sie ist also nicht # oeffentlich abrufbar. Fuer eine Veroeffentlichung als Datensatz braeuchte # es eine schriftliche Freigabe von press@wearepay.uk. OUT = REPO / "includes" / "payment-rails.json" VALUE_LINE = re.compile(r"^(.*?)\s+([\d,]{3,12})\s*$") BARE_VALUE = re.compile(r"^[\d,]{3,12}$") # Zeilen, die keine Datenzeile sind. NOISE = re.compile( r"^(MASTER|NPSO|Bacs Processing Statistics|Bacs processing statistics|\d{2}$" r"|All payment figures|Payment purpose|The Bacs annual|As well as detailed)", re.I) # Kopfzeilen, die der Text-Extraktor an das folgende Label klebt. Der # Seitenbanner der Ausgaben ab 2023 hat hier lange gefehlt - dadurch fielen # unter anderem beide "General B2B"-Zeilen aus der Zeitreihe. GLUED = [ re.compile(r"^Bacs transaction volumes by payment purpose\s*[-–]\s*\(Annual statistics for \d{4}\)\s*", re.I), re.compile(r"^Bacs Direct Credits\s*(Payment purpose)?\s*(\(000s\))?\s*", re.I), re.compile(r"^Direct Debits\s*(Payment purpose)?\s*(\(000s\))?\s*", re.I), re.compile(r"^Payment purpose\s*(\(000s\))?\s*", re.I), re.compile(r"^(Business to Business (payments|collections)|Employment[- ]related payments" r"|Financial services payments|State\s*/?\s*other Benefits|Other Payment Types" r"|Insurance Premiums|Subscriptions|Utilities\s*/?\s*household bills" r"|Mortgage\s*/?\s*loan repayments|Savings and investments|Contra)\s*" r"(\(000s\))?\s*", re.I), re.compile(r"^(Direct Debits Highlights|Bacs Direct Credits Highlights)\s*", re.I), ] # Zeilen, die keine Zahlungszwecke sind, sondern Technik bzw. Restgroessen. # Sie bleiben in 'series' (nachvollziehbar), zaehlen aber nicht als Trend. CONTROL = "Unspecified / unclassified credits" TECHNICAL = {"contra", "unpaid credits", "unpaid debits", "settlement credits", "unspecified unclassified credits", "unspecified unclassified debits"} def norm_key(s: str) -> str: """Vergleichsschluessel: ohne Gross-/Kleinschreibung und Zeichensetzung.""" return re.sub(r"[^a-z0-9]+", " ", s.casefold()).strip() def clean_label(label: str) -> str: s = re.sub(r"\s+", " ", label).strip() changed = True while changed: changed = False for pat in GLUED: new = pat.sub("", s, count=1) if new != s: s, changed = new.strip(), True return s.strip(" -–—:·") def parse_pdf(path: Path) -> list[tuple[str, str, int]]: """(rail, label, value) je Datenzeile. Die Rail steht in der Seitenkopfzeile, und das ist das einzige Signal, das fuer beide Layouts traegt. 2020-2022: Seite 2 Lastschriften, Seite 3 Gutschriften, mit der Gesamtsumme der Gutschriften verwirrenderweise oben auf Seite 3. Ab 2023 umgekehrt. Seite 1 ist in beiden Faellen eine Infografik mit gerundeten Millionenwerten und wird uebersprungen. """ import pypdf rows: list[tuple[str, str, int]] = [] for page in pypdf.PdfReader(str(path)).pages: text = page.extract_text() or "" has_dc = bool(re.search(r"(?i)Bacs Direct Credits", text)) # "Direct Debits" steckt auch in "Bacs Direct Credits" nicht, aber die # Deckseite nennt beide - deshalb zaehlt nur die eindeutige Seite. has_dd = bool(re.search(r"(?i)(? 3: buf = buf[-3:] return rows def published_total(path: Path) -> int | None: """Die von Pay.UK selbst ausgewiesene Gesamtsumme, fuer die Abdeckungsangabe. Die Summenzeilen stehen in beiden Layouts unterschiedlich: 2020-2022 direkt hinter der Beschriftung, ab 2023 durch einen Fussnotentext getrennt ("Total Direct Debits All payment figures are '000s Total 5,029,499"). Deshalb wird die naechste Zahl nach der Beschriftung gesucht, nicht die unmittelbar folgende. """ import pypdf text = chr(10).join((p.extract_text() or "") for p in pypdf.PdfReader(str(path)).pages) def after(label: str) -> int | None: i = text.lower().find(label.lower()) if i < 0: return None m = re.search(r"([\d,]{7,})", text[i:i + 140]) return int(m.group(1).replace(",", "")) if m else None dd = after("Total Direct Debits") dc = after("Total Bacs Direct Credits") if dd and dc: return dd + dc return None def main() -> None: pdfs = sorted(RAW.glob("bacs-20??.pdf")) if not pdfs: raise SystemExit(f"Keine PDFs in {RAW}.") by_year: dict[str, dict[str, int]] = {} display: dict[str, str] = {} totals_published: dict[str, int | None] = {} for p in pdfs: year = p.stem.split("-")[1] cats: dict[str, int] = {} for rail, label, val in parse_pdf(p): key = f"{rail}::{norm_key(label)}" if key in cats: continue # erste Nennung gewinnt; Wiederholungen sind Kopfzeilen cats[key] = val display.setdefault(key, label) by_year[year] = cats totals_published[year] = published_total(p) print(f"{year}: {len(cats)} Zeilen ({sum(1 for k in cats if k.startswith('DC'))} Direct Credit, " f"{sum(1 for k in cats if k.startswith('DD'))} Direct Debit)") years = sorted(by_year) common = set(by_year[years[0]]) for y in years[1:]: common &= set(by_year[y]) series = {} for key in sorted(common, key=lambda k: display[k].lower()): rail = "Direct Credit" if key.startswith("DC") else "Direct Debit" series[display[key]] = { "rail": rail, "values": {y: by_year[y][key] for y in years}, } dropped = {y: sorted({display[k] for k in set(by_year[y]) - common}) for y in years} # --- Sprungtest -------------------------------------------------------- # # Eine Umklassifizierung sieht im PDF aus wie Wachstum: "Facilities # management" liegt vier Jahre bei 1,1-1,6 Mio und steht 2024 bei 125 Mio. # # Regel A: der groesste Jahressprung ueberschreitet 50% UND liegt ueber dem # Zehnfachen des Medians der uebrigen Jahresaenderungen. # Regel B faengt den Fall, den Regel A per Konstruktion verfehlt: wird eine # Kategorie hinein- und wieder herausgebucht, hebt der zweite # Sprung den Median und das Verhaeltnis faellt zusammen. Deshalb # zusaetzlich: Maximum/Minimum ueber 3 und Rueckkehr auf 25% des # Ausgangsniveaus. def yoy(v): return [((v[i + 1] - v[i]) / v[i] * 100) if v[i] else 0.0 for i in range(len(v) - 1)] trendable, stepped = {}, {} for cat, entry in series.items(): vals = [entry["values"][y] for y in years] ch = yoy(vals) mags = sorted((abs(c) for c in ch), reverse=True) largest, rest = mags[0], mags[1:] median_rest = statistics.median(rest) if rest else 0.0 # Bodenwert, damit near-flache Reihen die Division nicht sprengen. ratio = largest / max(median_rest, 0.5) rule_a = largest > 50 and ratio > 10 span = (max(vals) / min(vals)) if min(vals) > 0 else float("inf") returns = abs(vals[-1] - vals[0]) / vals[0] * 100 < 25 if vals[0] else False rule_b = span > 3 and returns biggest = max(ch, key=abs) if ch else 0.0 rec = { "rail": entry["rail"], "values": entry["values"], "change_pct": round((vals[-1] - vals[0]) / vals[0] * 100, 1) if vals[0] else None, "largest_yoy_pct": round(biggest, 1), "largest_yoy_year": years[ch.index(biggest) + 1] if ch else None, "step_ratio": round(ratio, 1), "max_min_span": round(span, 1) if span != float("inf") else None, "excluded_by": ("step" if rule_a else ("round_trip" if rule_b else None)), } (stepped if (rule_a or rule_b) else trendable)[cat] = rec # Technik- und Restzeilen sind keine Zahlungszwecke. technical = {c: trendable.pop(c) for c in list(trendable) if norm_key(c) in TECHNICAL} print() print(f"Sprungtest: {len(trendable)} Reihen als Trend verwertbar, " f"{len(stepped)} ausgeschlossen, {len(technical)} technische Zeilen aussortiert") for cat, r in sorted(stepped.items(), key=lambda kv: -abs(kv[1]["largest_yoy_pct"])): print(f" {r['excluded_by']:10} {cat[:40]:40} {r['largest_yoy_pct']:+9.0f}% in {r['largest_yoy_year']}") # --- Verdraengungstest ------------------------------------------------- displacement = [] for i in range(len(years) - 1): y0, y1 = years[i], years[i + 1] deltas = {c: e["values"][y1] - e["values"][y0] for c, e in series.items() if c != CONTROL} falls = sorted((d for d in deltas.items() if d[1] < 0), key=lambda kv: kv[1])[:3] unclass = series[CONTROL]["values"][y1] - series[CONTROL]["values"][y0] if CONTROL in series else None total_fall = -sum(v for _, v in falls) share = None if unclass is not None and unclass > 0 and total_fall > 0: share = round(unclass / total_fall * 100, 1) displacement.append({ "year": y1, "largest_falls": [{"category": c, "change": v} for c, v in falls], "unclassified_change": unclass, "share_of_falls_absorbed_pct": share, }) print() print("Verdraengungstest:") for r in displacement: sh = f"{r['share_of_falls_absorbed_pct']}%" if r['share_of_falls_absorbed_pct'] is not None else '-' uc = f"{r['unclassified_change']:+,}" if r['unclassified_change'] is not None else 'n/a' print(f" {r['year']}: Sammelzeile {uc}, deckt {sh} der drei groessten Rueckgaenge") # --- Abdeckung --------------------------------------------------------- coverage = {} for y in years: trend_sum = sum(r["values"][y] for r in trendable.values()) pub = totals_published.get(y) coverage[y] = { "trendable_sum": trend_sum, "published_total": pub, "share_pct": round(trend_sum / pub * 100, 1) if pub else None, } print() print("Abdeckung gegen die von Pay.UK ausgewiesene Gesamtsumme:") for y in years: c = coverage[y] print(f" {y}: {c['trendable_sum']:>10,} von {c['published_total']:>10,} = {c['share_pct']}%") data = { "source": { "name": "Pay.UK — Bacs annual processing statistics", "url": "https://www.wearepay.uk/what-we-do/payment-systems/bacs-payment-system/bacs-payment-system-statistics/", "years": years, "unit": "thousands of transactions", "note": ("Pay.UK publishes one PDF per year with no time series. This file joins them on " "payment purpose, keyed separately for Direct Credit and Direct Debit because " "several labels appear on both rails. Only categories present in every year are " "kept; excluded labels are listed under 'not_comparable'."), }, "years": years, "series": series, "trendable": trendable, "reclassified": stepped, "technical": technical, "coverage": coverage, "not_comparable": dropped, "displacement_check": displacement, "control_series": CONTROL, "method": { "keying": ("Series are keyed on rail plus payment purpose. Labels are matched across " "editions ignoring case and punctuation, because Pay.UK varies both."), "step_test": ("Rule A: a series is treated as reclassified when its largest single " "year-on-year change exceeds 50% and is more than ten times the median of " "its other year-on-year changes (the median is floored at 0.5 percentage " "points so a near-flat series cannot make the ratio meaningless). Rule B " "catches what Rule A cannot: a category booked in and then back out again " "has two steps, which raises the median and collapses the ratio, so a " "series whose maximum is more than three times its minimum while ending " "within 25% of where it started is also excluded."), "displacement_test": ("Pay.UK carries a catch-all line, 'Unspecified / unclassified " "credits'. Where it grows in the same year that named categories " "fall, the fall cannot be read as volume leaving Bacs. Where it " "shrinks sharply, named categories may be inflated by volume " "moving the other way. Reported for every year."), "excluded_lines": ("Contra, unpaid credits and debits, settlement credits and the " "unclassified residual are reported in 'technical' rather than as " "trends: they are settlement and reject mechanics, not payment purposes."), }, } OUT.write_text(json.dumps(data, indent=1), encoding="utf-8") print() print(f"{len(series)} vergleichbare Kategorien, {len(trendable)} als Trend verwertbar") print(f"Geschrieben: {OUT} ({OUT.stat().st_size:,} Bytes)") if __name__ == "__main__": main()