from datetime import date from decimal import Decimal from app.engine.loans import loan_schedule from app.engine.scenario import (PlainModifier, PlainPlanned, PlainRecurring, build_cashflows) W = (date(2026, 8, 1), date(2026, 10, 31)) def test_recurring_with_percent_cut(): rec = [PlainRecurring(id=1, name="Marketing", amount=Decimal("-1000"), rhythm="monthly", due_day=5, start_date=None, end_date=None, category_id=7)] mods = [PlainModifier(target_type="category", target_id=7, kind="percent", value=Decimal("50"))] flows = build_cashflows(rec, [], [], [], mods, *W) assert flows == [(date(2026, 8, 5), Decimal("-500.00")), (date(2026, 9, 5), Decimal("-500.00")), (date(2026, 10, 5), Decimal("-500.00"))] def test_remove_modifier_and_planned_and_loan(): rec = [PlainRecurring(id=2, name="Abo", amount=Decimal("-50"), rhythm="monthly", due_day=1, start_date=None, end_date=None, category_id=None)] mods = [PlainModifier(target_type="recurring", target_id=2, kind="remove", value=Decimal("0"))] planned = [PlainPlanned(name="Steuer", amount=Decimal("-2000"), due=date(2026, 9, 15), category_id=None)] sched = loan_schedule(Decimal("10000"), Decimal("6"), 48, date(2026, 8, 1), "annuity") flows = build_cashflows(rec, planned, [sched], [(date(2026, 8, 1), Decimal("10000"))], mods, *W) days = [d for d, _ in flows] assert date(2026, 8, 1) in days # Kredit-Auszahlung assert date(2026, 9, 15) in days # Einmalposten assert all(a != Decimal("-50") for _, a in flows) # Abo entfernt def test_absolute_modifier_reduces_outflow(): """Test absolute modifier for outflows: -200 - 50 = -150""" rec = [PlainRecurring(id=1, name="Marketing", amount=Decimal("-200"), rhythm="monthly", due_day=10, start_date=None, end_date=None, category_id=3)] mods = [PlainModifier(target_type="category", target_id=3, kind="absolute", value=Decimal("50"))] flows = build_cashflows(rec, [], [], [], mods, date(2026, 8, 1), date(2026, 8, 31)) assert flows == [(date(2026, 8, 10), Decimal("-150.00"))] def test_absolute_modifier_clamps_to_zero(): """Test absolute modifier clamping: -200 + 300 = 0 (never flips sign)""" rec = [PlainRecurring(id=1, name="Marketing", amount=Decimal("-200"), rhythm="monthly", due_day=10, start_date=None, end_date=None, category_id=3)] mods = [PlainModifier(target_type="category", target_id=3, kind="absolute", value=Decimal("300"))] flows = build_cashflows(rec, [], [], [], mods, date(2026, 8, 1), date(2026, 8, 31)) assert flows == [(date(2026, 8, 10), Decimal("0.00"))] def test_ende_modifier_beendet_posten_im_szenario(): r = PlainRecurring(1, "Miete", Decimal("-2000.00"), "monthly", 1, None, None, None) m = PlainModifier("recurring", 1, "ende", Decimal("0"), date(2026, 8, 31)) flows = build_cashflows([r], [], [], [], [m], date(2026, 7, 1), date(2026, 12, 31)) assert [d for d, _ in flows] == [date(2026, 7, 1), date(2026, 8, 1)] assert all(a == Decimal("-2000.00") for _, a in flows) def test_ende_modifier_minimum_mit_eigenem_ende(): # Posten endet selbst am 15.10.; Szenario-Ende 31.08. ist frueher -> gewinnt. r = PlainRecurring(1, "Miete", Decimal("-100.00"), "monthly", 1, None, date(2026, 10, 15), None) m = PlainModifier("recurring", 1, "ende", Decimal("0"), date(2026, 8, 31)) flows = build_cashflows([r], [], [], [], [m], date(2026, 7, 1), date(2026, 12, 31)) assert [d for d, _ in flows] == [date(2026, 7, 1), date(2026, 8, 1)] # Umgekehrt: Szenario-Ende NACH eigenem Ende aendert nichts. m_spaet = PlainModifier("recurring", 1, "ende", Decimal("0"), date(2026, 12, 31)) flows2 = build_cashflows([r], [], [], [], [m_spaet], date(2026, 7, 1), date(2026, 12, 31)) assert [d for d, _ in flows2][-1] == date(2026, 10, 1) def test_ende_modifier_auf_kategorie_trifft_alle_posten_der_kategorie(): r1 = PlainRecurring(1, "Miete A", Decimal("-10.00"), "monthly", 1, None, None, 7) r2 = PlainRecurring(2, "Miete B", Decimal("-20.00"), "monthly", 1, None, None, 7) r3 = PlainRecurring(3, "Strom", Decimal("-5.00"), "monthly", 1, None, None, 9) m = PlainModifier("category", 7, "ende", Decimal("0"), date(2026, 7, 31)) flows = build_cashflows([r1, r2, r3], [], [], [], [m], date(2026, 7, 1), date(2026, 9, 30)) betraege = sorted(a for _, a in flows) # Miete A/B nur je 1x (Juli), Strom 3x (Juli-Sept). assert betraege == [Decimal("-20.00"), Decimal("-10.00"), Decimal("-5.00"), Decimal("-5.00"), Decimal("-5.00")] def test_ende_modifier_ohne_datum_wirkt_nicht(): r = PlainRecurring(1, "Miete", Decimal("-10.00"), "monthly", 1, None, None, None) m = PlainModifier("recurring", 1, "ende", Decimal("0"), None) flows = build_cashflows([r], [], [], [], [m], date(2026, 7, 1), date(2026, 9, 30)) assert len(flows) == 3 def test_ende_modifier_aendert_betraege_nicht(): r = PlainRecurring(1, "Miete", Decimal("-2000.00"), "monthly", 1, None, None, None) m = PlainModifier("recurring", 1, "ende", Decimal("50"), date(2026, 12, 31)) flows = build_cashflows([r], [], [], [], [m], date(2026, 7, 1), date(2026, 8, 31)) assert all(a == Decimal("-2000.00") for _, a in flows) D = Decimal def _rec(amount="-100.00", due_day=14, id=1, cat=None): return PlainRecurring(id, "R", D(amount), "monthly", due_day, None, None, cat) def test_zeitraum_grenzen_inklusiv(): mods = [PlainModifier("recurring", 1, "percent", D("50"), end_date=date(2030, 3, 14), start_date=date(2030, 2, 14))] flows = build_cashflows([_rec()], [], [], [], mods, date(2030, 1, 1), date(2030, 4, 30)) assert flows == [(date(2030, 1, 14), D("-100.00")), (date(2030, 2, 14), D("-50.00")), (date(2030, 3, 14), D("-50.00")), (date(2030, 4, 14), D("-100.00"))] def test_ab_stichtag_erst_folgemonat(): # Fällig am 14., gültig ab 16.10. -> Oktober voll, ab November halbiert. mods = [PlainModifier("recurring", 1, "percent", D("50"), start_date=date(2030, 10, 16))] flows = build_cashflows([_rec("-200.00")], [], [], [], mods, date(2030, 10, 1), date(2030, 12, 31)) assert flows == [(date(2030, 10, 14), D("-200.00")), (date(2030, 11, 14), D("-100.00")), (date(2030, 12, 14), D("-100.00"))] def test_neuer_betrag_dann_prozent(): mods = [PlainModifier("recurring", 1, "set", D("-80.00"), start_date=date(2030, 2, 1)), PlainModifier("recurring", 1, "percent", D("25"))] flows = build_cashflows([_rec(due_day=1)], [], [], [], mods, date(2030, 1, 1), date(2030, 2, 28)) assert flows == [(date(2030, 1, 1), D("-75.00")), (date(2030, 2, 1), D("-60.00"))] def test_entfaellt_nur_im_zeitraum(): mods = [PlainModifier("recurring", 1, "remove", D("0"), end_date=date(2030, 2, 28), start_date=date(2030, 2, 1))] flows = build_cashflows([_rec(due_day=1)], [], [], [], mods, date(2030, 1, 1), date(2030, 3, 31)) assert [d for d, _ in flows] == [date(2030, 1, 1), date(2030, 3, 1)] def test_szenario_zahlung_ohne_id_nur_von_kategorie_getroffen(): rec = [PlainRecurring(None, "S", D("-100.00"), "monthly", 1, None, None, 7)] mods = [PlainModifier("recurring", 1, "remove", D("0")), PlainModifier("category", 7, "percent", D("10"))] flows = build_cashflows(rec, [], [], [], mods, date(2030, 1, 1), date(2030, 1, 31)) assert flows == [(date(2030, 1, 1), D("-90.00"))] def test_einmalposten_stichtag_ist_faelligkeit(): planned = [PlainPlanned("E", D("-100.00"), date(2030, 1, 31), 7), PlainPlanned("F", D("-100.00"), date(2030, 2, 1), 7)] mods = [PlainModifier("category", 7, "percent", D("50"), start_date=date(2030, 2, 1))] flows = build_cashflows([], planned, [], [], mods, date(2030, 1, 1), date(2030, 2, 28)) assert flows == [(date(2030, 1, 31), D("-100.00")), (date(2030, 2, 1), D("-50.00"))] def test_ende_unveraendert_mit_zeitraum_feldern(): # "ende" wirkt weiter nur über effective_end, nie als Betragsänderung. mods = [PlainModifier("recurring", 1, "ende", D("0"), date(2030, 2, 15))] flows = build_cashflows([_rec(due_day=1)], [], [], [], mods, date(2030, 1, 1), date(2030, 4, 30)) assert flows == [(date(2030, 1, 1), D("-100.00")), (date(2030, 2, 1), D("-100.00"))]