"""PRD §5.4 FRFM014/018-025 + FRFM031-039 — finance analytics. Pure-function analytics over the existing finance schema. No new persisted models — every helper rebuilds from base data so reports never go stale. Two main groupings: - ``compute_variance`` / ``aggregate_variance_snapshot``: shared by the ``scan_budget_variances`` Celery beat and the new finance dashboard view. - ``cost_breakdown_by_category`` / ``spend_curve`` / ``donor_finance_packet``: power the variance dashboard, donor reporting export, and (optionally) cross-system donor reporting via ``apps.mel.reports.builders``. """ from __future__ import annotations from collections import OrderedDict, defaultdict from dataclasses import dataclass, field from datetime import date, timedelta from decimal import Decimal from typing import Iterable from django.db.models import Sum from django.utils import timezone from apps.rims.finance.models import ( Budget, BudgetLine, DisbursementRequest, Expenditure, FundingSource, ) from apps.rims.finance.services import ( budget_ceiling, total_disbursed, total_expended, ) # --------------------------------------------------------------------------- # Variance — FRFM014 / FRFM018-025 # --------------------------------------------------------------------------- @dataclass(frozen=True) class BudgetVariance: """A budget's pacing snapshot at a point in time. ``gap`` is positive when the budget is over-running (spent more than elapsed), negative when under-running. """ budget_id: int budget_name: str award_id: int ceiling: Decimal disbursed: Decimal elapsed_ratio: float spent_ratio: float gap: float direction: str # "over", "under", "on-track" currency: str = "" @property def gap_pct(self) -> float: return self.gap * 100.0 def compute_variance(budget: Budget, *, today: date | None = None) -> BudgetVariance | None: """Compute the spend-vs-elapsed variance for a single budget. Returns ``None`` when the budget cannot meaningfully be analysed (no activation date, zero project window, zero ceiling). Pure function — no DB writes, no notifications. Shared with the :func:`apps.rims.finance.tasks.scan_budget_variances` celery beat so the operator dashboard and the alert path agree. """ today = today or timezone.now().date() award = budget.award if not award.activation_date or not award.project_end_date: return None total_window = (award.project_end_date - award.activation_date).days if total_window <= 0: return None elapsed = (today - award.activation_date).days if elapsed <= 0: return None ceiling = budget_ceiling(budget) if not ceiling: return None disbursed = total_disbursed(budget) elapsed_ratio = min(elapsed / total_window, 1.0) spent_ratio = float(disbursed) / float(ceiling) if ceiling else 0.0 gap = spent_ratio - elapsed_ratio direction = "over" if gap > 0.001 else ("under" if gap < -0.001 else "on-track") return BudgetVariance( budget_id=budget.pk, budget_name=budget.name, award_id=award.pk, ceiling=Decimal(ceiling), disbursed=Decimal(disbursed), elapsed_ratio=elapsed_ratio, spent_ratio=spent_ratio, gap=gap, direction=direction, currency=budget.currency or "", ) def aggregate_variance_snapshot( *, threshold_ratio: float = 0.10, today: date | None = None ) -> list[BudgetVariance]: """Return all active-award budgets whose variance breaches the threshold. ``threshold_ratio`` is a fraction (0.10 = 10%). Sorted by absolute gap descending so worst offenders surface first. """ today = today or timezone.now().date() qs = Budget.objects.select_related("award__application__call").filter( award__status__in=["active", "ready_for_closeout"] ) flagged: list[BudgetVariance] = [] for budget in qs: v = compute_variance(budget, today=today) if v is None: continue if abs(v.gap) >= threshold_ratio: flagged.append(v) flagged.sort(key=lambda v: abs(v.gap), reverse=True) return flagged # --------------------------------------------------------------------------- # Cost breakdown — FRFM018 (payable identification) / FRFM019 / FRFM022 # --------------------------------------------------------------------------- @dataclass(frozen=True) class CategorySlice: category: str label: str budgeted: Decimal expended: Decimal @property def utilisation(self) -> float: return float(self.expended / self.budgeted) if self.budgeted else 0.0 def cost_breakdown_by_category(budget: Budget) -> list[CategorySlice]: """Per-category roll-up of budgeted vs actually expended for one budget.""" cats: "OrderedDict[str, dict]" = OrderedDict() for choice_value, choice_label in BudgetLine.Category.choices: cats[choice_value] = {"label": choice_label, "budgeted": Decimal("0"), "expended": Decimal("0")} line_totals = ( BudgetLine.objects.filter(budget=budget) .values("category") .annotate(total=Sum("amount")) ) for row in line_totals: cat = row["category"] or BudgetLine.Category.OTHER cats.setdefault(cat, {"label": cat, "budgeted": Decimal("0"), "expended": Decimal("0")}) cats[cat]["budgeted"] = Decimal(row["total"] or 0) expended_totals = ( Expenditure.objects.filter(budget_line__budget=budget) .values("budget_line__category") .annotate(total=Sum("amount")) ) for row in expended_totals: cat = row["budget_line__category"] or BudgetLine.Category.OTHER cats.setdefault(cat, {"label": cat, "budgeted": Decimal("0"), "expended": Decimal("0")}) cats[cat]["expended"] = Decimal(row["total"] or 0) return [ CategorySlice(category=key, label=val["label"], budgeted=val["budgeted"], expended=val["expended"]) for key, val in cats.items() if val["budgeted"] or val["expended"] ] # --------------------------------------------------------------------------- # Spend curve — FRFM023 (cashflow validation) / FRFM050 (financial reports) # --------------------------------------------------------------------------- def spend_curve(budget: Budget, *, bucket: str = "month") -> list[dict]: """Time-series of disbursed + expended amounts per ``bucket``. ``bucket`` accepts ``"month"`` (default) or ``"week"``. Each row is ``{"label": "YYYY-MM" or "YYYY-Www", "disbursed": float, "expended": float}``. """ if bucket not in {"month", "week"}: raise ValueError("bucket must be 'month' or 'week'") def _key(d: date) -> str: if bucket == "month": return d.strftime("%Y-%m") # ISO week, Monday start iso_year, iso_week, _ = d.isocalendar() return f"{iso_year}-W{iso_week:02d}" buckets: "OrderedDict[str, dict]" = OrderedDict() for d in ( DisbursementRequest.objects.filter( budget=budget, status__in=[DisbursementRequest.Status.APPROVED, DisbursementRequest.Status.PAID], approved_at__isnull=False, ) .values_list("approved_at", "amount") ): approved_at, amount = d key = _key(approved_at.date()) buckets.setdefault(key, {"label": key, "disbursed": Decimal("0"), "expended": Decimal("0")}) buckets[key]["disbursed"] += Decimal(amount or 0) for e in ( Expenditure.objects.filter(budget_line__budget=budget).values_list("incurred_on", "amount") ): incurred_on, amount = e if not incurred_on: continue key = _key(incurred_on) buckets.setdefault(key, {"label": key, "disbursed": Decimal("0"), "expended": Decimal("0")}) buckets[key]["expended"] += Decimal(amount or 0) rows = sorted(buckets.values(), key=lambda r: r["label"]) return [ { "label": row["label"], "disbursed": float(row["disbursed"]), "expended": float(row["expended"]), } for row in rows ] # --------------------------------------------------------------------------- # Donor reporting — FRFM031-039 / FRFM050-051 # --------------------------------------------------------------------------- @dataclass class DonorFinancePacket: funding_source_id: int | None funding_source_name: str period_start: date | None period_end: date | None budget_count: int = 0 ceiling_total: Decimal = field(default_factory=lambda: Decimal("0")) disbursed_total: Decimal = field(default_factory=lambda: Decimal("0")) expended_total: Decimal = field(default_factory=lambda: Decimal("0")) rows: list[dict] = field(default_factory=list) @property def utilisation(self) -> float: return float(self.disbursed_total / self.ceiling_total) if self.ceiling_total else 0.0 def _budgets_for_donor(donor: FundingSource): """Resolve every budget that draws on this donor (header OR any line).""" direct_ids = list(Budget.objects.filter(funding_source=donor).values_list("pk", flat=True)) line_ids = list( BudgetLine.objects.filter(funding_source=donor).values_list("budget_id", flat=True).distinct() ) return Budget.objects.filter(pk__in=set(direct_ids) | set(line_ids)).select_related( "award__application__call" ) def donor_finance_packet( donor: FundingSource, *, period_start: date | None = None, period_end: date | None = None, closeout_only: bool = False, ) -> DonorFinancePacket: """Assemble a finance packet for a donor across the optional period. ``period_start`` / ``period_end`` constrain the disbursement and expenditure totals (inclusive on both ends). Ceilings always reflect the budget headers — period filtering doesn't reduce a contractual ceiling. When ``closeout_only=True`` (PRD §5.1 FRFA-CO019), rows are restricted to awards with status=CLOSED and each row carries close-out-specific fields: ``closed_at``, ``archived_at``, ``residual_returned``, ``forced_closure``. """ packet = DonorFinancePacket( funding_source_id=donor.pk, funding_source_name=donor.name, period_start=period_start, period_end=period_end, ) budgets = list(_budgets_for_donor(donor)) if closeout_only: from apps.rims.grants.models import Award budgets = [b for b in budgets if b.award.status == Award.Status.CLOSED] packet.budget_count = len(budgets) for budget in budgets: ceiling = Decimal(budget_ceiling(budget) or 0) disb_qs = DisbursementRequest.objects.filter( budget=budget, status__in=[DisbursementRequest.Status.APPROVED, DisbursementRequest.Status.PAID], ) exp_qs = Expenditure.objects.filter(budget_line__budget=budget) if period_start: disb_qs = disb_qs.filter(approved_at__date__gte=period_start) exp_qs = exp_qs.filter(incurred_on__gte=period_start) if period_end: disb_qs = disb_qs.filter(approved_at__date__lte=period_end) exp_qs = exp_qs.filter(incurred_on__lte=period_end) disb = Decimal(disb_qs.aggregate(t=Sum("amount"))["t"] or 0) exp = Decimal(exp_qs.aggregate(t=Sum("amount"))["t"] or 0) packet.ceiling_total += ceiling packet.disbursed_total += disb packet.expended_total += exp award = budget.award applicant = getattr(getattr(award, "application", None), "applicant", None) row = { "budget_id": budget.pk, "budget_name": budget.name, "award_id": award.pk, "awardee": ( applicant.get_full_name() or applicant.email if applicant else f"Award #{award.pk}" ), "currency": budget.currency, "ceiling": ceiling, "disbursed": disb, "expended": exp, } if closeout_only: row["closed_at"] = award.closed_at row["archived_at"] = award.archived_at closeout = getattr(award, "closeout_record", None) snap = (closeout.financial_snapshot or {}) if closeout else {} row["residual_returned"] = snap.get("residual_returned", "") forced_just = getattr(closeout, "forced_closure_justification", "") if closeout else "" row["forced_closure"] = bool(forced_just) row["forced_closure_justification"] = forced_just packet.rows.append(row) return packet def donor_finance_summary(*, period_start: date | None = None, period_end: date | None = None) -> list[DonorFinancePacket]: """Build a packet for every donor that has at least one budget reference. Used by the donor-reports list page to render a roll-up table. """ return [ donor_finance_packet(d, period_start=period_start, period_end=period_end) for d in FundingSource.objects.order_by("name") ] # --------------------------------------------------------------------------- # PRD §5.4 FRFM020 / FRFM021 / FRFM023 / FRFM025 — variance + cost analytics # detail rows. These power the per-budget detail tables on the analytics # dashboard and the donor close-out report. # --------------------------------------------------------------------------- def variance_detail_rows(budget) -> list[dict]: """PRD §5.4 FRFM020 — per-line variance breakdown for a single budget. Returns one row per BudgetLine with: {"line": BudgetLine, "budgeted": Decimal, "expended": Decimal, "remaining": Decimal, "pct_used": float} """ rows: list[dict] = [] for line in budget.lines.prefetch_related("expenditures").all(): budgeted = Decimal(line.amount or 0) expended = sum( (Decimal(exp.amount or 0) for exp in line.expenditures.all()), Decimal("0"), ) remaining = budgeted - expended pct = float(expended / budgeted * 100) if budgeted else 0.0 rows.append( { "line": line, "budgeted": budgeted, "expended": expended, "remaining": remaining, "pct_used": round(pct, 1), } ) return rows def cashflow_lookback(budget, *, months: int = 6) -> list[dict]: """PRD §5.4 FRFM023 — N-month look-back of disbursed vs expended flows. Wraps spend_curve() and trims to the last ``months`` buckets. """ full = spend_curve(budget, bucket="month") return full[-months:] if months > 0 else full def cost_centre_breakdown(budget) -> list[dict]: """PRD §5.4 FRFM025 — totals grouped by budget-line category (cost centre). Returns rows: [{"category": "personnel", "label": "Personnel", "budgeted": Decimal, "expended": Decimal}, ...] """ from collections import defaultdict from apps.rims.finance.models import BudgetLine rollup: dict[str, dict] = defaultdict( lambda: {"budgeted": Decimal("0"), "expended": Decimal("0")} ) for line in budget.lines.prefetch_related("expenditures").all(): cat = line.category or BudgetLine.Category.OTHER rollup[cat]["budgeted"] += Decimal(line.amount or 0) for exp in line.expenditures.all(): rollup[cat]["expended"] += Decimal(exp.amount or 0) labels = dict(BudgetLine.Category.choices) return [ { "category": cat, "label": labels.get(cat, cat.title()), "budgeted": vals["budgeted"], "expended": vals["expended"], } for cat, vals in sorted(rollup.items()) ]