andy
db06108d2b
Adds remaining files from parallel development: - Services: analysis, csv_reader, forecasting, normalizer, recurring - UI: recurring_view, settings_view, sidebar, themes (dark/light) - Tests: analysis, csv_reader, forecasting, import_categorize, normalizer, recurring, integration - App entry point (main.py) and CLAUDE.md 52 tests passing across all modules. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
70 lines
2.3 KiB
Python
70 lines
2.3 KiB
Python
# src/services/recurring.py
|
|
from collections import defaultdict
|
|
from sqlalchemy.orm import Session
|
|
|
|
from src.models.transaction import Transaction
|
|
|
|
|
|
class RecurringDetector:
|
|
def __init__(self, session: Session):
|
|
self.session = session
|
|
|
|
def detect(self, amount_tolerance: float = 0.10) -> list[dict]:
|
|
txns = (
|
|
self.session.query(Transaction)
|
|
.filter(Transaction.amount < 0)
|
|
.filter(Transaction.is_transfer == False)
|
|
.order_by(Transaction.date)
|
|
.all()
|
|
)
|
|
|
|
groups: dict[str, list[Transaction]] = defaultdict(list)
|
|
for txn in txns:
|
|
groups[txn.description].append(txn)
|
|
|
|
results = []
|
|
for desc, group in groups.items():
|
|
if len(group) < 2:
|
|
continue
|
|
|
|
amounts = [abs(float(t.amount)) for t in group]
|
|
avg_amount = sum(amounts) / len(amounts)
|
|
|
|
if any(abs(a - avg_amount) / avg_amount > amount_tolerance for a in amounts):
|
|
continue
|
|
|
|
dates = sorted(t.date for t in group)
|
|
intervals = [(dates[i + 1] - dates[i]).days for i in range(len(dates) - 1)]
|
|
avg_interval = sum(intervals) / len(intervals)
|
|
|
|
frequency = self._classify_frequency(avg_interval)
|
|
if frequency is None:
|
|
continue
|
|
|
|
annual_multiplier = {"weekly": 52, "biweekly": 26, "monthly": 12, "quarterly": 4, "annual": 1}
|
|
|
|
results.append({
|
|
"description": desc,
|
|
"typical_amount": round(avg_amount, 2),
|
|
"frequency": frequency,
|
|
"annual_cost": round(avg_amount * annual_multiplier[frequency], 2),
|
|
"occurrences": len(group),
|
|
"last_date": max(dates),
|
|
})
|
|
|
|
results.sort(key=lambda r: r["annual_cost"], reverse=True)
|
|
return results
|
|
|
|
def _classify_frequency(self, avg_days: float) -> str | None:
|
|
if 5 <= avg_days <= 9:
|
|
return "weekly"
|
|
elif 12 <= avg_days <= 16:
|
|
return "biweekly"
|
|
elif 25 <= avg_days <= 35:
|
|
return "monthly"
|
|
elif 80 <= avg_days <= 100:
|
|
return "quarterly"
|
|
elif 350 <= avg_days <= 380:
|
|
return "annual"
|
|
return None
|