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)} %." )