229 lines
9.7 KiB
Python
229 lines
9.7 KiB
Python
from __future__ import annotations
|
|
|
|
from decimal import Decimal, ROUND_HALF_UP
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.extensions import db
|
|
from app.models import Account, AllocationSuggestion, MonthlyAllocation, to_decimal
|
|
from app.utils.users import personal_account_names
|
|
|
|
|
|
class AllocationSuggestionService:
|
|
TARGET_SLUGS = ["sparen", "urlaub", "freizeit", "persoenlich-flo", "persoenlich-desi"]
|
|
PRIORITY_TARGETS = ["sparen", "urlaub", "freizeit"]
|
|
PERSONAL_TARGETS = ["persoenlich-flo", "persoenlich-desi"]
|
|
|
|
def __init__(self, target_rules: dict[str, dict], default_personal_split_desi_pct: float):
|
|
self.target_rules = target_rules
|
|
self.default_personal_split_desi_pct = default_personal_split_desi_pct
|
|
|
|
FIELD_MAP = {
|
|
"sparen": ("savings_min_pct", "savings_max_pct", "Sparen"),
|
|
"urlaub": ("vacation_min_pct", "vacation_max_pct", "Urlaub"),
|
|
"freizeit": ("leisure_min_pct", "leisure_max_pct", "Freizeit"),
|
|
}
|
|
|
|
def recompute(
|
|
self,
|
|
month,
|
|
distributable_total: Decimal,
|
|
total_income: Decimal,
|
|
reason: str = "",
|
|
) -> list[AllocationSuggestion]:
|
|
distributable_total = max(Decimal("0.00"), to_decimal(distributable_total))
|
|
total_income = max(Decimal("0.00"), to_decimal(total_income))
|
|
targets = self._target_accounts()
|
|
allocations = {item.target_account.slug: item for item in month.allocations if item.target_account}
|
|
suggested_values = {slug: Decimal("0.00") for slug in self.TARGET_SLUGS}
|
|
|
|
for slug in self.PRIORITY_TARGETS:
|
|
allocation = allocations.get(slug)
|
|
current_amount = to_decimal(allocation.amount) if allocation is not None else Decimal("0.00")
|
|
target_amount = self._minimum_target_amount(total_income, month, slug)
|
|
suggested_values[slug] = max(current_amount, target_amount) if current_amount > target_amount else target_amount
|
|
|
|
for slug in self.PERSONAL_TARGETS:
|
|
allocation = allocations.get(slug)
|
|
if allocation is not None:
|
|
suggested_values[slug] = to_decimal(allocation.amount)
|
|
|
|
AllocationSuggestion.query.filter_by(month_id=month.id).delete()
|
|
suggestions = []
|
|
for account in targets:
|
|
rule = self.month_target_rule(month, account.slug)
|
|
reason_text = reason or self._reason_for(account.slug, rule, month.personal_split_desi_pct)
|
|
suggestion = AllocationSuggestion(
|
|
month_id=month.id,
|
|
target_account_id=account.id,
|
|
suggested_amount=max(Decimal("0.00"), to_decimal(suggested_values.get(account.slug, 0))),
|
|
reason=reason_text,
|
|
strategy_key="income-targets-then-personal-split",
|
|
)
|
|
db.session.add(suggestion)
|
|
suggestions.append(suggestion)
|
|
db.session.flush()
|
|
return suggestions
|
|
|
|
def accept_all(self, month) -> None:
|
|
suggestions = {
|
|
item.target_account_id: item
|
|
for item in AllocationSuggestion.query.filter_by(month_id=month.id).all()
|
|
}
|
|
allocations = {
|
|
item.target_account_id: item for item in MonthlyAllocation.query.filter_by(month_id=month.id).all()
|
|
}
|
|
for target_account in self._target_accounts():
|
|
suggestion = suggestions.get(target_account.id)
|
|
allocation = allocations.get(target_account.id)
|
|
if not suggestion:
|
|
continue
|
|
if allocation is None:
|
|
allocation = MonthlyAllocation(
|
|
month_id=month.id,
|
|
target_account_id=target_account.id,
|
|
label=target_account.name,
|
|
sort_order=target_account.sort_order,
|
|
source="accepted_suggestion",
|
|
amount=suggestion.suggested_amount,
|
|
)
|
|
db.session.add(allocation)
|
|
continue
|
|
if allocation.is_locked:
|
|
continue
|
|
allocation.amount = suggestion.suggested_amount
|
|
allocation.source = "accepted_suggestion"
|
|
|
|
def accept_single(self, month, account_id: int) -> None:
|
|
suggestion = AllocationSuggestion.query.filter_by(
|
|
month_id=month.id, target_account_id=account_id
|
|
).first()
|
|
if not suggestion:
|
|
return
|
|
allocation = MonthlyAllocation.query.filter_by(
|
|
month_id=month.id, target_account_id=account_id
|
|
).first()
|
|
if allocation is None:
|
|
account = db.session.get(Account, account_id)
|
|
allocation = MonthlyAllocation(
|
|
month_id=month.id,
|
|
target_account_id=account_id,
|
|
label=account.name,
|
|
amount=suggestion.suggested_amount,
|
|
source="accepted_suggestion",
|
|
sort_order=account.sort_order,
|
|
)
|
|
db.session.add(allocation)
|
|
return
|
|
if not allocation.is_locked:
|
|
allocation.amount = suggestion.suggested_amount
|
|
allocation.source = "accepted_suggestion"
|
|
|
|
def strategy_hint(self, month, slug: str) -> dict[str, str | Decimal] | None:
|
|
rule = self.month_target_rule(month, slug)
|
|
if rule is None:
|
|
return None
|
|
return {
|
|
"label": rule.get("label", slug.title()),
|
|
"range_label": f"{int(rule['min_pct'])} bis {int(rule['max_pct'])} %",
|
|
"min_pct": to_decimal(rule["min_pct"]),
|
|
"max_pct": to_decimal(rule["max_pct"]),
|
|
}
|
|
|
|
def month_target_rule(self, month, slug: str) -> dict | None:
|
|
field_map = self.FIELD_MAP.get(slug)
|
|
base_rule = self.target_rules.get(slug)
|
|
if field_map is None or base_rule is None:
|
|
return base_rule
|
|
min_field, max_field, label = field_map
|
|
min_pct = to_decimal(getattr(month, min_field, Decimal(str(base_rule["min_pct"] * 100))))
|
|
max_pct = to_decimal(getattr(month, max_field, Decimal(str(base_rule["max_pct"] * 100))))
|
|
return {
|
|
"label": label,
|
|
"min_pct": min_pct,
|
|
"max_pct": max_pct,
|
|
"recommended_pct": ((min_pct + max_pct) / Decimal("2.00")).quantize(Decimal("0.01")),
|
|
}
|
|
|
|
def _target_accounts(self) -> list[Account]:
|
|
stmt = (
|
|
select(Account)
|
|
.where(Account.slug.in_(self.TARGET_SLUGS), Account.is_active.is_(True))
|
|
.order_by(Account.sort_order.asc(), Account.id.asc())
|
|
)
|
|
return list(db.session.scalars(stmt))
|
|
|
|
def _target_amount(self, total_income: Decimal, month, slug: str) -> Decimal:
|
|
rule = self.month_target_rule(month, slug)
|
|
if rule is None:
|
|
return Decimal("0.00")
|
|
recommended_pct = to_decimal(rule["recommended_pct"]) / Decimal("100.00")
|
|
return (total_income * recommended_pct).quantize(
|
|
Decimal("0.01"), rounding=ROUND_HALF_UP
|
|
)
|
|
|
|
def _minimum_target_amount(self, total_income: Decimal, month, slug: str) -> Decimal:
|
|
rule = self.month_target_rule(month, slug)
|
|
if rule is None:
|
|
return Decimal("0.00")
|
|
min_pct = to_decimal(rule["min_pct"]) / Decimal("100.00")
|
|
return (total_income * min_pct).quantize(
|
|
Decimal("0.01"), rounding=ROUND_HALF_UP
|
|
)
|
|
|
|
def _personal_split_values(
|
|
self,
|
|
remaining_pool: Decimal,
|
|
desi_pct: Decimal,
|
|
unlocked_targets: list[str],
|
|
) -> dict[str, Decimal]:
|
|
if remaining_pool <= Decimal("0.00") or not unlocked_targets:
|
|
return {slug: Decimal("0.00") for slug in self.PERSONAL_TARGETS}
|
|
|
|
desi_pct = min(Decimal("100.00"), max(Decimal("0.00"), to_decimal(desi_pct)))
|
|
flo_pct = Decimal("100.00") - desi_pct
|
|
weights = {
|
|
"persoenlich-flo": flo_pct,
|
|
"persoenlich-desi": desi_pct,
|
|
}
|
|
unlocked_weight_total = sum((weights[slug] for slug in unlocked_targets), Decimal("0.00"))
|
|
if unlocked_weight_total <= Decimal("0.00"):
|
|
even_share = (remaining_pool / Decimal(len(unlocked_targets))).quantize(
|
|
Decimal("0.01"), rounding=ROUND_HALF_UP
|
|
)
|
|
result = {slug: Decimal("0.00") for slug in self.PERSONAL_TARGETS}
|
|
assigned = Decimal("0.00")
|
|
for index, slug in enumerate(unlocked_targets):
|
|
if index == len(unlocked_targets) - 1:
|
|
result[slug] = remaining_pool - assigned
|
|
else:
|
|
result[slug] = even_share
|
|
assigned += even_share
|
|
return result
|
|
|
|
result = {slug: Decimal("0.00") for slug in self.PERSONAL_TARGETS}
|
|
assigned = Decimal("0.00")
|
|
for index, slug in enumerate(unlocked_targets):
|
|
if index == len(unlocked_targets) - 1:
|
|
amount = remaining_pool - assigned
|
|
else:
|
|
ratio = weights[slug] / unlocked_weight_total
|
|
amount = (remaining_pool * ratio).quantize(Decimal("0.01"), rounding=ROUND_HALF_UP)
|
|
assigned += amount
|
|
result[slug] = max(Decimal("0.00"), amount)
|
|
return result
|
|
|
|
def _reason_for(self, slug: str, rule: dict | None, desi_pct: Decimal) -> str:
|
|
if slug in self.PRIORITY_TARGETS and rule is not None:
|
|
return (
|
|
f"Zielbereich {to_decimal(rule['min_pct'])} bis {to_decimal(rule['max_pct'])} % "
|
|
f"vom Einkommen."
|
|
)
|
|
flo_pct = Decimal("100.00") - to_decimal(desi_pct)
|
|
personal_labels = personal_account_names()
|
|
return (
|
|
"Restbetrag nach Sparen, Urlaub und Freizeit mit Split "
|
|
f"{personal_labels['persoenlich-flo']} {flo_pct} % / "
|
|
f"{personal_labels['persoenlich-desi']} {to_decimal(desi_pct)} %."
|
|
)
|