Functional Weave
Code in Rust

banking.fees-cap@1.0.0

impl/python.py

2,981 bytes · the Python implementation · view raw

Imports name this capability’s declared dependencies, which fune builds next to it in your project; each one links to its page.

from typing import Dict, List, Sequence

from .banking_fees_cap_types import CappedCharge, FeeCapResult, FeeCharge
from .dates_add_days import parse_iso_date  ← from dates.add-days ^1.0.0 · built alongside by fune
from .money_amount import Money, money  ← from money.amount ^1.0.0 · built alongside by fune


def _period_key(iso: str, cycle_day: int) -> str:
    """The charging period a date falls in, named by the year and month it
    starts in. With cycle_day 15, 2026-03-14 belongs to the period that began
    on 2026-02-15."""
    d = parse_iso_date(iso)
    year, month = d.year, d.month
    if d.day < cycle_day:
        month -= 1
        if month == 0:
            month = 12
            year -= 1
    return "%d-%d" % (year, month)


def apply_fee_cap(charges: Sequence[FeeCharge], monthly_cap: Money, cycle_day: int) -> FeeCapResult:
    """Apply a monthly cap to a list of charges.

    Within each charging period the charges are taken in date order (ties in
    input order) until the cap is used up: the charge that crosses the cap is
    reduced to what is left, and every later charge in that period is waived
    in full. The result keeps the input order.
    """
    if isinstance(cycle_day, bool) or not isinstance(cycle_day, int) or cycle_day < 1 or cycle_day > 28:
        raise ValueError("cycleDay must be 1 to 28, received %s" % (cycle_day,))
    if monthly_cap.minor < 0:
        raise ValueError("monthlyCap must not be negative, received %d" % (monthly_cap.minor,))
    currency = monthly_cap.currency
    keys: List[str] = []
    for charge in charges:
        if charge.amount.currency != currency:
            raise ValueError("currency mismatch: %s and %s" % (charge.amount.currency, currency))
        if charge.amount.minor < 0:
            raise ValueError('charge "%s" must not be negative, received %d' % (charge.label, charge.amount.minor))
        keys.append(_period_key(charge.date, cycle_day))

    # sorted() is stable, so equal dates keep their input order.
    order = sorted(range(len(charges)), key=lambda i: charges[i].date)

    used: Dict[str, int] = {}
    charged_minor = [0] * len(charges)
    for index in order:
        spent = used.get(keys[index], 0)
        take = min(charges[index].amount.minor, monthly_cap.minor - spent)
        charged_minor[index] = take
        used[keys[index]] = spent + take

    out: List[CappedCharge] = []
    total_charged = 0
    total_waived = 0
    for index, charge in enumerate(charges):
        charged = charged_minor[index]
        waived = charge.amount.minor - charged
        total_charged += charged
        total_waived += waived
        out.append(
            CappedCharge(
                date=charge.date,
                label=charge.label,
                amount=money(charge.amount.minor, currency),
                charged=money(charged, currency),
                waived=money(waived, currency),
            )
        )
    return FeeCapResult(
        charges=out,
        total_charged=money(total_charged, currency),
        total_waived=money(total_waived, currency),
    )