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 _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"} def _seed_balance(db, amount="1000.00"): # Portiert von Statement-closing-Logik auf Konto-Saldo-Anker (Ausbaustufe 3 # Task 1): der Anker am Konto uebernimmt die Rolle der frueheren # Statement.closing_balance als Saldo-Basis. acc = Account(bank="dkb", iban="DE01", name="G", type="giro") db.add(acc) db.flush() acc.anchor_date = date(2026, 6, 30) acc.anchor_balance = Decimal(amount) db.commit() def test_scenario_projection(client, db): _seed_balance(db) client.post("/api/recurring", headers=H, json={ "name": "Miete", "amount": "-600.00", "rhythm": "monthly", "due_day": 1}) sc = client.post("/api/scenarios", headers=H, json={"name": "Basis"}).json() r = client.post(f"/api/scenarios/{sc['id']}/project", headers=H, params={"horizon_days": 92, "start_date": "2026-07-15"}) assert r.status_code == 200 body = r.json() # 3 Mietzahlungen (01.08., 01.09., 01.10.): 400 -> -200 -> -800 assert Decimal(body["low_point_balance"]) == Decimal("-800.00") assert body["below_zero_date"] == "2026-09-01" assert len(body["series"]) == 92 def test_projektion_mit_ende_modifier_und_szenario_einmalzahlung(client, db): _seed_balance(db) rec = client.post("/api/recurring", headers=H, json={ "name": "Miete-Proj", "amount": "-1000.00", "rhythm": "monthly", "due_day": 1}).json() sc = client.post("/api/scenarios", headers=H, json={"name": "BestCase-Test", "include_planned": False}).json() client.post(f"/api/scenarios/{sc['id']}/modifiers", headers=H, json={ "target_type": "recurring", "target_id": rec["id"], "kind": "ende", "end_date": "2026-08-31"}) client.post(f"/api/scenarios/{sc['id']}/planned", headers=H, json={ "name": "Zufluss", "amount": "5000.00", "due": "2026-07-30"}) r = client.post(f"/api/scenarios/{sc['id']}/project", headers=H, params={"horizon_days": 150, "start_date": "2026-07-15"}) assert r.status_code == 200 series = {p["day"]: Decimal(p["balance"]) for p in r.json()["series"]} # Start 1000; +5000 am 30.07.; Miete nur noch am 01.08. (ende 31.08.); # include_planned=False, aber die SZENARIO-Zahlung zaehlt trotzdem. assert series["2026-07-30"] == Decimal("6000.00") assert series["2026-08-01"] == Decimal("5000.00") assert series["2026-12-01"] == Decimal("5000.00") # keine Miete mehr ab Sept. def test_loan_in_scenario_keeps_balance_positive(client, db): _seed_balance(db) client.post("/api/recurring", headers=H, json={ "name": "Miete", "amount": "-600.00", "rhythm": "monthly", "due_day": 1}) loan = client.post("/api/loans", headers=H, json={ "name": "K1", "principal": "5000.00", "annual_rate_pct": "6.0", "term_months": 48, "payout_date": "2026-07-20", "repayment_type": "annuity"}).json() sc = client.post("/api/scenarios", headers=H, json={"name": "Kredit"}).json() client.post(f"/api/scenarios/{sc['id']}/loans/{loan['id']}", headers=H) body = client.post(f"/api/scenarios/{sc['id']}/project", headers=H, params={"horizon_days": 92, "start_date": "2026-07-15"}).json() assert Decimal(body["low_point_balance"]) > Decimal("0") _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 = _acc(db) cat = Category(name="Miete") db.add(cat) db.flush() # Dez 2025 -> Jan 2026 -> Feb 2026: 3 aufeinanderfolgende Monate ueber den # 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, today=date(2026, 2, 10)) # 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": 1, "start_date": None, "category_id": cat.id, "hinweis": ""}] def test_suggest_recurring_two_months_no_suggestion(db): 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() # 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): r = client.post("/api/recurring", headers=H, json={ "name": "Falschrum", "amount": "-10.00", "rhythm": "monthly", "due_day": 1, "start_date": "2026-10-01", "end_date": "2026-09-01"}) assert r.status_code == 422 def test_recurring_start_ende_roundtrip_und_patch_validierung(client): r = client.post("/api/recurring", headers=H, json={ "name": "Befristet", "amount": "-10.00", "rhythm": "monthly", "due_day": 1, "start_date": "2026-01-01", "end_date": "2026-12-31"}) assert r.status_code == 201 item = r.json() assert item["start_date"] == "2026-01-01" and item["end_date"] == "2026-12-31" # PATCH, der Ende vor den (bestehenden) Start schieben will -> 422: bad = client.patch(f"/api/recurring/{item['id']}", headers=H, json={"end_date": "2025-06-30"}) assert bad.status_code == 422 # Datum explizit loeschen (null) ist erlaubt: ok = client.patch(f"/api/recurring/{item['id']}", headers=H, json={"end_date": None}) assert ok.status_code == 200 and ok.json()["end_date"] is None def test_suggest_recurring_excludes_existing_recurring_item(db): 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 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 (Live-Gate-Fund) # Live-Gate-Fund (Muster, keine echten Kontodaten - Namen/Betraege hier rein # synthetisch): ein kuratiertes Sammel-Fixposten unter Alias-Namen des # Anbieters ("Kreditkarten-Abrechnung ... (variabel)") deckte den vom # Algorithmus erkannten Vorschlag desselben Anbieters unter dessen # regulaerem Empfaenger-Namen nicht ab, weil weder Substring- noch # Betrags-Toleranz-Regel griffen. def test_suggest_alias_recurring_item_token_match(db): acc = _acc(db) db.add(RecurringItem(name="Kreditkarten-Abrechnung Musterbank (variabel)", amount=Decimal("-250.00"), rhythm="monthly", due_day=7)) for d, a in [(date(2026, 4, 5), "-560.00"), (date(2026, 5, 5), "-575.00"), (date(2026, 6, 5), "-590.00"), (date(2026, 7, 5), "-575.00")]: _tx(db, acc, d, a, "Musterbank Neustadt eG") db.commit() # Substring-Match (a) schlaegt fehl (kein Teilstring gemeinsam), Betrags- # Toleranz (b) auch (-575 vs. -250.00, >10%) - erst der Token-Match (c) # ueber das gemeinsame Token "musterbank" (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()