wlfb/bin
SHA256
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: API für Szenario-Zeitraum, Neuer Betrag und szenario-eigene wiederkehrende Zahlungen (A14-T3)

Browse Source 
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
rdms
2026-09-30 13:03:16 +02:00
co-authored by Claude Opus 5.5
parent aaf05b9209
commit 2aed685d7e
4 changed files with 259 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
finance/app/routers/scenarios.py
+78 -6
View File
@@ -12,7 +12,8 @@ from app.config import get_settings
from app.db import get_session
from app.models.tables import (Loan, ProjectionPoint, ProjectionResult,
Scenario, ScenarioLoan, ScenarioModifier,
ScenarioPlannedItem)
ScenarioPlannedItem, ScenarioRecurringItem)
from app.routers.planning import RecurringIn, _check_category
from app.reserved import check_reserved_name
from app.services.projection_service import run_projection
@@ -47,16 +48,23 @@ class ScenarioPatch(BaseModel):
class ModifierIn(BaseModel):
target_type: Literal["category", "recurring"]
target_id: int
kind: Literal["percent", "absolute", "remove", "ende"]
kind: Literal["percent", "absolute", "set", "remove", "ende"]
value: Decimal = Decimal("0")
start_date: date | None = None
end_date: date | None = None
@model_validator(mode="after")
def _ende_braucht_datum(self):
if self.kind == "ende" and self.end_date is None:
def _art_und_zeitraum(self):
# end_date = Enddatum bei "ende", sonst „Gültig bis" (A14).
if self.kind == "ende":
if self.end_date is None:
raise ValueError("Art »Ende« erfordert ein Datum")
if self.kind != "ende":
self.end_date = None
self.start_date = None
if self.kind == "set" and self.target_type != "recurring":
raise ValueError("Art »Neuer Betrag« ist nur für Posten möglich")
if (self.start_date is not None and self.end_date is not None
and self.start_date > self.end_date):
raise ValueError("»Gültig ab« liegt nach »Gültig bis«")
return self
@@ -127,6 +135,8 @@ def delete_scenario(scenario_id: int, session: Session = Depends(get_session)):
scenario = _get_scenario(session, scenario_id)
# No DB-level cascades exist, so remove dependent rows explicitly to avoid
# foreign-key violations (would surface as a 500 on Postgres).
session.execute(delete(ScenarioRecurringItem).where(
ScenarioRecurringItem.scenario_id == scenario_id))
session.execute(delete(ProjectionPoint).where(
ProjectionPoint.scenario_id == scenario_id))
session.execute(delete(ProjectionResult).where(
@@ -203,6 +213,7 @@ class ScenarioPlannedIn(BaseModel):
name: str
amount: Decimal
due: date
category_id: int | None = None
class ScenarioPlannedOut(ScenarioPlannedIn):
@@ -225,6 +236,7 @@ def list_scenario_planned(scenario_id: int, session: Session = Depends(get_sessi
def add_scenario_planned(scenario_id: int, data: ScenarioPlannedIn,
session: Session = Depends(get_session)):
_get_scenario(session, scenario_id)
_check_category(session, data.category_id)
item = ScenarioPlannedItem(scenario_id=scenario_id, **data.model_dump())
session.add(item)
session.commit()
@@ -248,6 +260,7 @@ def patch_scenario_planned(scenario_id: int, item_id: int, data: ScenarioPlanned
session: Session = Depends(get_session)):
# Full-Body-Update, Muster patch_modifier.
_get_scenario(session, scenario_id)
_check_category(session, data.category_id)
item = session.get(ScenarioPlannedItem, item_id)
if item is None or item.scenario_id != scenario_id:
raise HTTPException(404, "Einmalzahlung nicht gefunden")
@@ -278,3 +291,62 @@ def project_scenario(scenario_id: int, horizon_days: int | None = None,
below_threshold_date=result.below_threshold_date,
series=[SeriesPointOut(day=p.day, balance=p.balance) for p in points],
)
class ScenarioRecurringOut(RecurringIn):
model_config = ConfigDict(from_attributes=True)
id: int
scenario_id: int
def _get_scenario_recurring(session: Session, scenario_id: int,
item_id: int) -> ScenarioRecurringItem:
item = session.get(ScenarioRecurringItem, item_id)
if item is None or item.scenario_id != scenario_id:
raise HTTPException(404, "Wiederkehrende Zahlung nicht gefunden")
return item
@router.get("/{scenario_id}/recurring", response_model=list[ScenarioRecurringOut])
def list_scenario_recurring(scenario_id: int, session: Session = Depends(get_session)):
_get_scenario(session, scenario_id)
return [ScenarioRecurringOut.model_validate(r) for r in session.execute(
select(ScenarioRecurringItem)
.where(ScenarioRecurringItem.scenario_id == scenario_id)
.order_by(ScenarioRecurringItem.name, ScenarioRecurringItem.id)).scalars()]
@router.post("/{scenario_id}/recurring", response_model=ScenarioRecurringOut,
status_code=201)
def add_scenario_recurring(scenario_id: int, data: RecurringIn,
session: Session = Depends(get_session)):
# Bewusst ohne Doppelzählungs-Prüfung (gilt nur für globale Fixposten).
_get_scenario(session, scenario_id)
_check_category(session, data.category_id)
item = ScenarioRecurringItem(scenario_id=scenario_id, **data.model_dump())
session.add(item)
session.commit()
session.refresh(item)
return ScenarioRecurringOut.model_validate(item)
@router.patch("/{scenario_id}/recurring/{item_id}", response_model=ScenarioRecurringOut)
def patch_scenario_recurring(scenario_id: int, item_id: int, data: RecurringIn,
session: Session = Depends(get_session)):
# Full-Body-Update, Muster patch_scenario_planned.
_get_scenario(session, scenario_id)
item = _get_scenario_recurring(session, scenario_id, item_id)
_check_category(session, data.category_id)
for key, value in data.model_dump().items():
setattr(item, key, value)
session.commit()
session.refresh(item)
return ScenarioRecurringOut.model_validate(item)
@router.delete("/{scenario_id}/recurring/{item_id}", status_code=204)
def delete_scenario_recurring(scenario_id: int, item_id: int,
session: Session = Depends(get_session)):
_get_scenario(session, scenario_id)
session.delete(_get_scenario_recurring(session, scenario_id, item_id))
session.commit()
finance/app/services/projection_service.py
+14 -6
View File
@@ -15,7 +15,7 @@ from app.engine.scenario import (PlainModifier, PlainPlanned, PlainRecurring,
from app.models.tables import (Loan, PlannedItem, ProjectionPoint,
ProjectionResult, RecurringItem, Scenario,
ScenarioLoan, ScenarioModifier,
ScenarioPlannedItem)
ScenarioPlannedItem, ScenarioRecurringItem)
from app.services.balances import total_balance
@@ -31,7 +31,8 @@ class ScenarioInputs:
def load_scenario_inputs(session: Session, scenario: Scenario | None) -> ScenarioInputs:
"""Lädt die Eingaben eines Szenarios. scenario=None ist die reine
Planung („System-Plan"): alle Fixposten, alle globalen Einmalposten,
alle Kredite, keine Modifikatoren, keine Szenario-Einmalposten."""
alle Kredite, keine Modifikatoren, keine Szenario-Einmalposten und keine
szenario-eigenen wiederkehrenden Zahlungen."""
recurring = []
if scenario is None or scenario.include_recurring:
recurring = [PlainRecurring(r.id, r.name, Decimal(r.amount), r.rhythm,
@@ -46,9 +47,16 @@ def load_scenario_inputs(session: Session, scenario: Scenario | None) -> Scenari
if scenario is None:
loans = list(session.execute(select(Loan).order_by(Loan.id)).scalars())
return ScenarioInputs(recurring, planned, loans, [])
# Szenario-eigene Einmalzahlungen zaehlen IMMER fuer dieses Szenario -
# include_planned steuert nur die globalen PlannedItems.
planned += [PlainPlanned(p.name, Decimal(p.amount), p.due, None)
# Szenario-eigene Zahlungen zählen IMMER für dieses Szenario -
# include_recurring/include_planned steuern nur die globalen Posten.
# id=None: Posten-Einträge (target_type="recurring") treffen sie nie,
# die IDs zweier Tabellen könnten sonst kollidieren.
recurring += [PlainRecurring(None, r.name, Decimal(r.amount), r.rhythm, r.due_day,
r.start_date, r.end_date, r.category_id)
for r in session.execute(select(ScenarioRecurringItem).where(
ScenarioRecurringItem.scenario_id == scenario.id)
.order_by(ScenarioRecurringItem.id)).scalars()]
planned += [PlainPlanned(p.name, Decimal(p.amount), p.due, p.category_id)
for p in session.execute(select(ScenarioPlannedItem).where(
ScenarioPlannedItem.scenario_id == scenario.id)
.order_by(ScenarioPlannedItem.id)).scalars()]
@@ -57,7 +65,7 @@ def load_scenario_inputs(session: Session, scenario: Scenario | None) -> Scenari
.where(ScenarioLoan.scenario_id == scenario.id)
.order_by(Loan.id)).scalars())
modifiers = [PlainModifier(m.target_type, m.target_id, m.kind,
Decimal(m.value), m.end_date)
Decimal(m.value), m.end_date, m.start_date)
for m in session.execute(select(ScenarioModifier).where(
ScenarioModifier.scenario_id == scenario.id)
.order_by(ScenarioModifier.id)).scalars()]
finance/tests/test_crud_api.py
+118 -4
View File
@@ -225,11 +225,11 @@ def test_modifier_ende_validierung(client, db):
"end_date": "2026-08-31"})
assert r.status_code == 201
assert r.json()["end_date"] == "2026-08-31"
# andere Arten nullen end_date
# andere Arten behalten end_date als „Gültig bis"
r = client.post(f"/api/scenarios/{sc['id']}/modifiers", headers=H, json={
"target_type": "recurring", "target_id": 1, "kind": "remove",
"end_date": "2026-08-31"})
assert r.status_code == 201 and r.json()["end_date"] is None
assert r.status_code == 201 and r.json()["end_date"] == "2026-08-31"
def test_scenario_planned_crud_und_cleanup(client, db):
@@ -269,11 +269,11 @@ def test_modifier_patch(client, db):
r = client.patch(f"/api/scenarios/{sc['id']}/modifiers/{m['id']}", headers=H, json={
"target_type": "recurring", "target_id": 1, "kind": "ende"})
assert r.status_code == 422
# zurueck auf remove: end_date wird genullt
# zurueck auf remove: end_date bleibt als „Gültig bis" erhalten
r = client.patch(f"/api/scenarios/{sc['id']}/modifiers/{m['id']}", headers=H, json={
"target_type": "recurring", "target_id": 1, "kind": "remove",
"end_date": "2026-08-31"})
assert r.status_code == 200 and r.json()["end_date"] is None
assert r.status_code == 200 and r.json()["end_date"] == "2026-08-31"
# fremdes Szenario -> 404
other = client.post("/api/scenarios", headers=H, json={"name": "ModPatch2"}).json()
r = client.patch(f"/api/scenarios/{other['id']}/modifiers/{m['id']}", headers=H, json={
@@ -318,3 +318,117 @@ def test_szenario_name_system_ist_ist_reserviert_patch(client):
def test_szenario_normale_namen_weiter_erlaubt(client):
assert client.post("/api/scenarios", json={"name": "Ist-Zustand plus"},
headers=H).status_code == 201
def _msgs(r):
return " ".join(e["msg"] for e in r.json()["detail"])
def test_modifier_zeitraum_und_neuer_betrag(client):
sc = client.post("/api/scenarios", headers=H, json={"name": "Zeitraum"}).json()
url = f"/api/scenarios/{sc['id']}/modifiers"
r = client.post(url, headers=H, json={
"target_type": "recurring", "target_id": 1, "kind": "percent", "value": "50",
"start_date": "2026-10-16", "end_date": "2027-03-31"})
assert r.status_code == 201
assert (r.json()["start_date"], r.json()["end_date"]) == ("2026-10-16", "2027-03-31")
r = client.post(url, headers=H, json={
"target_type": "recurring", "target_id": 1, "kind": "percent", "value": "50",
"start_date": "2027-04-01", "end_date": "2027-03-31"})
assert r.status_code == 422 and "»Gültig ab« liegt nach »Gültig bis«" in _msgs(r)
r = client.post(url, headers=H, json={
"target_type": "category", "target_id": 1, "kind": "set", "value": "-80.00"})
assert r.status_code == 422 and "Art »Neuer Betrag« ist nur für Posten möglich" in _msgs(r)
r = client.post(url, headers=H, json={
"target_type": "recurring", "target_id": 1, "kind": "set", "value": "-137.50",
"start_date": "2026-10-16"})
assert r.status_code == 201
assert (r.json()["kind"], r.json()["value"], r.json()["end_date"]) == ("set", "-137.50", None)
r = client.post(url, headers=H, json={
"target_type": "recurring", "target_id": 1, "kind": "ende",
"start_date": "2026-10-01", "end_date": "2026-10-15"})
assert r.status_code == 201
assert (r.json()["start_date"], r.json()["end_date"]) == (None, "2026-10-15")
def test_modifier_patch_artwechsel_zeitraum(client):
sc = client.post("/api/scenarios", headers=H, json={"name": "Artwechsel"}).json()
m = client.post(f"/api/scenarios/{sc['id']}/modifiers", headers=H, json={
"target_type": "recurring", "target_id": 1, "kind": "percent", "value": "50",
"start_date": "2026-10-16", "end_date": "2027-03-31"}).json()
url = f"/api/scenarios/{sc['id']}/modifiers/{m['id']}"
r = client.patch(url, headers=H, json={
"target_type": "recurring", "target_id": 1, "kind": "ende",
"start_date": "2026-10-16", "end_date": "2026-12-31"})
assert r.status_code == 200
assert (r.json()["start_date"], r.json()["end_date"]) == (None, "2026-12-31")
r = client.patch(url, headers=H, json={
"target_type": "recurring", "target_id": 1, "kind": "percent", "value": "10",
"end_date": "2026-12-31"})
assert r.status_code == 200
assert (r.json()["kind"], r.json()["start_date"], r.json()["end_date"]) == \
("percent", None, "2026-12-31")
def test_scenario_recurring_crud_und_cleanup(client, db):
from app.models.tables import Category, ScenarioRecurringItem
cat = Category(name="Wohnen-SR")
db.add(cat)
db.commit()
sc = client.post("/api/scenarios", headers=H, json={"name": "SR-CRUD"}).json()
url = f"/api/scenarios/{sc['id']}/recurring"
body = {"name": "Miete neu", "amount": "-137.50", "rhythm": "monthly",
"due_day": 14, "start_date": "2026-10-16", "category_id": cat.id}
r = client.post(url, headers=H, json=body)
assert r.status_code == 201
item = r.json()
assert (item["scenario_id"], item["amount"], item["end_date"], item["category_id"]) == \
(sc["id"], "-137.50", None, cat.id)
assert [i["name"] for i in client.get(url, headers=H).json()] == ["Miete neu"]
r = client.patch(f"{url}/{item['id']}", headers=H, json={
"name": "Miete neu", "amount": "-250.00", "rhythm": "quarterly", "due_day": 1,
"start_date": None, "end_date": "2027-12-31", "category_id": None})
assert r.status_code == 200
assert (r.json()["amount"], r.json()["rhythm"], r.json()["category_id"]) == \
("-250.00", "quarterly", None)
# Validierung wie globale Fixposten
assert client.post(url, headers=H, json={**body, "due_day": 0}).status_code == 422
assert client.post(url, headers=H, json={**body, "rhythm": "weekly"}).status_code == 422
assert client.post(url, headers=H, json={**body, "end_date": "2026-10-01"}).status_code == 422
r = client.post(url, headers=H, json={**body, "category_id": 9999})
assert r.status_code == 404 and r.json()["detail"] == "Kategorie nicht gefunden"
# fremdes/fehlendes Szenario
other = client.post("/api/scenarios", headers=H, json={"name": "SR-Other"}).json()
r = client.delete(f"/api/scenarios/{other['id']}/recurring/{item['id']}", headers=H)
assert r.status_code == 404 and r.json()["detail"] == "Wiederkehrende Zahlung nicht gefunden"
assert client.get("/api/scenarios/99999/recurring", headers=H).status_code == 404
# Szenario löschen räumt mit ab (kein FK-Fehler)
assert client.delete(f"/api/scenarios/{sc['id']}", headers=H).status_code == 204
assert db.query(ScenarioRecurringItem).count() == 0
def test_scenario_recurring_delete(client):
sc = client.post("/api/scenarios", headers=H, json={"name": "SR-Del"}).json()
url = f"/api/scenarios/{sc['id']}/recurring"
item = client.post(url, headers=H, json={
"name": "X", "amount": "-1.00", "rhythm": "monthly", "due_day": 1}).json()
assert client.delete(f"{url}/{item['id']}", headers=H).status_code == 204
assert client.get(url, headers=H).json() == []
def test_scenario_planned_mit_kategorie(client, db):
from app.models.tables import Category
cat = Category(name="Kat-SPI")
db.add(cat)
db.commit()
sc = client.post("/api/scenarios", headers=H, json={"name": "SPI-Kat"}).json()
url = f"/api/scenarios/{sc['id']}/planned"
r = client.post(url, headers=H, json={
"name": "E", "amount": "-5.00", "due": "2026-11-01", "category_id": cat.id})
assert r.status_code == 201 and r.json()["category_id"] == cat.id
assert client.post(url, headers=H, json={
"name": "E", "amount": "-5.00", "due": "2026-11-01",
"category_id": 9999}).status_code == 404
r = client.patch(f"{url}/{r.json()['id']}", headers=H, json={
"name": "E", "amount": "-5.00", "due": "2026-11-01", "category_id": None})
assert r.status_code == 200 and r.json()["category_id"] is None
finance/tests/test_scenario_inputs.py
+48 -5
View File
@@ -1,11 +1,11 @@
from datetime import date
from decimal import Decimal
from app.engine.scenario import (PlainModifier, PlainRecurring, effective_end,
modified_amount)
from app.models.tables import (Loan, PlannedItem, RecurringItem, Scenario,
ScenarioLoan, ScenarioModifier,
ScenarioPlannedItem)
from app.engine.scenario import (PlainModifier, PlainRecurring, build_cashflows,
effective_end, modified_amount)
from app.models.tables import (Category, Loan, PlannedItem, RecurringItem,
Scenario, ScenarioLoan, ScenarioModifier,
ScenarioPlannedItem, ScenarioRecurringItem)
from app.services.projection_service import load_scenario_inputs
@@ -66,3 +66,46 @@ def test_load_scenario_inputs_system_plan(db):
assert [p.name for p in inp.planned] == ["P"]
assert [l.name for l in inp.loans] == ["K1", "K2"]
assert inp.modifiers == []
def test_szenario_zahlung_zaehlt_immer_und_id_kollision(db):
rec = RecurringItem(name="R", amount=Decimal("-10.00"), rhythm="monthly", due_day=1)
sc = Scenario(name="S-A14", description="", include_recurring=False,
include_planned=False)
db.add_all([rec, sc])
db.flush()
sri = ScenarioRecurringItem(scenario_id=sc.id, name="SR", amount=Decimal("-100.00"),
rhythm="monthly", due_day=1)
db.add(sri)
db.flush()
# Posten-Eintrag, dessen target_id zufällig der ID der Szenario-Zahlung entspricht:
db.add(ScenarioModifier(scenario_id=sc.id, target_type="recurring",
target_id=sri.id, kind="remove", value=Decimal("0")))
db.commit()
inp = load_scenario_inputs(db, sc)
# include_recurring=False: globaler Posten fehlt, Szenario-Zahlung zählt, id=None.
assert [(r.id, r.name) for r in inp.recurring] == [(None, "SR")]
flows = build_cashflows(inp.recurring, inp.planned, [], [], inp.modifiers,
date(2030, 1, 1), date(2030, 1, 31))
assert flows == [(date(2030, 1, 1), Decimal("-100.00"))]
# System-Plan kennt keine Szenario-Zahlungen.
assert [r.name for r in load_scenario_inputs(db, None).recurring] == ["R"]
def test_zeitraum_und_kategorie_werden_geladen(db):
cat = Category(name="K-A14")
sc = Scenario(name="S-A14b", description="")
db.add_all([cat, sc])
db.flush()
db.add_all([
ScenarioModifier(scenario_id=sc.id, target_type="category", target_id=cat.id,
kind="percent", value=Decimal("50"),
start_date=date(2030, 2, 1), end_date=date(2030, 3, 31)),
ScenarioPlannedItem(scenario_id=sc.id, name="E", amount=Decimal("-5.00"),
due=date(2030, 2, 1), category_id=cat.id),
])
db.commit()
inp = load_scenario_inputs(db, sc)
m = inp.modifiers[0]
assert (m.start_date, m.end_date) == (date(2030, 2, 1), date(2030, 3, 31))
assert [(p.name, p.category_id) for p in inp.planned] == [("E", cat.id)]