Functional Weave
Code in Python

inventory.valuation-fifo@1.0.0

impl/python.py

3,169 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 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_fifo_types import FifoValuation, StockLayer, StockMovement
from .money_amount import Money, assert_same_currency, money  ← from money.amount ^1.0.0 · built alongside by fune


def fifo_valuation(movements: Sequence[StockMovement], currency: str) -> FifoValuation:
    """Value stock first in, first out, and cost each issue from the oldest layers.

    Every cost is a whole quantity times a unit cost, so nothing is rounded and
    closing value plus cost of sales always equals what was received.
    """
    zero = money(0, currency)
    # [date, quantity, unit cost]: mutable while issues consume them.
    layers: List[list] = []
    issue_costs: List[Money] = []
    cost_of_sales = 0
    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)
                )
            layers.append([line.date, line.quantity, line.unit_cost])
            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)
            )
        wanted = -line.quantity
        on_hand = sum(layer[1] for layer in layers)
        if wanted > on_hand:
            raise ValueError(
                "insufficient stock: an issue of %d on %s exceeds the %d on hand" % (wanted, line.date, on_hand)
            )
        cost = 0
        while wanted > 0:
            oldest = layers[0]
            used = min(wanted, oldest[1])
            cost += used * oldest[2].minor
            oldest[1] -= used
            wanted -= used
            if oldest[1] == 0:
                layers.pop(0)
        issue_costs.append(money(cost, currency))
        cost_of_sales += cost
    closing_quantity = sum(layer[1] for layer in layers)
    closing_value = sum(layer[1] * layer[2].minor for layer in layers)
    return FifoValuation(
        closing_quantity=closing_quantity,
        closing_value=money(closing_value, currency),
        cost_of_sales=money(cost_of_sales, currency),
        issue_costs=issue_costs,
        layers=[StockLayer(date=d, quantity=q, unit_cost=c) for d, q, c in layers],
    )