93 lines
2.8 KiB
Python
93 lines
2.8 KiB
Python
"""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
|
|
max_days_overdue: 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,
|
|
'max_days_overdue': self.max_days_overdue,
|
|
'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
|
|
if days_overdue > report.max_days_overdue:
|
|
report.max_days_overdue = days_overdue
|
|
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
|