impl/python.py

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 List, Optional, Sequence

from .dates_add_days import epoch_day_from_iso  ← from dates.add-days ^1.0.0 · built alongside by fune
from .inventory_valuation_weighted_average_types import AverageCostValuation, StockMovement
from .math_round_div import RoundingMode, round_div  ← from math.round-div ^1.0.0 · built alongside by fune
from .money_amount import Money, assert_same_currency, money  ← from money.amount ^1.0.0 · built alongside by fune


def weighted_average_valuation(
    movements: Sequence[StockMovement], currency: str, mode: RoundingMode
) -> AverageCostValuation:
    """Perpetual weighted average cost. The stock's value is held exactly and
    each issue takes value x issued / on hand, rounded once, so the last issue
    takes exactly what is left and an empty item is worth nothing.
    """
    zero = money(0, currency)
    round_div(0, 1, mode)  # an unknown mode fails even on a ledger with no issues
    quantity = 0
    value = 0
    cost_of_sales = 0
    issue_costs: List[Money] = []
    previous_day: Optional[int] = None
    previous_date = ""
    for line in movements:
        day = epoch_day_from_iso(line.date)
        if previous_day is not None and day < previous_day:
            raise ValueError("movements must be in date order: %s comes after %s" % (line.date, previous_date))
        previous_day = day
        previous_date = line.date
        if isinstance(line.quantity, bool) or not isinstance(line.quantity, int) or line.quantity == 0:
            raise ValueError(
                "quantity must be a non-zero whole number, received %r on %s" % (line.quantity, line.date)
            )
        if line.quantity > 0:
            if line.unit_cost is None:
                raise ValueError("a receipt needs a unitCost: %d on %s" % (line.quantity, line.date))
            assert_same_currency(zero, line.unit_cost)
            if line.unit_cost.minor < 0:
                raise ValueError(
                    "unitCost must not be negative, received %d on %s" % (line.unit_cost.minor, line.date)
                )
            quantity += line.quantity
            value += line.quantity * line.unit_cost.minor
            continue
        if line.unit_cost is not None:
            raise ValueError(
                "an issue takes its cost from stock, so its unitCost must be null: %d on %s"
                % (line.quantity, line.date)
            )
        issued = -line.quantity
        if issued > quantity:
            raise ValueError(
                "insufficient stock: an issue of %d on %s exceeds the %d on hand" % (issued, line.date, quantity)
            )
        cost = round_div(value * issued, quantity, mode)
        issue_costs.append(money(cost, currency))
        cost_of_sales += cost
        value -= cost
        quantity -= issued
    return AverageCostValuation(
        closing_quantity=quantity,
        closing_value=money(value, currency),
        cost_of_sales=money(cost_of_sales, currency),
        issue_costs=issue_costs,
        average_unit_cost=money(0 if quantity == 0 else round_div(value, quantity, mode), currency),
    )