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

refactor: Modifikator-Helfer öffentlich, load_scenario_inputs aus run_projection (A13-T1)

Browse Source 
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
rdms
2026-09-29 16:05:43 +02:00
co-authored by Claude Opus 5.5
parent acabc3d2a4
commit 0f5bb45a55
3 changed files with 117 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
finance/app/engine/scenario.py
+6 -6
View File
@@ -37,8 +37,8 @@ class PlainModifier:
end_date: date | None = None
def _modified(amount: Decimal, category_id: int | None, recurring_id: int | None,
modifiers: list[PlainModifier]) -> Decimal | None:
def modified_amount(amount: Decimal, category_id: int | None, recurring_id: int | None,
modifiers: list[PlainModifier]) -> Decimal | None:
for m in modifiers:
hit = ((m.target_type == "category" and category_id == m.target_id)
or (m.target_type == "recurring" and recurring_id == m.target_id))
@@ -56,7 +56,7 @@ def _modified(amount: Decimal, category_id: int | None, recurring_id: int | None
return amount
def _effective_end(r: PlainRecurring, modifiers: list[PlainModifier]) -> date | None:
def effective_end(r: PlainRecurring, modifiers: list[PlainModifier]) -> date | None:
"""Fruehestes Ende aus eigenem end_date und allen treffenden ende-Modifikatoren."""
end = r.end_date
for m in modifiers:
@@ -76,14 +76,14 @@ def build_cashflows(recurring: list[PlainRecurring], planned: list[PlainPlanned]
window_start: date, window_end: date) -> list[tuple[date, Decimal]]:
flows: list[tuple[date, Decimal]] = []
for r in recurring:
amount = _modified(r.amount, r.category_id, r.id, modifiers)
amount = modified_amount(r.amount, r.category_id, r.id, modifiers)
if amount is None:
continue
for d in occurrences(r.rhythm, r.due_day, window_start, window_end,
r.start_date, _effective_end(r, modifiers)):
r.start_date, effective_end(r, modifiers)):
flows.append((d, amount))
for p in planned:
amount = _modified(p.amount, p.category_id, None, modifiers)
amount = modified_amount(p.amount, p.category_id, None, modifiers)
if amount is not None and window_start <= p.due <= window_end:
flows.append((p.due, amount))
for d, payout in loan_payouts:
finance/app/services/projection_service.py
+43 -17
View File
@@ -1,3 +1,4 @@
from dataclasses import dataclass
from datetime import date, timedelta
from datetime import datetime, timezone
from decimal import Decimal
@@ -18,36 +19,61 @@ from app.models.tables import (Loan, PlannedItem, ProjectionPoint,
from app.services.balances import total_balance
def run_projection(session: Session, scenario: Scenario, horizon_days: int,
start_date: date) -> ProjectionResult:
end = start_date + timedelta(days=horizon_days)
@dataclass(frozen=True)
class ScenarioInputs:
"""Eingaben einer Szenario-Rechnung (Projektion wie Aufteilung)."""
recurring: list[PlainRecurring]
planned: list[PlainPlanned]
loans: list[Loan]
modifiers: list[PlainModifier]
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."""
recurring = []
if scenario.include_recurring:
if scenario is None or scenario.include_recurring:
recurring = [PlainRecurring(r.id, 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(RecurringItem)).scalars()]
for r in session.execute(
select(RecurringItem).order_by(RecurringItem.id)).scalars()]
planned = []
if scenario.include_planned:
if scenario is None or scenario.include_planned:
planned = [PlainPlanned(p.name, Decimal(p.amount), p.due, p.category_id)
for p in session.execute(select(PlannedItem)).scalars()]
for p in session.execute(
select(PlannedItem).order_by(PlannedItem.id)).scalars()]
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)
for p in session.execute(select(ScenarioPlannedItem).where(
ScenarioPlannedItem.scenario_id == scenario.id)).scalars()]
loans = session.execute(
ScenarioPlannedItem.scenario_id == scenario.id)
.order_by(ScenarioPlannedItem.id)).scalars()]
loans = list(session.execute(
select(Loan).join(ScenarioLoan, ScenarioLoan.loan_id == Loan.id)
.where(ScenarioLoan.scenario_id == scenario.id)).scalars().all()
schedules = [loan_schedule(Decimal(l.principal), Decimal(l.annual_rate_pct),
l.term_months, l.payout_date, l.repayment_type)
for l in loans]
payouts = [(l.payout_date, Decimal(l.principal)) for l in loans]
.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)
for m in session.execute(select(ScenarioModifier).where(
ScenarioModifier.scenario_id == scenario.id)).scalars()]
flows = build_cashflows(recurring, planned, schedules, payouts, modifiers,
start_date, end)
ScenarioModifier.scenario_id == scenario.id)
.order_by(ScenarioModifier.id)).scalars()]
return ScenarioInputs(recurring, planned, loans, modifiers)
def run_projection(session: Session, scenario: Scenario, horizon_days: int,
start_date: date) -> ProjectionResult:
end = start_date + timedelta(days=horizon_days)
inputs = load_scenario_inputs(session, scenario)
schedules = [loan_schedule(Decimal(l.principal), Decimal(l.annual_rate_pct),
l.term_months, l.payout_date, l.repayment_type)
for l in inputs.loans]
payouts = [(l.payout_date, Decimal(l.principal)) for l in inputs.loans]
flows = build_cashflows(inputs.recurring, inputs.planned, schedules, payouts,
inputs.modifiers, start_date, end)
proj = project(total_balance(session), start_date, flows, horizon_days,
threshold=get_settings().warn_threshold)
session.execute(delete(ProjectionPoint).where(
finance/tests/test_scenario_inputs.py
+68
View File
@@ -0,0 +1,68 @@
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.services.projection_service import load_scenario_inputs
def test_modifikator_helfer_oeffentlich():
mods = [PlainModifier("recurring", 1, "percent", Decimal("10")),
PlainModifier("recurring", 1, "ende", Decimal("0"), date(2030, 6, 30))]
assert modified_amount(Decimal("-100.00"), None, 1, mods) == Decimal("-90.00")
assert modified_amount(Decimal("-100.00"), None, 2, mods) == Decimal("-100.00")
r = PlainRecurring(1, "R", Decimal("-100.00"), "monthly", 1, None,
date(2030, 12, 31), None)
assert effective_end(r, mods) == date(2030, 6, 30)
def _daten(db):
rec = RecurringItem(name="R", amount=Decimal("-10.00"), rhythm="monthly", due_day=1)
pln = PlannedItem(name="P", amount=Decimal("-20.00"), due=date(2030, 1, 1))
k1 = Loan(name="K1", principal=Decimal("100.00"), annual_rate_pct=Decimal("0"),
term_months=2, payout_date=date(2030, 1, 1), repayment_type="annuity")
k2 = Loan(name="K2", principal=Decimal("200.00"), annual_rate_pct=Decimal("0"),
term_months=2, payout_date=date(2030, 1, 1), repayment_type="annuity")
sc = Scenario(name="S", description="", include_recurring=True,
include_planned=False)
db.add_all([rec, pln, k1, k2, sc])
db.flush()
db.add_all([
ScenarioLoan(scenario_id=sc.id, loan_id=k1.id),
ScenarioModifier(scenario_id=sc.id, target_type="recurring",
target_id=rec.id, kind="percent", value=Decimal("50")),
ScenarioPlannedItem(scenario_id=sc.id, name="SP", amount=Decimal("-5.00"),
due=date(2030, 2, 1)),
])
db.commit()
return sc
def test_load_scenario_inputs_szenario(db):
sc = _daten(db)
inp = load_scenario_inputs(db, sc)
assert [r.name for r in inp.recurring] == ["R"]
# include_planned=False: globale Einmalposten fehlen, Szenario-eigene zählen immer.
assert [p.name for p in inp.planned] == ["SP"]
assert inp.planned[0].category_id is None
assert [l.name for l in inp.loans] == ["K1"]
assert [(m.kind, m.value) for m in inp.modifiers] == [("percent", Decimal("50"))]
def test_load_scenario_inputs_ohne_fixposten(db):
sc = _daten(db)
sc.include_recurring = False
db.commit()
assert load_scenario_inputs(db, sc).recurring == []
def test_load_scenario_inputs_system_plan(db):
_daten(db)
inp = load_scenario_inputs(db, None)
assert [r.name for r in inp.recurring] == ["R"]
assert [p.name for p in inp.planned] == ["P"]
assert [l.name for l in inp.loans] == ["K1", "K2"]
assert inp.modifiers == []