docs: Plan-Haekchen Ausbaustufe 9

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

docs/superpowers/plans/2026-07-20-ausbaustufe-9.md

@@ -0,0 +1,196 @@
+# Ausbaustufe 9 Implementation Plan — Vorschlags-Algorithmus v2 (v0.9.0)
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [x]`) syntax for tracking.
+
+**Goal:** `suggest_recurring` erkennt monatliche/vierteljährliche/jährliche Serien mit letztem Betrag, Aktiv-Check, Betrags-Clustern, Umfirmierungs-Merge und robustem Bestandsabgleich; GUI zeigt Rhythmus/Start/Hinweis; Release v0.9.0 mit Live-Gate gegen die echten Daten.
+
+**Architecture:** Vollständiger Rewrite von `app/services/suggestions.py` (reine Session-in/dict-out-Funktion, Parameter als Modul-Konstanten); `SuggestionOut`-Erweiterung in `routers/planning.py`; Template-Anpassung der Vorschlags-Tabelle. Kein Datenmodell-/Migrationsbedarf.
+
+**Tech Stack:** SQLAlchemy 2, Pydantic v2, Jinja2, pytest (synthetische Daten).
+
+**Spec:** `docs/superpowers/specs/2026-07-20-vorschlags-algorithmus-v2-design.md` — die dortigen Abschnitte „Algorithmus" (8 Schritte, Konstanten) und „Tests" sind bindend und Teil dieses Plans.
+
+## Global Constraints
+
+- Beträge `Decimal` (keine float-Arithmetik, auch nicht in Toleranzvergleichen — relative Differenzen als `Decimal`-Quotienten).
+- `date.today()` nur an EINER Stelle (Parameter `today: date | None = None` der Hauptfunktion, Default heute) — Tests injizieren ein festes Datum.
+- Tests ausschließlich mit synthetischen Daten (DATENSCHUTZ: keine echten Namen/Beträge aus der Live-DB in Tests/Commits).
+- GUI deutsch, TT.MM.JJJJ, `|eur`, `|de_label`; API Punkt-Dezimal.
+- Fable-Testagent-Gate je Task VOR Commit; Ledger-Eintrag je Task.
+- Testlauf: `cd /home/wlfb/bin/finance && .venv/bin/python -m pytest -q` — Basis 187 passed, muss grün bleiben (drei bestehende Suggestion-Tests DÜRFEN an die neue Semantik angepasst werden, siehe Task 1 Step 4).
+
+---
+
+### Task 1: Algorithmus-Rewrite + API-Schema
+
+**Files:**
+- Rewrite: `app/services/suggestions.py`
+- Modify: `app/routers/planning.py` (`SuggestionOut`)
+- Modify: `tests/test_planning_api.py`
+
+**Interfaces:**
+- Produces: `suggest_recurring(session, today: date | None = None) -> list[dict]` mit Keys `name, amount (Decimal), rhythm, due_day, start_date (date|None), category_id, hinweis (str)`; `SuggestionOut` mit denselben Feldern (`start_date: date | None = None`, `hinweis: str = ""`). Task 2 rendert genau diese Felder.
+
+- [x] **Step 1: Failing Tests** — in `tests/test_planning_api.py` die drei bestehenden Suggestion-Tests ERSETZEN/ERWEITERN durch die Spec-Fälle (Helper zum Anlegen synthetischer Buchungen schreiben; `dedup_hash` eindeutig, `status="confirmed"`; ein Account genügt; `today=date(2026, 7, 20)` in alle Aufrufe injizieren):
+
+```python
+from datetime import date
+from decimal import Decimal
+
+def _tx(db, acc_id, d, amount, cp, cat=None):
+    db.add(Transaction(account_id=acc_id, booking_date=d, amount=Decimal(amount),
+                       purpose="p", counterparty=cp, category_id=cat,
+                       status="confirmed", dedup_hash=f"h-{cp}-{d}-{amount}"))
+
+TODAY = date(2026, 7, 20)
+
+def test_suggest_letzter_betrag_bei_preiserhoehung(db):
+    acc = _acc(db)  # Helper: Account anlegen, gibt id zurueck
+    for d, a in [(date(2026, 3, 1), "-190.65"), (date(2026, 4, 1), "-202.94"),
+                 (date(2026, 5, 4), "-202.94"), (date(2026, 6, 1), "-202.94"),
+                 (date(2026, 7, 1), "-202.94")]:
+        _tx(db, acc, d, a, "Entis Lebensversicherung AG")
+    db.commit()
+    out = suggest_recurring(db, today=TODAY)
+    assert len(out) == 1
+    s = out[0]
+    assert s["amount"] == Decimal("-202.94") and s["rhythm"] == "monthly"
+    assert s["due_day"] == 1 and s["start_date"] is None
+
+def test_suggest_quartal_mit_phase(db):
+    acc = _acc(db)
+    for d in [date(2025, 9, 15), date(2025, 12, 15), date(2026, 3, 16), date(2026, 6, 15)]:
+        _tx(db, acc, d, "-55.08", "Rundfunk ARD ZDF")
+    db.commit()
+    out = suggest_recurring(db, today=TODAY)
+    assert len(out) == 1
+    assert out[0]["rhythm"] == "quarterly"
+    assert out[0]["start_date"] == date(2026, 6, 15) and out[0]["due_day"] == 15
+
+def test_suggest_jahr_mit_zwei_belegen(db):
+    acc = _acc(db)
+    for d, a in [(date(2025, 6, 16), "-409.92"), (date(2026, 6, 16), "-467.33")]:
+        _tx(db, acc, d, a, "Kraftfahrer-Schutz e.V.")
+    db.commit()
+    out = suggest_recurring(db, today=TODAY)
+    assert len(out) == 1
+    assert out[0]["rhythm"] == "yearly" and out[0]["amount"] == Decimal("-467.33")
+    assert out[0]["start_date"] == date(2026, 6, 16)
+    assert "409.92" in out[0]["hinweis"]  # Betrag zuletzt gestiegen
+
+def test_suggest_tote_serie_kein_vorschlag(db):
+    acc = _acc(db)
+    for m in (9, 10, 11, 12):
+        _tx(db, acc, date(2025, m, 1), "-35.00", "WWK Alt")
+    db.commit()
+    assert suggest_recurring(db, today=TODAY) == []
+
+def test_suggest_umfirmierung_merge(db):
+    acc = _acc(db)
+    for m in (11, 12):
+        _tx(db, acc, date(2025, m, 1), "-190.65", "Heidelberger Leben")
+    for m in (1, 2, 3):
+        _tx(db, acc, date(2026, m, 2), "-190.65", "Entis Lebensversicherung")
+    for m in (4, 5, 6, 7):
+        _tx(db, acc, date(2026, m, 1), "-202.94", "Entis Lebensversicherung")
+    db.commit()
+    out = suggest_recurring(db, today=TODAY)
+    assert len(out) == 1
+    assert "Entis" in out[0]["name"] and out[0]["amount"] == Decimal("-202.94")
+
+def test_suggest_bestandsabgleich_trotz_preisdrift(db):
+    acc = _acc(db)
+    db.add(RecurringItem(name="Entis Lebensversicherung AG", amount=Decimal("-190.65"),
+                         rhythm="monthly", due_day=1))
+    for m in (4, 5, 6, 7):
+        _tx(db, acc, date(2026, m, 1), "-202.94", "Entis Lebensversicherung AG")
+    db.commit()
+    assert suggest_recurring(db, today=TODAY) == []  # Namens-Match schlaegt an
+
+def test_suggest_zwei_vertraege_getrennt(db):
+    acc = _acc(db)
+    for m in (4, 5, 6, 7):
+        _tx(db, acc, date(2026, m, 1), "-346.23", "Heidelberger LV")
+        _tx(db, acc, date(2026, m, 2), "-145.21", "Heidelberger LV")
+    db.commit()
+    out = suggest_recurring(db, today=TODAY)
+    assert len(out) == 2
+    assert {s["amount"] for s in out} == {Decimal("-346.23"), Decimal("-145.21")}
+```
+
+(`_acc`-Helper analog bestehender Tests; `RecurringItem`/`Transaction`-Importe existieren.) Die drei Alt-Tests (`three_consecutive_months_with_year_wrap`, `two_months_no_suggestion`, `excludes_existing_recurring_item`) an die neue Signatur/Semantik anpassen: feste `today`-Injektion; Daten ggf. ins Fenster schieben; der Exclusion-Test bleibt inhaltlich gültig (Name-Match).
+
+Run: `.venv/bin/python -m pytest tests/test_planning_api.py -q` → neue Tests FAIL
+
+- [x] **Step 2: Rewrite `app/services/suggestions.py`** gemäß Spec-Abschnitt „Algorithmus" (8 Schritte, Konstanten `WINDOW_DAYS=460`, Rhythmus-Tabelle monthly 25-36/≥3, quarterly 80-105/≥3, yearly 330-400/≥2, `STEP_DAYS={"monthly":30,"quarterly":91,"yearly":365}`, `ACTIVITY_FACTOR` 7/4 als `Fraction` oder Tage-Vergleich ganzzahlig, Cluster 35 %, Merge 25 %, Bestand 10 % — alle Toleranzvergleiche als `Decimal`). Struktur: `_norm`, `_rel_diff`, `_amount_clusters` (greedy gegen letztes Mitglied, gleiches Vorzeichen), `_classify` (Median der Abstände), Merge-Pass je Konto über alle Serien, `_covered_by_existing`, Hauptfunktion `suggest_recurring(session, today=None)`. Deutsche Docstrings/Kommentare zur Begründung der Toleranzen.
+
+- [x] **Step 3: `SuggestionOut` erweitern** — `routers/planning.py`:
+
+```python
+class SuggestionOut(BaseModel):
+    name: str
+    amount: Decimal
+    rhythm: str
+    due_day: int
+    start_date: date | None = None
+    category_id: int | None = None
+    hinweis: str = ""
+```
+
+- [x] **Step 4: Tests + Suite grün** — `.venv/bin/python -m pytest -q` → PASS (Alt-Test-Anpassungen im Report begründen).
+
+- [x] **Step 5: Fable-Testagent-Abnahme** (Faktencheck: Toleranz-Arithmetik Decimal-rein; Aktiv-Check-Grenzen; Merge-Bedingungen; keine `date.today()`-Streuung; Alt-Test-Anpassungen berechtigt). Erst nach VERIFIED weiter.
+
+- [x] **Step 6: Commit** — `git add finance/app/services/suggestions.py finance/app/routers/planning.py finance/tests/test_planning_api.py && git commit -m "feat: Vorschlags-Algorithmus v2 (Rhythmen, letzter Betrag, Aktiv-Check, Merge)"`
+
+---
+
+### Task 2: GUI — Rhythmus/Start/Hinweis in der Vorschlags-Tabelle
+
+**Files:**
+- Modify: `app/templates/planning.html` (Fieldset „Vorschläge aus Buchungen")
+- Modify: `tests/test_gui.py`
+
+**Interfaces:**
+- Consumes: `SuggestionOut`-Felder aus Task 1; Filter `|eur`/`|de_label`.
+
+- [x] **Step 1: Failing GUI-Test** — in `tests/test_gui.py` (synthetische Serie seeden, `/planung` laden):
+
+```python
+def test_vorschlaege_zeigen_rhythmus_und_start(client, db):
+    client.post("/login", data={"username": "admin", "password": "geheim"})
+    acc = Account(bank="dkb", iban="DE-SUG-1", name="S", type="giro")
+    db.add(acc)
+    db.flush()
+    for d in (date(2025, 9, 15), date(2025, 12, 15), date(2026, 3, 16), date(2026, 6, 15)):
+        db.add(Transaction(account_id=acc.id, booking_date=d, amount=Decimal("-55.08"),
+                           purpose="p", counterparty="Rundfunk Synth", status="confirmed",
+                           dedup_hash=f"sug-{d}"))
+    db.commit()
+    r = client.get("/planung").text
+    assert "vierteljährlich" in r          # de_label des Rhythmus
+    assert "15.06.2026" in r               # Start-Spalte TT.MM.JJJJ
+    assert 'name="start_date"' in r        # hidden input der Uebernahme
+```
+
+WICHTIG: Der Test hängt von `date.today()` der App ab (Aktiv-Check!) — Serie so legen, dass sie um den echten Testlauf-Zeitpunkt herum aktiv ist, oder (besser) `suggest_recurring` in `planung_page` unverändert lassen und den Test mit relativen Daten um `date.today()` konstruieren (letzte Buchung ≤ 45 Tage vor heute, Quartalsschritte rückwärts). Die Variante mit relativen Daten umsetzen; die obigen Fixdaten sind als Muster zu verstehen und auf `date.today()`-relative Werte umzustellen (inkl. erwartetem Start-String via `.strftime('%d.%m.%Y')`).
+
+- [x] **Step 2: Template** — Vorschlags-Tabelle: Kopf `Name | Betrag | Rhythmus | Fälligkeitstag | Start | (Aktion)`; Zellen `{{ s.rhythm|de_label }}`, `{{ s.start_date.strftime('%d.%m.%Y') if s.start_date else '–' }}`; Betrag-Zelle ergänzt `{% if s.hinweis %}<span class="muted">{{ s.hinweis }}</span>{% endif %}`; Übernahme-Formular: hidden inputs unverändert plus `<input type="hidden" name="start_date" value="{{ s.start_date.isoformat() if s.start_date else '' }}">` (json-form macht leer → null). Hinweistext unter dem Fieldset: „Erkannt werden monatliche, vierteljährliche und jährliche Serien; Betrag = jeweils letzte Buchung."
+
+- [x] **Step 3: Suite grün**; **Step 4: Fable-Abnahme** (Live-Approximation: Rendering + Übernahme-Roundtrip eines Quartals-Vorschlags inkl. start_date); **Step 5: Commit** `feat: Vorschlaege mit Rhythmus, Start und Hinweis`.
+
+---
+
+### Task 3: Release v0.9.0 + Live-Gate gegen echte Daten
+
+- [x] **Step 1:** Suite final; `VERSION` → 0.9.0; Commit; `./create_pod_finance.sh`; `/api/version` == 0.9.0.
+- [x] **Step 2: Fable-Release-Gate (LIVE, lesend):** `GET /api/recurring/suggestions` gegen die echte DB. Prüfen: (a) KEINER der bestehenden ~41 Fixposten wird erneut vorgeschlagen (Bestandsabgleich wirkt, auch bei gedrifteten Beträgen); (b) keine bekannten toten Serien (gelöschte PayPal-−4,99-Serie, ausgelaufene WWK-−35-Police) im Ergebnis; (c) verbleibende Vorschläge einzeln gegen die Buchungen plausibilisieren (echte aktive Serie? korrekte Werte?). Ergebnisliste NUR im Chat/Bericht, nie committen. Bei Fehlklassifikationen: Befund zurück an Task 1 (Toleranzen), Fix + Re-Gate.
+- [x] **Step 3:** Ledger (generisch) + Plan-Häkchen + Push; Kandidatenliste dem Nutzer berichten.
+
+---
+
+## Self-Review
+
+- Spec-Abdeckung: Algorithmus/Schema → Task 1; GUI → Task 2; Release/Live-Gate → Task 3. Testfälle der Spec vollständig in Task 1 Step 1 kodiert.
+- Platzhalter: Task 1 Step 2 verweist bewusst auf den bindenden Spec-Abschnitt (8 nummerierte Schritte + Konstanten) statt den vollen Code zu duplizieren; alle Schnittstellen/Konstanten sind exakt benannt.
+- Typ-Konsistenz: `suggest_recurring(session, today)`-Signatur = Testaufrufe; `SuggestionOut`-Felder = Template-Zugriffe (`s.rhythm`, `s.start_date`, `s.hinweis`).

docs/superpowers/specs/2026-07-20-vorschlags-algorithmus-v2-design.md

@@ -0,0 +1,81 @@
+# Design — Ausbaustufe 9: Vorschlags-Algorithmus v2 (v0.9.0)
+
+> Status: vom Nutzer freigegeben (Chat 2026-07-20, Direktdurchlauf). Anlass:
+> Der bisherige `suggest_recurring` gruppiert nach exaktem Betrag (jede
+> Preiserhöhung zerreißt die Serie), erkennt nur monatliche Serien, prüft
+> keine Aktivität (schlägt tote Serien vor) und nutzt Median-Werte. Ein
+> manueller Vollabgleich (A8-Datenpflege, siehe Ledger) fand 16 fehlende
+> Posten — der Algorithmus soll solche Serien künftig selbst finden.
+
+## Anforderungen (Nutzer)
+
+- Betrag = **letzte** Buchung, nicht Median (Preissteigerungen relevant).
+- Erkennung **monatlich, vierteljährlich, jährlich**.
+- Empfänger-Gruppierung robust (Schreibweisen, Preisänderungen,
+  Umfirmierungen); mehrere Verträge desselben Anbieters getrennt.
+- Keine „Leichen": abgerissene Serien werden nicht vorgeschlagen.
+- Kein Wiedervorschlagen bereits gepflegter Posten (auch bei zwischenzeitlich
+  geändertem Betrag).
+
+## Algorithmus (`app/services/suggestions.py`, vollständiger Rewrite)
+
+Parameter als Modul-Konstanten (Toleranzen zentral änderbar):
+`WINDOW_DAYS=460` (~15 Monate), Rhythmen mit Intervallgrenzen und
+Mindestbelegen: monthly 25–36 Tage/≥3, quarterly 80–105/≥3, yearly
+330–400/≥2; `ACTIVITY_FACTOR=1.75`; Betrags-Cluster-Toleranz 35 %;
+Merge-Toleranz 25 %; Bestandsabgleich-Toleranz 10 %.
+
+1. **Datenbasis:** bestätigte Buchungen der letzten `WINDOW_DAYS`, je Konto.
+2. **Gruppierung:** Schlüssel = (account_id, normalisierter Empfänger)
+   (`casefold`, Whitespace kollabiert).
+3. **Betrags-Cluster** innerhalb der Gruppe (chronologisch, greedy gegen das
+   jeweils letzte Cluster-Mitglied, gleiches Vorzeichen, relative Differenz
+   ≤ 35 %) — trennt parallele Verträge, hält Preisdrift zusammen.
+4. **Rhythmus je Cluster:** Median der Buchungsabstände gegen die
+   Intervallgrenzen; Mindestbelege je Rhythmus.
+5. **Aktiv-Check:** letzte Buchung ≤ `ACTIVITY_FACTOR` × Rhythmus-Schrittweite
+   (30/91/365 Tage) her, sonst kein Vorschlag.
+6. **Umfirmierungs-Merge** (über Gruppengrenzen, je Konto): Serie A endet,
+   Serie B beginnt danach (Lücke 0,4–1,6 Schrittweiten), gleicher Rhythmus,
+   Fälligkeitstag ±3, Betrag ±25 % → eine Serie; Name/Betrag der neueren.
+7. **Vorschlagswerte:** Name = Empfänger-Schreibweise der neuesten Buchung;
+   Betrag = neueste Buchung; Fälligkeitstag = Tag der neuesten Buchung;
+   `start_date` = Datum der neuesten Buchung bei quarterly/yearly (Phase!),
+   sonst None; Kategorie = häufigste in der Serie; `hinweis` = Text
+   „Betrag zuletzt gestiegen (vorher X)" wenn die vorletzte Buchung
+   betragskleiner war, sonst leer.
+8. **Bestandsabgleich:** Vorschlag entfällt, wenn ein `RecurringItem`
+   existiert mit (a) Namens-Substring-Match (normalisiert, in beide
+   Richtungen) ODER (b) gleichem Rhythmus + Fälligkeitstag ±2 + Betrag
+   ±10 %.
+
+## API/GUI
+
+- `SuggestionOut` (routers/planning.py): + `start_date: date | None`,
+  + `hinweis: str = ""`.
+- Vorschlags-Tabelle (planning.html): Spalten Rhythmus (`|de_label`) und
+  Start (TT.MM.JJJJ bzw. „–"); `hinweis` als `muted`-Text hinter dem Betrag;
+  „Vorschlag übernehmen" überträgt `start_date` mit (hidden input).
+- Hinweistext unter der Tabelle aktualisiert: monatliche/vierteljährliche/
+  jährliche Serien, Betrag = letzte Buchung.
+
+## Tests (synthetische Daten, keine Fixtures)
+
+Preiserhöhungs-Serie → letzter Betrag + hinweis; Quartals-/Jahres-Serie mit
+korrektem start_date; tote Serie (letzte Buchung zu alt) → kein Vorschlag;
+Umbenennungs-Merge → ein Vorschlag mit neuem Namen; Bestandsabgleich:
+existierender Posten mit altem Betrag verhindert Wiedervorschlag; zwei
+parallele Verträge eines Anbieters → zwei getrennte Vorschläge; bestehende
+drei Suggestion-Tests an die neue Semantik anpassen.
+
+## Release
+
+`VERSION` → 0.9.0, Redeploy, **Live-Gate gegen echte Daten**: kein einziger
+der bestehenden Fixposten darf erneut vorgeschlagen werden; keine als
+beendet bekannten Serien (z.B. gelöschte PayPal-Leiche, ausgelaufene
+WWK-Police) im Ergebnis; verbleibende Vorschläge werden dem Nutzer als
+Kandidatenliste berichtet (nur Chat, kein Commit). Fable-Gate je Task.
+
+**Außerhalb des Scopes:** halbjährliche Rhythmen (nicht im Datenmodell),
+automatische Übernahme ohne Nutzer-Klick, Einnahmen-Prognose des
+Geschäftskontos.

finance/VERSION

@@ -1 +1 @@
-0.8.0
+0.9.0

finance/app/routers/planning.py

@@ -56,7 +56,9 @@ class SuggestionOut(BaseModel):
     amount: Decimal
     rhythm: str
     due_day: int
+    start_date: date | None = None
     category_id: int | None = None
+    hinweis: str = ""
 
 
 def _check_category(session: Session, category_id: int | None) -> None:

finance/app/services/suggestions.py

@@ -1,52 +1,369 @@
+"""Vorschlagsalgorithmus fuer wiederkehrende Buchungen (Ausbaustufe 9, v2).
+
+Ersetzt die reine exakte-Betrags-Gruppierung (v1) durch: Empfaenger-Cluster
+mit Toleranz (haelt Preisdrift in einer Serie zusammen, trennt aber parallele
+Vertraege desselben Anbieters), Rhythmus-Erkennung ueber den Median der
+Buchungsabstaende (monatlich/vierteljaehrlich/jaehrlich statt nur monatlich),
+einen Aktiv-Check (keine "Leichen"-Serien) sowie einen Merge-Pass fuer
+Umfirmierungen (Anbieter aendert den Namen, die Serie laeuft inhaltlich
+weiter). Bindende Spec:
+docs/superpowers/specs/2026-07-20-vorschlags-algorithmus-v2-design.md.
+
+Alle Betrags-Toleranzvergleiche verwenden ausschliesslich `Decimal`
+(CLAUDE.md: "Decimal, nicht float" - Rundungsfehler bei Geldbetraegen sind
+inakzeptabel). Tage-Vergleiche (Rhythmus, Aktiv-Check, Merge-Luecke) sind
+ganzzahlige Tage-Arithmetik, niemals float/Decimal-Bruchteile von Tagen.
+"""
+from __future__ import annotations
+
+import re
 import statistics
-from collections import Counter, defaultdict
+from collections import Counter
+from dataclasses import dataclass
+from datetime import date, timedelta
 from decimal import Decimal
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
+from app.formats import eur
 from app.models.tables import RecurringItem, Transaction
 
+# Betrachtungsfenster (Schritt 1): ~15 Monate. Muss mindestens die zwei
+# Belege einer jaehrlichen Serie (bis zu 400 Tage auseinander) plus etwas
+# Puffer fuer Cluster-/Merge-Bildung abdecken.
+WINDOW_DAYS = 460
 
-def _max_consecutive_months(months: list[tuple[int, int]]) -> int:
-    if not months:
-        return 0
-    best = current = 1
-    for prev, cur in zip(months, months[1:]):
-        prev_idx = prev[0] * 12 + prev[1]
-        cur_idx = cur[0] * 12 + cur[1]
-        current = current + 1 if cur_idx == prev_idx + 1 else 1
-        best = max(best, current)
-    return best
+# Rhythmus-Tabelle: (min_tage, max_tage, mindestbelege) je Rhythmus. Der
+# Median der Buchungsabstaende einer Serie muss ins Intervall fallen, UND es
+# muessen mindestens so viele Buchungen vorliegen (ein einzelner Zufallstreffer
+# mit "passendem" Abstand soll nicht als Serie gelten).
+RHYTHMS: dict[str, tuple[int, int, int]] = {
+    "monthly": (25, 36, 3),
+    "quarterly": (80, 105, 3),
+    "yearly": (330, 400, 2),
+}
+
+# Nominelle Schrittweite je Rhythmus in Tagen - Referenzwert fuer Aktiv-Check
+# und Merge-Luecken-Fenster (Schritt 5/6).
+STEP_DAYS: dict[str, int] = {"monthly": 30, "quarterly": 91, "yearly": 365}
+
+# Aktiv-Check (Schritt 5): die letzte Buchung darf hoechstens das 1,75-fache
+# der Rhythmus-Schrittweite zurueckliegen, sonst gilt die Serie als beendet
+# ("Leiche") und wird nicht vorgeschlagen. Als Fraction 7/4 ausgedrueckt und
+# ganzzahlig verglichen (delta_tage * 4 <= schrittweite * 7), damit keine
+# Gleitkomma-Rundung ueber "aktiv"/"inaktiv" entscheidet.
+ACTIVITY_FACTOR_NUM = 7
+ACTIVITY_FACTOR_DEN = 4
+
+# Relative Toleranzen (immer als Decimal verglichen, nie float):
+CLUSTER_TOL = Decimal("0.35")   # Schritt 3: Betrags-Cluster (haelt Preisdrift zusammen)
+MERGE_TOL = Decimal("0.25")     # Schritt 6: Umfirmierungs-Merge ueber Gruppengrenzen
+BESTAND_TOL = Decimal("0.10")   # Schritt 8: Bestandsabgleich gegen RecurringItem
+
+# Merge-Luecke (Schritt 6): die Zeit zwischen dem Ende von Serie A und dem
+# Beginn von Serie B muss zwischen dem 0,4- und 1,6-fachen der
+# Rhythmus-Schrittweite liegen (als ganzzahlige Bruchvergleiche, aus
+# demselben Grund wie beim Aktiv-Check).
+MERGE_GAP_MIN_NUM, MERGE_GAP_MIN_DEN = 4, 10   # 0.4
+MERGE_GAP_MAX_NUM, MERGE_GAP_MAX_DEN = 16, 10  # 1.6
+MERGE_DUE_DAY_TOL = 3    # Schritt 6: Faelligkeitstag-Toleranz in Tagen
+BESTAND_DUE_DAY_TOL = 2  # Schritt 8: Faelligkeitstag-Toleranz in Tagen
+
+# Bestandsabgleich, Token-Match (Live-Gate A9-Fund, Nachtrag 4): kuratierte
+# Fixposten tragen haeufig einen Alias-/Variabel-Namen, der weder Substring
+# noch betragsaehnlich zum automatisch erkannten Vorschlag ist (Beispiel aus
+# der echten Datenbasis: Fixposten "Mastercard-Abrechnung Volksbank
+# (variabel, letzter Wert)" vs. erkannter Vorschlag "Volksbank Ulm-Biberach
+# eG" - Betrag weicht um >10% ab, kein Substring-Treffer). Ein gemeinsames,
+# hinreichend spezifisches Namens-Token (>=5 Zeichen, um generische Woerter
+# wie "Bank" nicht faelschlich matchen zu lassen) bei gleichem Rhythmus und
+# nahem Faelligkeitstag gilt als ausreichendes Indiz fuer denselben Fixposten.
+TOKEN_MIN_LEN = 5
+
+# Volatilitaets-Hinweis (Live-Gate A9-Fund, Nachtrag 4): wenn der
+# Betrags-Cluster-Split (Schritt 3) die neueste Buchung der Empfaenger-Gruppe
+# abgetrennt hat (weil sie zu stark vom Serien-Betrag abweicht), ist der
+# vorgeschlagene Betrag ggf. schon wieder veraltet - keine Unterdrueckung,
+# nur ein Warnhinweis fuer die Nutzerin/den Nutzer.
+VOLATILITAETS_HINWEIS = "Beträge schwanken stark – letzte Buchung weicht ab"
 
 
-def suggest_recurring(session: Session) -> list[dict]:
+def _norm(name: str) -> str:
+    """Normalisiert einen Empfaenger-Namen fuer Gruppen- und
+    Substring-Vergleich: Bankexporte schreiben denselben Empfaenger nicht
+    einheitlich (Gross-/Kleinschreibung, mehrfache Leerzeichen), das ist fuer
+    die Erkennung irrelevant."""
+    return " ".join(name.split()).casefold()
+
+
+def _tokens(name: str) -> set[str]:
+    """Zerlegt einen normalisierten Namen an Nicht-Alphanumerik in Tokens
+    (fuer den Token-Match im Bestandsabgleich, Schritt 8). Nur Tokens ab
+    TOKEN_MIN_LEN Zeichen zaehlen, damit kurze generische Woerter ("eG",
+    "AG", "Bank") keine falschen Treffer erzeugen."""
+    return {tok for tok in re.split(r"[^a-z0-9]+", _norm(name)) if len(tok) >= TOKEN_MIN_LEN}
+
+
+def _rel_diff(a: Decimal, b: Decimal) -> Decimal:
+    """Relative Differenz von Betrag a zur Referenz b (immer >= 0), als
+    Decimal. b=0 kommt praktisch nicht vor (eine Nullbuchung bildet keine
+    erkennbare Serie); fuer diesen Sonderfall gilt "keine Aehnlichkeit"."""
+    if b == 0:
+        return Decimal("Infinity") if a != 0 else Decimal("0")
+    return abs(a - b) / abs(b)
+
+
+@dataclass
+class _Series:
+    """Eine erkannte Serie: chronologisch sortierte Buchungen eines
+    Betrags-Clusters mit zugeordnetem Rhythmus.
+
+    `volatile` markiert, dass der Cluster-Split (Schritt 3) innerhalb der
+    Empfaenger-Gruppe eine NEUERE, betragsmaessig abweichende Buchung
+    abgetrennt hat - der hier vorgeschlagene Betrag koennte also schon
+    wieder veraltet sein (siehe VOLATILITAETS_HINWEIS)."""
+    items: list[Transaction]
+    rhythm: str
+    volatile: bool = False
+
+    @property
+    def first(self) -> Transaction:
+        return self.items[0]
+
+    @property
+    def last(self) -> Transaction:
+        return self.items[-1]
+
+
+def _amount_clusters(items: list[Transaction]) -> list[list[Transaction]]:
+    """Schritt 3: teilt chronologisch sortierte Buchungen einer
+    Empfaenger-Gruppe in Betrags-Cluster. Eine Buchung haengt sich an das
+    Cluster, dessen zuletzt aufgenommenes Mitglied gleiches Vorzeichen und
+    eine relative Differenz <= CLUSTER_TOL hat (greedy, erstes passendes
+    Cluster gewinnt) - das haelt eine langsam driftende Serie (Preiserhoehung)
+    zusammen, trennt aber parallele Vertraege mit deutlich anderem Betrag."""
+    clusters: list[list[Transaction]] = []
+    for t in items:
+        for cluster in clusters:
+            last = cluster[-1]
+            same_sign = (t.amount > 0) == (last.amount > 0)
+            if same_sign and _rel_diff(Decimal(t.amount), Decimal(last.amount)) <= CLUSTER_TOL:
+                cluster.append(t)
+                break
+        else:
+            clusters.append([t])
+    return clusters
+
+
+def _classify(dates: list[date]) -> str | None:
+    """Schritt 4: bestimmt den Rhythmus einer Serie ueber den Median der
+    Buchungsabstaende (robust gegen einzelne Ausreisser, z.B.
+    Wochenend-/Feiertagsverschiebung einer einzelnen Buchung)."""
+    if len(dates) < 2:
+        return None
+    gaps = [(b - a).days for a, b in zip(dates, dates[1:])]
+    median_gap = statistics.median(gaps)
+    for rhythm, (lo, hi, min_belege) in RHYTHMS.items():
+        if len(dates) >= min_belege and lo <= median_gap <= hi:
+            return rhythm
+    return None
+
+
+def _merge_gap_ok(gap_days: int, step: int) -> bool:
+    """Schritt 6: Luecke zwischen Serienende und -beginn im Fenster
+    [0,4; 1,6] * Schrittweite (ganzzahliger Bruchvergleich, keine Rundung)."""
+    return (gap_days * MERGE_GAP_MIN_DEN >= MERGE_GAP_MIN_NUM * step
+            and gap_days * MERGE_GAP_MAX_DEN <= MERGE_GAP_MAX_NUM * step)
+
+
+def _mergeable(a: _Series, b: _Series) -> bool:
+    """Prueft die Umfirmierungs-Merge-Bedingungen aus Schritt 6 fuer ein
+    Paar (A endet, B beginnt danach): gleicher Rhythmus, plausible Luecke,
+    Faelligkeitstag nah beieinander (Transitionspunkte: letzte Buchung von A
+    gegen erste Buchung von B), Betrag nicht sprunghaft veraendert."""
+    if a.rhythm != b.rhythm:
+        return False
+    if a.last.booking_date >= b.first.booking_date:
+        return False
+    step = STEP_DAYS[a.rhythm]
+    gap = (b.first.booking_date - a.last.booking_date).days
+    if not _merge_gap_ok(gap, step):
+        return False
+    if abs(a.last.booking_date.day - b.first.booking_date.day) > MERGE_DUE_DAY_TOL:
+        return False
+    same_sign = (a.last.amount > 0) == (b.first.amount > 0)
+    if not same_sign:
+        return False
+    return _rel_diff(Decimal(b.first.amount), Decimal(a.last.amount)) <= MERGE_TOL
+
+
+def _try_merge(series_list: list[_Series]) -> list[_Series]:
+    """Schritt 6: fasst Serien desselben Kontos ueber Gruppengrenzen
+    (unterschiedlicher normalisierter Empfaenger-Name, z.B. nach einer
+    Umfirmierung) zusammen, solange `_mergeable` zutrifft. Laeuft iterativ
+    bis zum Fixpunkt, damit eine bereits gemergte Serie mit einer weiteren,
+    noch juengeren Serie erneut zusammengefasst werden kann (z.B. zwei
+    Umbenennungen hintereinander)."""
+    series_list = list(series_list)
+    changed = True
+    while changed:
+        changed = False
+        for i, a in enumerate(series_list):
+            for j, b in enumerate(series_list):
+                if i == j or not _mergeable(a, b):
+                    continue
+                merged = _Series(
+                    items=sorted(a.items + b.items, key=lambda t: t.booking_date),
+                    rhythm=a.rhythm,
+                    volatile=a.volatile or b.volatile,
+                )
+                series_list = [s for k, s in enumerate(series_list) if k not in (i, j)]
+                series_list.append(merged)
+                changed = True
+                break
+            if changed:
+                break
+    return series_list
+
+
+def _covered_by_existing(cand_name: str, cand_amount: Decimal, rhythm: str, due_day: int,
+                         existing: list[RecurringItem]) -> bool:
+    """Schritt 8 (Bestandsabgleich): ein Vorschlag entfaellt, wenn er bereits
+    als Fixposten gepflegt ist - ueber einen von drei Wegen:
+    (a) Namens-Substring-Match (normalisiert, in beide Richtungen: sowohl
+        Kurz- als auch Langschreibweisen kommen in der Praxis in beiden
+        Datenquellen vor);
+    (b) Rhythmus + Faelligkeitstag + Betrag innerhalb enger Toleranz (falls
+        der Fixposten unter einem ganz anderen Namen gepflegt wurde);
+    (c) Token-Match: gleicher Rhythmus, Faelligkeitstag-Differenz <= 2 UND
+        mindestens ein gemeinsames Namens-Token (>=5 Zeichen) - faengt
+        kuratierte Alias-/Variabel-Fixposten, deren Name UND Betrag stark
+        vom automatisch erkannten Vorschlag abweichen (Live-Gate A9-Fund:
+        Fixposten "Mastercard-Abrechnung Volksbank (variabel, letzter
+        Wert)" deckt den Vorschlag "Volksbank Ulm-Biberach eG" ab, obwohl
+        weder (a) noch (b) greifen)."""
+    cand_norm = _norm(cand_name)
+    cand_tokens = _tokens(cand_name)
+    for item in existing:
+        item_norm = _norm(item.name)
+        if cand_norm in item_norm or item_norm in cand_norm:
+            return True
+        if (item.rhythm == rhythm
+                and abs(item.due_day - due_day) <= BESTAND_DUE_DAY_TOL
+                and _rel_diff(cand_amount, Decimal(item.amount)) <= BESTAND_TOL):
+            return True
+        if (item.rhythm == rhythm
+                and abs(item.due_day - due_day) <= BESTAND_DUE_DAY_TOL
+                and cand_tokens & _tokens(item.name)):
+            return True
+    return False
+
+
+def suggest_recurring(session: Session, today: date | None = None) -> list[dict]:
+    """Ermittelt Vorschlaege fuer wiederkehrende Posten aus bestaetigten
+    Buchungen der letzten WINDOW_DAYS Tage. `today` ist ausschliesslich zu
+    Testzwecken injizierbar (deterministischer Aktiv-Check) - im
+    Produktivbetrieb liefert der Default `date.today()`. Reihenfolge der
+    Schritte gemaess Spec, mit einer bewussten Umstellung gegenueber der
+    Nummerierung dort: der Aktiv-Check (Schritt 5) laeuft NACH dem
+    Umfirmierungs-Merge (Schritt 6) auf der ggf. gemergten Serie - sonst
+    wuerde eine per Umfirmierung fortgesetzte Serie an ihrem alten,
+    laengst inaktiven Teil scheitern, bevor der Merge sie retten kann."""
+    if today is None:
+        today = date.today()
+    cutoff = today - timedelta(days=WINDOW_DAYS)
+
+    # Schritt 1: Datenbasis.
     txs = session.execute(
-        select(Transaction).where(Transaction.status == "confirmed")
+        select(Transaction)
+        .where(Transaction.status == "confirmed", Transaction.booking_date >= cutoff)
     ).scalars().all()
-    groups: dict[tuple, list[Transaction]] = defaultdict(list)
-    for t in txs:
-        groups[(t.account_id, t.counterparty, t.amount)].append(t)
 
-    existing = {(r.name, Decimal(r.amount))
-               for r in session.execute(select(RecurringItem)).scalars()}
+    # Schritt 2: Gruppierung je (Konto, normalisierter Empfaenger).
+    groups: dict[tuple[int, str], list[Transaction]] = {}
+    for t in txs:
+        groups.setdefault((t.account_id, _norm(t.counterparty)), []).append(t)
+
+    # Schritt 3+4: je Gruppe Betrags-Cluster bilden und Rhythmus klassifizieren.
+    series_by_account: dict[int, list[_Series]] = {}
+    for (account_id, _name_norm), items in groups.items():
+        items_sorted = sorted(items, key=lambda t: t.booking_date)
+        clusters = _amount_clusters(items_sorted)
+        classified = [(cluster, _classify([t.booking_date for t in cluster]))
+                     for cluster in clusters]
+        # Fuer den Volatilitaets-Check zaehlt eine neuere Buchung nur dann als
+        # "abgetrennt", wenn sie NICHT bereits zu einem ANDEREN qualifizierten
+        # (klassifizierten) Cluster derselben Gruppe gehoert - sonst waeren
+        # zwei parallele, stabile Vertraege (jeder fuer sich eine gueltige
+        # eigene Serie) faelschlich als "volatil" markiert, nur weil der
+        # jeweils andere Vertrag zufaellig spaeter im Monat faellig ist
+        # (Nachtrag 3b, Fable-Gate-Korrektur nach dem ersten Live-Gate-Fund).
+        qualified_items = {t for cluster, rhythm in classified if rhythm is not None
+                           for t in cluster}
+        for cluster, rhythm in classified:
+            if rhythm is None:
+                continue
+            # Volatilitaets-Hinweis: hat der Cluster-Split innerhalb DIESER
+            # Empfaenger-Gruppe (gleiches Konto, gleiches Vorzeichen) eine
+            # NEUERE Buchung in einen UNQUALIFIZIERTEN Cluster abgetrennt
+            # (z.B. eine einzelne Ausreisser-Buchung, die allein keine Serie
+            # bildet), ist der hier vorgeschlagene (letzte) Betrag ggf. schon
+            # veraltet.
+            cluster_sign = cluster[-1].amount > 0
+            volatile = any(
+                (t.amount > 0) == cluster_sign
+                and t.booking_date > cluster[-1].booking_date
+                and t not in qualified_items
+                for t in items_sorted
+            )
+            series_by_account.setdefault(account_id, []).append(
+                _Series(items=cluster, rhythm=rhythm, volatile=volatile))
+
+    existing = list(session.execute(select(RecurringItem)).scalars())
 
     suggestions: list[dict] = []
-    for (_account_id, counterparty, amount), items in groups.items():
-        months = sorted({(t.booking_date.year, t.booking_date.month) for t in items})
-        if _max_consecutive_months(months) < 3:
-            continue
-        name = counterparty
-        if (name, Decimal(amount)) in existing:
-            continue
-        due_day = int(statistics.median(sorted(t.booking_date.day for t in items)))
-        cat_counts = Counter(t.category_id for t in items if t.category_id is not None)
-        category_id = cat_counts.most_common(1)[0][0] if cat_counts else None
-        suggestions.append({
-            "name": name,
-            "amount": Decimal(amount),
-            "rhythm": "monthly",
-            "due_day": due_day,
-            "category_id": category_id,
-        })
+    for series_list in series_by_account.values():
+        # Schritt 6: Umfirmierungs-Merge ueber Gruppengrenzen, je Konto.
+        for s in _try_merge(series_list):
+            # Schritt 5: Aktiv-Check auf der (ggf. gemergten) finalen Serie.
+            step = STEP_DAYS[s.rhythm]
+            delta_tage = (today - s.last.booking_date).days
+            if delta_tage * ACTIVITY_FACTOR_DEN > step * ACTIVITY_FACTOR_NUM:
+                continue
+
+            # Schritt 7: Vorschlagswerte aus der neuesten Buchung.
+            last = s.last
+            name = last.counterparty
+            amount = Decimal(last.amount)
+            due_day = last.booking_date.day
+            start_date = last.booking_date if s.rhythm in ("quarterly", "yearly") else None
+            cat_counts = Counter(t.category_id for t in s.items if t.category_id is not None)
+            category_id = cat_counts.most_common(1)[0][0] if cat_counts else None
+
+            hinweis = ""
+            if len(s.items) >= 2:
+                previous = Decimal(s.items[-2].amount)
+                # "Gestiegen" bezieht sich auf den Betragswert (Ausgaben sind
+                # negativ: gestiegen heisst betragsmaessig groesser, also
+                # abs(neu) > abs(alt)), nicht auf das Vorzeichen.
+                if abs(amount) > abs(previous):
+                    hinweis = f"Betrag zuletzt gestiegen (vorher {eur(abs(previous))} €)"
+
+            if s.volatile:
+                hinweis = f"{hinweis} {VOLATILITAETS_HINWEIS}".strip()
+
+            # Schritt 8: Bestandsabgleich.
+            if _covered_by_existing(name, amount, s.rhythm, due_day, existing):
+                continue
+
+            suggestions.append({
+                "name": name,
+                "amount": amount,
+                "rhythm": s.rhythm,
+                "due_day": due_day,
+                "start_date": start_date,
+                "category_id": category_id,
+                "hinweis": hinweis,
+            })
     return suggestions

finance/app/templates/planning.html

@@ -88,15 +88,19 @@
     {% if suggestions %}
     <table>
       <thead>
-        <tr><th>Name</th><th>Betrag</th><th>Rhythmus</th><th>Fälligkeitstag</th><th></th></tr>
+        <tr><th>Name</th><th>Betrag</th><th>Rhythmus</th><th>Fälligkeitstag</th><th>Start</th><th></th></tr>
       </thead>
       <tbody>
         {% for s in suggestions %}
         <tr>
           <td>{{ s.name }}</td>
-          <td class="{{ 'neg' if s.amount < 0 else '' }}">{{ s.amount|eur }} €</td>
+          <td class="{{ 'neg' if s.amount < 0 else '' }}">
+            {{ s.amount|eur }} €
+            {% if s.hinweis %}<span class="muted">{{ s.hinweis }}</span>{% endif %}
+          </td>
           <td>{{ s.rhythm|de_label }}</td>
           <td>{{ s.due_day }}</td>
+          <td>{{ s.start_date.strftime('%d.%m.%Y') if s.start_date else '–' }}</td>
           <td>
             <form class="inline-form" hx-ext="json-form" hx-post="/api/recurring" hx-swap="none"
                   hx-on::after-request="if(event.detail.successful){window.location.reload()}">
@@ -105,6 +109,7 @@
               <input type="hidden" name="rhythm" value="{{ s.rhythm }}">
               <input type="hidden" name="due_day" data-type="int" value="{{ s.due_day }}">
               <input type="hidden" name="category_id" data-type="int" value="{{ s.category_id if s.category_id is not none else '' }}">
+              <input type="hidden" name="start_date" value="{{ s.start_date.isoformat() if s.start_date else '' }}">
               <button type="submit">Vorschlag übernehmen</button>
             </form>
           </td>
@@ -113,8 +118,9 @@
       </tbody>
     </table>
     {% else %}
-    <p class="muted">Keine Vorschläge — erkannt werden Serien aus mindestens 3 Monaten gleichartiger Buchungen.</p>
+    <p class="muted">Keine Vorschläge.</p>
     {% endif %}
+    <p class="muted">Erkannt werden monatliche, vierteljährliche und jährliche Serien; Betrag = jeweils letzte Buchung.</p>
   </fieldset>
 </section>
 

finance/tests/test_gui.py

@@ -246,6 +246,33 @@ def test_planning_page_shows_empty_suggestions_hint(client):
     assert r.status_code == 200
     assert "Vorschläge aus Buchungen" in r.text
     assert "Keine Vorschläge" in r.text
+    assert ("Erkannt werden monatliche, vierteljährliche und jährliche "
+            "Serien; Betrag = jeweils letzte Buchung.") in r.text
+
+
+def test_vorschlaege_zeigen_rhythmus_und_start(client, db):
+    # Synthetische vierteljaehrliche Serie relativ zu date.today(), da die
+    # Route suggest_recurring() ohne today-Injektion aufruft (echter
+    # Aktiv-Check gegen date.today()). Schrittweite ~91 Tage rueckwaerts,
+    # letzte Buchung 30 Tage vor heute (innerhalb des Aktiv-Fensters).
+    from app.models.tables import Transaction
+
+    client.post("/login", data={"username": "admin", "password": "geheim"})
+    acc = Account(bank="dkb", iban="DE-SUG-1", name="S", type="giro")
+    db.add(acc)
+    db.flush()
+    today = date.today()
+    booking_dates = [today - timedelta(days=d) for d in (303, 212, 121, 30)]
+    for d in booking_dates:
+        db.add(Transaction(account_id=acc.id, booking_date=d, amount=Decimal("-55.08"),
+                           purpose="p", counterparty="Rundfunk Synth", status="confirmed",
+                           dedup_hash=f"sug-{d.isoformat()}"))
+    db.commit()
+    r = client.get("/planung").text
+    letzte_buchung = booking_dates[-1]
+    assert "vierteljährlich" in r
+    assert letzte_buchung.strftime("%d.%m.%Y") in r
+    assert 'name="start_date"' in r
 
 
 def test_salden_page_stichtag_and_month_overview(client, db):

finance/tests/test_planning_api.py

@@ -1,8 +1,13 @@
 from datetime import date
 from decimal import Decimal
+from itertools import count
 
 from app.models.tables import Account, Category, RecurringItem, Transaction
-from app.services.suggestions import suggest_recurring
+from app.services.suggestions import _Series, _try_merge, suggest_recurring
+
+# Feste Vergleichs-"heute" fuer alle Vorschlags-Tests (Ausbaustufe 9): macht
+# den Aktiv-Check deterministisch, ohne echtes date.today() im Testlauf.
+TODAY = date(2026, 7, 20)
 
 H = {"Authorization": "Bearer test-key"}
 
@@ -72,42 +77,60 @@ def test_loan_in_scenario_keeps_balance_positive(client, db):
     assert Decimal(body["low_point_balance"]) > Decimal("0")
 
 
-def _tx(acc, d, amount, counterparty, dedup, category_id=None):
-    return Transaction(account_id=acc.id, booking_date=d, amount=Decimal(amount),
-                       purpose="", counterparty=counterparty, status="confirmed",
-                       dedup_hash=dedup, category_id=category_id)
+_iban_seq = count(1)
+
+
+def _acc(db) -> int:
+    """Legt ein Konto an und gibt dessen id zurueck. IBAN ist je Aufruf
+    eindeutig (falls ein Test mehrere Konten braucht), Praefix "DE" plus
+    laufende Nummer reicht dafuer aus."""
+    acc = Account(bank="dkb", iban=f"DE{next(_iban_seq):032d}"[:34], name="G", type="giro")
+    db.add(acc)
+    db.flush()
+    return acc.id
+
+
+def _tx(db, acc_id, d, amount, cp, cat=None):
+    """Legt eine bestaetigte Buchung an. dedup_hash aus den Nutzdaten
+    abgeleitet reicht fuer Testzwecke (muss nur innerhalb eines Tests
+    eindeutig sein)."""
+    db.add(Transaction(account_id=acc_id, booking_date=d, amount=Decimal(amount),
+                       purpose="p", counterparty=cp, category_id=cat,
+                       status="confirmed", dedup_hash=f"h-{cp}-{d}-{amount}"))
 
 
 def test_suggest_recurring_three_consecutive_months_with_year_wrap(db):
-    acc = Account(bank="dkb", iban="DE01", name="G", type="giro")
-    db.add(acc)
-    db.flush()
+    acc = _acc(db)
     cat = Category(name="Miete")
     db.add(cat)
     db.flush()
     # Dez 2025 -> Jan 2026 -> Feb 2026: 3 aufeinanderfolgende Monate ueber den
-    # Jahreswechsel hinweg (prueft die Monats-Linearisierung y*12+m).
-    db.add(_tx(acc, date(2025, 12, 1), "-600.00", "Vermieter", "h1", cat.id))
-    db.add(_tx(acc, date(2026, 1, 15), "-600.00", "Vermieter", "h2", cat.id))
-    db.add(_tx(acc, date(2026, 2, 28), "-600.00", "Vermieter", "h3", cat.id))
+    # Jahreswechsel hinweg, gleichmaessiger 31-Tage-Abstand (Median im
+    # monatlichen Fenster 25-36 Tage; die alte Jahreswechsel-Pruefung galt der
+    # Monats-Linearisierung, die es im neuen Tage-basierten Median-Ansatz
+    # nicht mehr braucht).
+    _tx(db, acc, date(2025, 12, 1), "-600.00", "Vermieter", cat.id)
+    _tx(db, acc, date(2026, 1, 1), "-600.00", "Vermieter", cat.id)
+    _tx(db, acc, date(2026, 2, 1), "-600.00", "Vermieter", cat.id)
     db.commit()
 
-    out = suggest_recurring(db)
+    out = suggest_recurring(db, today=date(2026, 2, 10))
 
-    # Median der Tage [1, 15, 28] = 15; Betrag unveraendert uebernommen.
+    # due_day = Tag der NEUESTEN Buchung (nicht mehr Median, Spec Schritt 7);
+    # Betrag unveraendert, keine Preissteigerung -> hinweis leer.
     assert out == [{"name": "Vermieter", "amount": Decimal("-600.00"),
-                    "rhythm": "monthly", "due_day": 15, "category_id": cat.id}]
+                    "rhythm": "monthly", "due_day": 1, "start_date": None,
+                    "category_id": cat.id, "hinweis": ""}]
 
 
 def test_suggest_recurring_two_months_no_suggestion(db):
-    acc = Account(bank="dkb", iban="DE01", name="G", type="giro")
-    db.add(acc)
-    db.flush()
-    db.add(_tx(acc, date(2026, 3, 10), "-50.00", "Zweimonatig", "h1"))
-    db.add(_tx(acc, date(2026, 4, 10), "-50.00", "Zweimonatig", "h2"))
+    acc = _acc(db)
+    _tx(db, acc, date(2026, 3, 10), "-50.00", "Zweimonatig")
+    _tx(db, acc, date(2026, 4, 10), "-50.00", "Zweimonatig")
     db.commit()
 
-    assert suggest_recurring(db) == []
+    # Nur 2 Belege: Mindestbelege fuer monthly (>=3) nicht erreicht.
+    assert suggest_recurring(db, today=date(2026, 4, 20)) == []
 
 
 def test_recurring_ende_vor_start_wird_abgelehnt(client):
@@ -135,15 +158,212 @@ def test_recurring_start_ende_roundtrip_und_patch_validierung(client):
 
 
 def test_suggest_recurring_excludes_existing_recurring_item(db):
-    acc = Account(bank="dkb", iban="DE01", name="G", type="giro")
-    db.add(acc)
-    db.flush()
-    db.add(_tx(acc, date(2026, 1, 5), "-30.00", "Streaming", "h1"))
-    db.add(_tx(acc, date(2026, 2, 5), "-30.00", "Streaming", "h2"))
-    db.add(_tx(acc, date(2026, 3, 5), "-30.00", "Streaming", "h3"))
+    acc = _acc(db)
+    _tx(db, acc, date(2026, 1, 5), "-30.00", "Streaming")
+    _tx(db, acc, date(2026, 2, 5), "-30.00", "Streaming")
+    _tx(db, acc, date(2026, 3, 5), "-30.00", "Streaming")
     db.add(RecurringItem(name="Streaming", amount=Decimal("-30.00"),
                          rhythm="monthly", due_day=5))
     db.commit()
 
-    # Gleicher Name + Betrag wie ein bereits vorhandenes RecurringItem -> ausgelassen.
-    assert suggest_recurring(db) == []
+    # Gleicher Name wie ein bereits vorhandenes RecurringItem -> Bestandsabgleich
+    # (Schritt 8, Namens-Match) greift, unabhaengig vom Betrag.
+    assert suggest_recurring(db, today=date(2026, 3, 20)) == []
+
+
+# --------------------------------------------------- Ausbaustufe 9: Algorithmus v2
+# Synthetische Faelle aus der Spec (siehe
+# docs/superpowers/specs/2026-07-20-vorschlags-algorithmus-v2-design.md).
+
+def test_suggest_letzter_betrag_bei_preiserhoehung(db):
+    acc = _acc(db)
+    for d, a in [(date(2026, 3, 1), "-190.65"), (date(2026, 4, 1), "-202.94"),
+                 (date(2026, 5, 4), "-202.94"), (date(2026, 6, 1), "-202.94"),
+                 (date(2026, 7, 1), "-202.94")]:
+        _tx(db, acc, d, a, "Entis Lebensversicherung AG")
+    db.commit()
+    out = suggest_recurring(db, today=TODAY)
+    assert len(out) == 1
+    s = out[0]
+    assert s["amount"] == Decimal("-202.94") and s["rhythm"] == "monthly"
+    assert s["due_day"] == 1 and s["start_date"] is None
+
+
+def test_suggest_quartal_mit_phase(db):
+    acc = _acc(db)
+    for d in [date(2025, 9, 15), date(2025, 12, 15), date(2026, 3, 16), date(2026, 6, 15)]:
+        _tx(db, acc, d, "-55.08", "Rundfunk ARD ZDF")
+    db.commit()
+    out = suggest_recurring(db, today=TODAY)
+    assert len(out) == 1
+    assert out[0]["rhythm"] == "quarterly"
+    assert out[0]["start_date"] == date(2026, 6, 15) and out[0]["due_day"] == 15
+
+
+def test_suggest_jahr_mit_zwei_belegen(db):
+    acc = _acc(db)
+    for d, a in [(date(2025, 6, 16), "-409.92"), (date(2026, 6, 16), "-467.33")]:
+        _tx(db, acc, d, a, "Kraftfahrer-Schutz e.V.")
+    db.commit()
+    out = suggest_recurring(db, today=TODAY)
+    assert len(out) == 1
+    assert out[0]["rhythm"] == "yearly" and out[0]["amount"] == Decimal("-467.33")
+    assert out[0]["start_date"] == date(2026, 6, 16)
+    assert "409,92" in out[0]["hinweis"]  # Betrag zuletzt gestiegen (deutsches Format)
+
+
+def test_suggest_tote_serie_kein_vorschlag(db):
+    acc = _acc(db)
+    for m in (9, 10, 11, 12):
+        _tx(db, acc, date(2025, m, 1), "-35.00", "WWK Alt")
+    db.commit()
+    assert suggest_recurring(db, today=TODAY) == []
+
+
+def test_suggest_umfirmierung_merge(db):
+    acc = _acc(db)
+    for m in (11, 12):
+        _tx(db, acc, date(2025, m, 1), "-190.65", "Heidelberger Leben")
+    for m in (1, 2, 3):
+        _tx(db, acc, date(2026, m, 2), "-190.65", "Entis Lebensversicherung")
+    for m in (4, 5, 6, 7):
+        _tx(db, acc, date(2026, m, 1), "-202.94", "Entis Lebensversicherung")
+    db.commit()
+    out = suggest_recurring(db, today=TODAY)
+    assert len(out) == 1
+    assert "Entis" in out[0]["name"] and out[0]["amount"] == Decimal("-202.94")
+
+
+def test_suggest_bestandsabgleich_trotz_preisdrift(db):
+    acc = _acc(db)
+    db.add(RecurringItem(name="Entis Lebensversicherung AG", amount=Decimal("-190.65"),
+                         rhythm="monthly", due_day=1))
+    for m in (4, 5, 6, 7):
+        _tx(db, acc, date(2026, m, 1), "-202.94", "Entis Lebensversicherung AG")
+    db.commit()
+    assert suggest_recurring(db, today=TODAY) == []  # Namens-Match schlaegt an
+
+
+def test_suggest_zwei_vertraege_getrennt(db):
+    acc = _acc(db)
+    for m in (4, 5, 6, 7):
+        _tx(db, acc, date(2026, m, 1), "-346.23", "Heidelberger LV")
+        _tx(db, acc, date(2026, m, 2), "-145.21", "Heidelberger LV")
+    db.commit()
+    out = suggest_recurring(db, today=TODAY)
+    assert len(out) == 2
+    assert {s["amount"] for s in out} == {Decimal("-346.23"), Decimal("-145.21")}
+    # Nachtrag 3b: beide Vertraege sind fuer sich genommen stabile,
+    # qualifizierte Serien - die jeweils neuere Buchung des ANDEREN Vertrags
+    # gehoert selbst zu einer qualifizierten Serie und darf deshalb NICHT als
+    # "abgetrennte neueste Buchung" gewertet werden (sonst waere einer der
+    # beiden faelschlich als "volatil" markiert, nur weil der andere Vertrag
+    # einen Tag spaeter faellig ist).
+    assert all(s["hinweis"] == "" for s in out)
+
+
+def test_try_merge_kombiniert_serien_ueber_gruppengrenzen():
+    # Direkter, isolierter Test der Merge-Mechanik (Schritt 6): siehe Report
+    # fuer den rechnerischen Nachweis, dass ein End-to-End-Szenario, in dem
+    # die ALTE und die NEUE Serie GLEICHZEITIG unabhaengig voneinander den
+    # Aktiv-Check bestehen, fuer keinen der drei Rhythmen innerhalb von
+    # WINDOW_DAYS=460 konstruierbar ist (die alte Serie ist beim Aktiv-Check
+    # immer laengst "tot", sobald die neue genug eigene Belege hat, bzw. bei
+    # yearly passt die noetige Gesamtspanne nicht ins Fenster). Deshalb hier
+    # `_try_merge` direkt gegen zwei synthetische `_Series` geprueft, ganz ohne
+    # DB/Fenster/Aktiv-Check-Interaktion.
+    a = _Series(items=[
+        Transaction(booking_date=date(2026, 1, 3), amount=Decimal("-50.00"),
+                   counterparty="Alte Firma GmbH", category_id=None),
+        Transaction(booking_date=date(2026, 2, 3), amount=Decimal("-50.00"),
+                   counterparty="Alte Firma GmbH", category_id=None),
+        Transaction(booking_date=date(2026, 3, 3), amount=Decimal("-50.00"),
+                   counterparty="Alte Firma GmbH", category_id=None),
+    ], rhythm="monthly")
+    b = _Series(items=[
+        Transaction(booking_date=date(2026, 4, 5), amount=Decimal("-52.00"),
+                   counterparty="Neue Firma GmbH", category_id=None),
+        Transaction(booking_date=date(2026, 5, 5), amount=Decimal("-52.00"),
+                   counterparty="Neue Firma GmbH", category_id=None),
+        Transaction(booking_date=date(2026, 6, 5), amount=Decimal("-52.00"),
+                   counterparty="Neue Firma GmbH", category_id=None),
+    ], rhythm="monthly")
+
+    merged = _try_merge([a, b])
+
+    # Luecke A-Ende->B-Anfang = 33 Tage (in [12,48]), Faelligkeitstag 3 vs 5
+    # (Differenz 2 <= 3), Betrag +4% (<=25%) -> Bedingungen erfuellt, genau
+    # EINE kombinierte Serie mit allen 6 Buchungen, juengste zuerst.
+    assert len(merged) == 1
+    assert len(merged[0].items) == 6
+    assert merged[0].last.counterparty == "Neue Firma GmbH"
+
+
+def test_suggest_umfirmierung_merge_verschiebt_kategorie_mehrheit(db):
+    # End-to-End-Nachweis, dass Schritt 6 tatsaechlich in `suggest_recurring`
+    # verdrahtet ist: da die Anzahl der Vorschlaege sich (bewiesenermassen,
+    # siehe Report) end-to-end NICHT als Diskriminator eignet (die alte Serie
+    # faellt so oder so per Aktiv-Check heraus), wird hier die
+    # Kategorie-Mehrheit als Diskriminator genutzt - die haengt direkt davon
+    # ab, ob die Buchungen der alten Serie ueber den Merge in die Zaehlung
+    # eingehen. Alte Serie: 4 Buchungen Kategorie A. Neue Serie: 3 Buchungen
+    # Kategorie B. Ohne Merge zaehlen nur die 3 B-Buchungen (Mehrheit B). Mit
+    # Merge kommen die 4 A-Buchungen dazu und kippen die Mehrheit auf A.
+    acc = _acc(db)
+    cat_a = Category(name="Alt-Kategorie")
+    cat_b = Category(name="Neu-Kategorie")
+    db.add(cat_a)
+    db.add(cat_b)
+    db.flush()
+    for d in [date(2026, 1, 3), date(2026, 2, 3), date(2026, 3, 3), date(2026, 4, 3)]:
+        _tx(db, acc, d, "-50.00", "Alte Firma GmbH", cat_a.id)
+    for d in [date(2026, 5, 5), date(2026, 6, 5), date(2026, 7, 5)]:
+        _tx(db, acc, d, "-52.00", "Neue Firma GmbH", cat_b.id)
+    db.commit()
+
+    out = suggest_recurring(db, today=date(2026, 7, 20))
+
+    assert len(out) == 1
+    assert out[0]["name"] == "Neue Firma GmbH" and out[0]["amount"] == Decimal("-52.00")
+    # Kategorie-Mehrheit kippt durch den Merge von B (3) auf A (4):
+    assert out[0]["category_id"] == cat_a.id
+
+
+# ------------------------------------------------- Nachtrag 4 (A9-Live-Gate-Fund)
+# Echter Fall aus dem Live-Gate: Fixposten 44 "Mastercard-Abrechnung Volksbank
+# (variabel, letzter Wert)" (monthly, due_day 7, -296.07) deckte den Vorschlag
+# "Volksbank Ulm-Biberach eG" (monthly, due_day 5, -584.43) nicht ab, weil
+# weder Substring- noch Betrags-Toleranz-Regel griffen. Betraege/Namen hier
+# synthetisch nachgebildet (keine echten Kontodaten, siehe CLAUDE.md).
+
+def test_suggest_alias_recurring_item_token_match(db):
+    acc = _acc(db)
+    db.add(RecurringItem(name="Mastercard-Abrechnung Volksbank (variabel, letzter Wert)",
+                         amount=Decimal("-296.07"), rhythm="monthly", due_day=7))
+    for d, a in [(date(2026, 4, 5), "-560.00"), (date(2026, 5, 5), "-580.00"),
+                 (date(2026, 6, 5), "-590.00"), (date(2026, 7, 5), "-580.00")]:
+        _tx(db, acc, d, a, "Volksbank Ulm-Biberach eG")
+    db.commit()
+
+    # Substring-Match (a) schlaegt fehl (kein Teilstring gemeinsam), Betrags-
+    # Toleranz (b) auch (-580 vs. -296.07, >10%) - erst der Token-Match (c)
+    # ueber das gemeinsame Token "volksbank" (Rhythmus gleich, due_day 5 vs. 7
+    # -> Differenz 2 <= 2) deckt den Vorschlag ab.
+    assert suggest_recurring(db, today=TODAY) == []
+
+
+def test_suggest_volatilitaetshinweis_bei_abgetrennter_neuester_buchung(db):
+    acc = _acc(db)
+    for d in [date(2026, 1, 5), date(2026, 2, 5), date(2026, 3, 5), date(2026, 4, 5)]:
+        _tx(db, acc, d, "-200.00", "Schwankender Anbieter GmbH")
+    # Neueste Buchung weicht >35% vom Serien-Betrag ab -> eigener Cluster,
+    # klassifiziert selbst nicht (nur 1 Buchung) -> die vorgeschlagene Serie
+    # bleibt die -200.00-Serie, aber mit Volatilitaets-Warnhinweis.
+    _tx(db, acc, date(2026, 5, 5), "-600.00", "Schwankender Anbieter GmbH")
+    db.commit()
+
+    out = suggest_recurring(db, today=date(2026, 5, 20))
+
+    assert len(out) == 1
+    assert out[0]["amount"] == Decimal("-200.00")  # Betrag NICHT durch die 600er-Buchung verfaelscht
+    assert "schwanken" in out[0]["hinweis"].lower()