53 lines
1.8 KiB
Python
53 lines
1.8 KiB
Python
import statistics
|
|
from collections import Counter, defaultdict
|
|
from decimal import Decimal
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.tables import RecurringItem, Transaction
|
|
|
|
|
|
def _max_consecutive_months(months: list[tuple[int, int]]) -> int:
|
|
if not months:
|
|
return 0
|
|
best = current = 1
|
|
for prev, cur in zip(months, months[1:]):
|
|
prev_idx = prev[0] * 12 + prev[1]
|
|
cur_idx = cur[0] * 12 + cur[1]
|
|
current = current + 1 if cur_idx == prev_idx + 1 else 1
|
|
best = max(best, current)
|
|
return best
|
|
|
|
|
|
def suggest_recurring(session: Session) -> list[dict]:
|
|
txs = session.execute(
|
|
select(Transaction).where(Transaction.status == "confirmed")
|
|
).scalars().all()
|
|
groups: dict[tuple, list[Transaction]] = defaultdict(list)
|
|
for t in txs:
|
|
groups[(t.account_id, t.counterparty, t.amount)].append(t)
|
|
|
|
existing = {(r.name, Decimal(r.amount))
|
|
for r in session.execute(select(RecurringItem)).scalars()}
|
|
|
|
suggestions: list[dict] = []
|
|
for (_account_id, counterparty, amount), items in groups.items():
|
|
months = sorted({(t.booking_date.year, t.booking_date.month) for t in items})
|
|
if _max_consecutive_months(months) < 3:
|
|
continue
|
|
name = counterparty
|
|
if (name, Decimal(amount)) in existing:
|
|
continue
|
|
due_day = int(statistics.median(sorted(t.booking_date.day for t in items)))
|
|
cat_counts = Counter(t.category_id for t in items if t.category_id is not None)
|
|
category_id = cat_counts.most_common(1)[0][0] if cat_counts else None
|
|
suggestions.append({
|
|
"name": name,
|
|
"amount": Decimal(amount),
|
|
"rhythm": "monthly",
|
|
"due_day": due_day,
|
|
"category_id": category_id,
|
|
})
|
|
return suggestions
|