feat: Planungs- und Szenario-API mit Projektion und Vorschlaegen
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
243
finance/app/routers/planning.py
Normal file
243
finance/app/routers/planning.py
Normal file
@@ -0,0 +1,243 @@
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.auth import require_auth
|
||||
from app.db import get_session
|
||||
from app.engine.loans import loan_schedule
|
||||
from app.models.tables import Category, Loan, PlannedItem, RecurringItem
|
||||
from app.services.suggestions import suggest_recurring
|
||||
|
||||
router = APIRouter(prefix="/api", tags=["planning"],
|
||||
dependencies=[Depends(require_auth)])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- Recurring
|
||||
class RecurringIn(BaseModel):
|
||||
name: str
|
||||
amount: Decimal
|
||||
rhythm: str
|
||||
due_day: int
|
||||
start_date: date | None = None
|
||||
end_date: date | None = None
|
||||
category_id: int | None = None
|
||||
|
||||
|
||||
class RecurringOut(RecurringIn):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
id: int
|
||||
|
||||
|
||||
class RecurringPatch(BaseModel):
|
||||
name: str | None = None
|
||||
amount: Decimal | None = None
|
||||
rhythm: str | None = None
|
||||
due_day: int | None = None
|
||||
start_date: date | None = None
|
||||
end_date: date | None = None
|
||||
category_id: int | None = None
|
||||
|
||||
|
||||
class SuggestionOut(BaseModel):
|
||||
name: str
|
||||
amount: Decimal
|
||||
rhythm: str
|
||||
due_day: int
|
||||
category_id: int | None = None
|
||||
|
||||
|
||||
def _check_category(session: Session, category_id: int | None) -> None:
|
||||
if category_id is not None and session.get(Category, category_id) is None:
|
||||
raise HTTPException(404, "Kategorie nicht gefunden")
|
||||
|
||||
|
||||
@router.get("/recurring", response_model=list[RecurringOut])
|
||||
def list_recurring(session: Session = Depends(get_session)):
|
||||
return [RecurringOut.model_validate(r)
|
||||
for r in session.execute(select(RecurringItem)).scalars()]
|
||||
|
||||
|
||||
@router.post("/recurring", response_model=RecurringOut, status_code=201)
|
||||
def create_recurring(data: RecurringIn, session: Session = Depends(get_session)):
|
||||
_check_category(session, data.category_id)
|
||||
item = RecurringItem(**data.model_dump())
|
||||
session.add(item)
|
||||
session.commit()
|
||||
session.refresh(item)
|
||||
return RecurringOut.model_validate(item)
|
||||
|
||||
|
||||
# Muss vor /recurring/{item_id} stehen, sonst faengt die {id}-Route den Pfad ab.
|
||||
@router.get("/recurring/suggestions", response_model=list[SuggestionOut])
|
||||
def recurring_suggestions(session: Session = Depends(get_session)):
|
||||
return suggest_recurring(session)
|
||||
|
||||
|
||||
@router.patch("/recurring/{item_id}", response_model=RecurringOut)
|
||||
def patch_recurring(item_id: int, data: RecurringPatch,
|
||||
session: Session = Depends(get_session)):
|
||||
item = session.get(RecurringItem, item_id)
|
||||
if item is None:
|
||||
raise HTTPException(404, "Fixposten nicht gefunden")
|
||||
fields = data.model_dump(exclude_unset=True)
|
||||
if "category_id" in fields:
|
||||
_check_category(session, fields["category_id"])
|
||||
for key, value in fields.items():
|
||||
setattr(item, key, value)
|
||||
session.commit()
|
||||
session.refresh(item)
|
||||
return RecurringOut.model_validate(item)
|
||||
|
||||
|
||||
@router.delete("/recurring/{item_id}", status_code=204)
|
||||
def delete_recurring(item_id: int, session: Session = Depends(get_session)):
|
||||
item = session.get(RecurringItem, item_id)
|
||||
if item is None:
|
||||
raise HTTPException(404, "Fixposten nicht gefunden")
|
||||
session.delete(item)
|
||||
session.commit()
|
||||
|
||||
|
||||
# ------------------------------------------------------------------ Planned
|
||||
class PlannedIn(BaseModel):
|
||||
name: str
|
||||
amount: Decimal
|
||||
due: date
|
||||
category_id: int | None = None
|
||||
|
||||
|
||||
class PlannedOut(PlannedIn):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
id: int
|
||||
|
||||
|
||||
class PlannedPatch(BaseModel):
|
||||
name: str | None = None
|
||||
amount: Decimal | None = None
|
||||
due: date | None = None
|
||||
category_id: int | None = None
|
||||
|
||||
|
||||
@router.get("/planned", response_model=list[PlannedOut])
|
||||
def list_planned(session: Session = Depends(get_session)):
|
||||
return [PlannedOut.model_validate(p)
|
||||
for p in session.execute(select(PlannedItem)).scalars()]
|
||||
|
||||
|
||||
@router.post("/planned", response_model=PlannedOut, status_code=201)
|
||||
def create_planned(data: PlannedIn, session: Session = Depends(get_session)):
|
||||
_check_category(session, data.category_id)
|
||||
item = PlannedItem(**data.model_dump())
|
||||
session.add(item)
|
||||
session.commit()
|
||||
session.refresh(item)
|
||||
return PlannedOut.model_validate(item)
|
||||
|
||||
|
||||
@router.patch("/planned/{item_id}", response_model=PlannedOut)
|
||||
def patch_planned(item_id: int, data: PlannedPatch,
|
||||
session: Session = Depends(get_session)):
|
||||
item = session.get(PlannedItem, item_id)
|
||||
if item is None:
|
||||
raise HTTPException(404, "Einmalposten nicht gefunden")
|
||||
fields = data.model_dump(exclude_unset=True)
|
||||
if "category_id" in fields:
|
||||
_check_category(session, fields["category_id"])
|
||||
for key, value in fields.items():
|
||||
setattr(item, key, value)
|
||||
session.commit()
|
||||
session.refresh(item)
|
||||
return PlannedOut.model_validate(item)
|
||||
|
||||
|
||||
@router.delete("/planned/{item_id}", status_code=204)
|
||||
def delete_planned(item_id: int, session: Session = Depends(get_session)):
|
||||
item = session.get(PlannedItem, item_id)
|
||||
if item is None:
|
||||
raise HTTPException(404, "Einmalposten nicht gefunden")
|
||||
session.delete(item)
|
||||
session.commit()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------- Loans
|
||||
class LoanIn(BaseModel):
|
||||
name: str
|
||||
principal: Decimal
|
||||
annual_rate_pct: Decimal
|
||||
term_months: int
|
||||
payout_date: date
|
||||
repayment_type: str = "annuity"
|
||||
|
||||
|
||||
class LoanOut(LoanIn):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
id: int
|
||||
|
||||
|
||||
class LoanPatch(BaseModel):
|
||||
name: str | None = None
|
||||
principal: Decimal | None = None
|
||||
annual_rate_pct: Decimal | None = None
|
||||
term_months: int | None = None
|
||||
payout_date: date | None = None
|
||||
repayment_type: str | None = None
|
||||
|
||||
|
||||
class InstallmentOut(BaseModel):
|
||||
due: date
|
||||
payment: Decimal
|
||||
interest: Decimal
|
||||
principal: Decimal
|
||||
remaining: Decimal
|
||||
|
||||
|
||||
@router.get("/loans", response_model=list[LoanOut])
|
||||
def list_loans(session: Session = Depends(get_session)):
|
||||
return [LoanOut.model_validate(l)
|
||||
for l in session.execute(select(Loan)).scalars()]
|
||||
|
||||
|
||||
@router.post("/loans", response_model=LoanOut, status_code=201)
|
||||
def create_loan(data: LoanIn, session: Session = Depends(get_session)):
|
||||
loan = Loan(**data.model_dump())
|
||||
session.add(loan)
|
||||
session.commit()
|
||||
session.refresh(loan)
|
||||
return LoanOut.model_validate(loan)
|
||||
|
||||
|
||||
@router.patch("/loans/{loan_id}", response_model=LoanOut)
|
||||
def patch_loan(loan_id: int, data: LoanPatch, session: Session = Depends(get_session)):
|
||||
loan = session.get(Loan, loan_id)
|
||||
if loan is None:
|
||||
raise HTTPException(404, "Kredit nicht gefunden")
|
||||
for key, value in data.model_dump(exclude_unset=True).items():
|
||||
setattr(loan, key, value)
|
||||
session.commit()
|
||||
session.refresh(loan)
|
||||
return LoanOut.model_validate(loan)
|
||||
|
||||
|
||||
@router.delete("/loans/{loan_id}", status_code=204)
|
||||
def delete_loan(loan_id: int, session: Session = Depends(get_session)):
|
||||
loan = session.get(Loan, loan_id)
|
||||
if loan is None:
|
||||
raise HTTPException(404, "Kredit nicht gefunden")
|
||||
session.delete(loan)
|
||||
session.commit()
|
||||
|
||||
|
||||
@router.get("/loans/{loan_id}/schedule", response_model=list[InstallmentOut])
|
||||
def get_loan_schedule(loan_id: int, session: Session = Depends(get_session)):
|
||||
loan = session.get(Loan, loan_id)
|
||||
if loan is None:
|
||||
raise HTTPException(404, "Kredit nicht gefunden")
|
||||
plan = loan_schedule(Decimal(loan.principal), Decimal(loan.annual_rate_pct),
|
||||
loan.term_months, loan.payout_date, loan.repayment_type)
|
||||
return [InstallmentOut(due=i.due, payment=i.payment, interest=i.interest,
|
||||
principal=i.principal, remaining=i.remaining)
|
||||
for i in plan]
|
||||
169
finance/app/routers/scenarios.py
Normal file
169
finance/app/routers/scenarios.py
Normal file
@@ -0,0 +1,169 @@
|
||||
from datetime import date
|
||||
from decimal import Decimal
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.auth import require_auth
|
||||
from app.config import get_settings
|
||||
from app.db import get_session
|
||||
from app.models.tables import (Loan, ProjectionPoint, Scenario, ScenarioLoan,
|
||||
ScenarioModifier)
|
||||
from app.services.projection_service import run_projection
|
||||
|
||||
router = APIRouter(prefix="/api/scenarios", tags=["scenarios"],
|
||||
dependencies=[Depends(require_auth)])
|
||||
|
||||
|
||||
class ScenarioIn(BaseModel):
|
||||
name: str
|
||||
description: str = ""
|
||||
include_recurring: bool = True
|
||||
include_planned: bool = True
|
||||
|
||||
|
||||
class ScenarioOut(ScenarioIn):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
id: int
|
||||
|
||||
|
||||
class ScenarioPatch(BaseModel):
|
||||
name: str | None = None
|
||||
description: str | None = None
|
||||
include_recurring: bool | None = None
|
||||
include_planned: bool | None = None
|
||||
|
||||
|
||||
class ModifierIn(BaseModel):
|
||||
target_type: str
|
||||
target_id: int
|
||||
kind: str
|
||||
value: Decimal = Decimal("0")
|
||||
|
||||
|
||||
class ModifierOut(ModifierIn):
|
||||
model_config = ConfigDict(from_attributes=True)
|
||||
id: int
|
||||
scenario_id: int
|
||||
|
||||
|
||||
class SeriesPointOut(BaseModel):
|
||||
day: date
|
||||
balance: Decimal
|
||||
|
||||
|
||||
class ProjectionOut(BaseModel):
|
||||
low_point_date: date
|
||||
low_point_balance: Decimal
|
||||
below_zero_date: date | None
|
||||
below_threshold_date: date | None
|
||||
series: list[SeriesPointOut]
|
||||
|
||||
|
||||
def _get_scenario(session: Session, scenario_id: int) -> Scenario:
|
||||
scenario = session.get(Scenario, scenario_id)
|
||||
if scenario is None:
|
||||
raise HTTPException(404, "Szenario nicht gefunden")
|
||||
return scenario
|
||||
|
||||
|
||||
@router.get("", response_model=list[ScenarioOut])
|
||||
def list_scenarios(session: Session = Depends(get_session)):
|
||||
return [ScenarioOut.model_validate(s)
|
||||
for s in session.execute(select(Scenario)).scalars()]
|
||||
|
||||
|
||||
@router.post("", response_model=ScenarioOut, status_code=201)
|
||||
def create_scenario(data: ScenarioIn, session: Session = Depends(get_session)):
|
||||
if session.execute(select(Scenario).where(Scenario.name == data.name)).scalar():
|
||||
raise HTTPException(409, "Szenario existiert bereits")
|
||||
scenario = Scenario(**data.model_dump())
|
||||
session.add(scenario)
|
||||
session.commit()
|
||||
session.refresh(scenario)
|
||||
return ScenarioOut.model_validate(scenario)
|
||||
|
||||
|
||||
@router.patch("/{scenario_id}", response_model=ScenarioOut)
|
||||
def patch_scenario(scenario_id: int, data: ScenarioPatch,
|
||||
session: Session = Depends(get_session)):
|
||||
scenario = _get_scenario(session, scenario_id)
|
||||
for key, value in data.model_dump(exclude_unset=True).items():
|
||||
setattr(scenario, key, value)
|
||||
session.commit()
|
||||
session.refresh(scenario)
|
||||
return ScenarioOut.model_validate(scenario)
|
||||
|
||||
|
||||
@router.delete("/{scenario_id}", status_code=204)
|
||||
def delete_scenario(scenario_id: int, session: Session = Depends(get_session)):
|
||||
scenario = _get_scenario(session, scenario_id)
|
||||
session.delete(scenario)
|
||||
session.commit()
|
||||
|
||||
|
||||
@router.post("/{scenario_id}/loans/{loan_id}", status_code=204)
|
||||
def add_scenario_loan(scenario_id: int, loan_id: int,
|
||||
session: Session = Depends(get_session)):
|
||||
_get_scenario(session, scenario_id)
|
||||
if session.get(Loan, loan_id) is None:
|
||||
raise HTTPException(404, "Kredit nicht gefunden")
|
||||
if session.get(ScenarioLoan, (scenario_id, loan_id)) is None:
|
||||
session.add(ScenarioLoan(scenario_id=scenario_id, loan_id=loan_id))
|
||||
session.commit()
|
||||
|
||||
|
||||
@router.delete("/{scenario_id}/loans/{loan_id}", status_code=204)
|
||||
def remove_scenario_loan(scenario_id: int, loan_id: int,
|
||||
session: Session = Depends(get_session)):
|
||||
_get_scenario(session, scenario_id)
|
||||
link = session.get(ScenarioLoan, (scenario_id, loan_id))
|
||||
if link is not None:
|
||||
session.delete(link)
|
||||
session.commit()
|
||||
|
||||
|
||||
@router.post("/{scenario_id}/modifiers", response_model=ModifierOut, status_code=201)
|
||||
def add_modifier(scenario_id: int, data: ModifierIn,
|
||||
session: Session = Depends(get_session)):
|
||||
_get_scenario(session, scenario_id)
|
||||
modifier = ScenarioModifier(scenario_id=scenario_id, **data.model_dump())
|
||||
session.add(modifier)
|
||||
session.commit()
|
||||
session.refresh(modifier)
|
||||
return ModifierOut.model_validate(modifier)
|
||||
|
||||
|
||||
@router.delete("/{scenario_id}/modifiers/{mod_id}", status_code=204)
|
||||
def delete_modifier(scenario_id: int, mod_id: int,
|
||||
session: Session = Depends(get_session)):
|
||||
_get_scenario(session, scenario_id)
|
||||
modifier = session.get(ScenarioModifier, mod_id)
|
||||
if modifier is None or modifier.scenario_id != scenario_id:
|
||||
raise HTTPException(404, "Modifikator nicht gefunden")
|
||||
session.delete(modifier)
|
||||
session.commit()
|
||||
|
||||
|
||||
@router.post("/{scenario_id}/project", response_model=ProjectionOut)
|
||||
def project_scenario(scenario_id: int, horizon_days: int | None = None,
|
||||
start_date: date | None = None,
|
||||
session: Session = Depends(get_session)):
|
||||
scenario = _get_scenario(session, scenario_id)
|
||||
settings = get_settings()
|
||||
resolved_horizon = horizon_days if horizon_days is not None else settings.horizon_days
|
||||
resolved_start = start_date if start_date is not None else date.today()
|
||||
result = run_projection(session, scenario, resolved_horizon, resolved_start)
|
||||
points = session.execute(
|
||||
select(ProjectionPoint).where(ProjectionPoint.scenario_id == scenario_id)
|
||||
.order_by(ProjectionPoint.day)
|
||||
).scalars().all()
|
||||
return ProjectionOut(
|
||||
low_point_date=result.low_point_date,
|
||||
low_point_balance=result.low_point_balance,
|
||||
below_zero_date=result.below_zero_date,
|
||||
below_threshold_date=result.below_threshold_date,
|
||||
series=[SeriesPointOut(day=p.day, balance=p.balance) for p in points],
|
||||
)
|
||||
Reference in New Issue
Block a user