feat(fusion_accounting_followup): overdue_aging service with 6 buckets

Made-with: Cursor

fusion_accounting_followup/services/__init__.py

@@ -0,0 +1 @@
+from . import overdue_aging

fusion_accounting_followup/services/overdue_aging.py

@@ -0,0 +1,88 @@
+"""Aging bucket primitives.
+
+Pure-Python: callers pass a list of move-line dicts with `date_maturity`
+and `amount_residual`; we bucket them into 0/30/60/90/120+ days overdue."""
+
+from dataclasses import dataclass, field
+from datetime import date
+
+
+BUCKETS = [
+    ('current', 0, 0),
+    ('1_30', 1, 30),
+    ('31_60', 31, 60),
+    ('61_90', 61, 90),
+    ('91_120', 91, 120),
+    ('120_plus', 121, None),
+]
+
+
+@dataclass
+class AgingBucket:
+    name: str
+    days_min: int
+    days_max: int | None
+    amount: float = 0.0
+    line_count: int = 0
+
+
+@dataclass
+class AgingReport:
+    as_of: date
+    buckets: list[AgingBucket] = field(default_factory=list)
+    total_amount: float = 0.0
+    total_overdue_amount: float = 0.0
+    line_count: int = 0
+
+    def to_dict(self):
+        return {
+            'as_of': str(self.as_of),
+            'total_amount': self.total_amount,
+            'total_overdue_amount': self.total_overdue_amount,
+            'line_count': self.line_count,
+            'buckets': [{
+                'name': b.name, 'days_min': b.days_min, 'days_max': b.days_max,
+                'amount': b.amount, 'line_count': b.line_count,
+            } for b in self.buckets],
+        }
+
+
+def compute_aging(*, move_lines: list[dict], as_of: date | None = None) -> AgingReport:
+    """Bucket move-line dicts into aging brackets.
+
+    Each dict needs: date_maturity (date), amount_residual (float).
+    `as_of` defaults to today."""
+    as_of = as_of or date.today()
+    report = AgingReport(as_of=as_of)
+    for name, days_min, days_max in BUCKETS:
+        report.buckets.append(AgingBucket(name=name, days_min=days_min, days_max=days_max))
+
+    for ml in move_lines:
+        maturity = ml.get('date_maturity')
+        amount = ml.get('amount_residual', 0.0)
+        if not maturity:
+            continue
+        days_overdue = (as_of - maturity).days
+        bucket = _find_bucket(report.buckets, days_overdue)
+        if bucket:
+            bucket.amount += amount
+            bucket.line_count += 1
+        report.total_amount += amount
+        if days_overdue > 0:
+            report.total_overdue_amount += amount
+        report.line_count += 1
+
+    return report
+
+
+def _find_bucket(buckets: list[AgingBucket], days_overdue: int) -> AgingBucket | None:
+    if days_overdue <= 0:
+        return next((b for b in buckets if b.name == 'current'), None)
+    for b in buckets:
+        if b.name == 'current':
+            continue
+        if b.days_max is None and days_overdue >= b.days_min:
+            return b
+        if b.days_max is not None and b.days_min <= days_overdue <= b.days_max:
+            return b
+    return None