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 Callable, Dict, List, Sequence, Tuple

from .dates_add_days import epoch_day_from_iso  ← from dates.add-days ^1.0.0 · built alongside by fune
from .invest_money_weighted_return_types import DatedFlow, XirrResult
from .math_fractional_power import FIXED_SCALE, pow_fixed  ← from math.fractional-power ^1.0.0 · built alongside by fune

_MAX_SPAN_DAYS = 36500
_NO_SINGLE_RATE = "no single rate of return solves these cash flows"


def _sign(value: int) -> int:
    return (value > 0) - (value < 0)


def _round_half_away(n: int, d: int) -> int:
    """n / d rounded half away from zero; d > 0."""
    magnitude = (abs(n) * 2 + d) // (2 * d)
    return -magnitude if n < 0 else magnitude


def _net_flows(flows: Sequence[DatedFlow]) -> List[Tuple[int, int]]:
    """Net flow per day, in day order, counted from the earliest date with a nonzero net flow."""
    if len(flows) == 0:
        raise ValueError("flows must not be empty")
    currency = flows[0].amount.currency
    net: Dict[int, int] = {}
    for flow in flows:
        amount = flow.amount
        if amount.currency != currency:
            raise ValueError("currency mismatch: %s and %s" % (currency, amount.currency))
        if isinstance(amount.minor, bool) or not isinstance(amount.minor, int):
            raise ValueError("amounts must be whole minor units, received %r" % (amount.minor,))
        day = epoch_day_from_iso(flow.date)
        net[day] = net.get(day, 0) + amount.minor
    days = sorted((d, a) for d, a in net.items() if a != 0)
    if not any(a > 0 for _, a in days) or not any(a < 0 for _, a in days):
        raise ValueError("flows need at least one payment and one receipt")
    first = days[0][0]
    if days[-1][0] - first > _MAX_SPAN_DAYS:
        raise ValueError("flows must fall within %d days of each other" % _MAX_SPAN_DAYS)
    return [(d - first, a) for d, a in days]


def _bisect(f: Callable[[int], int]) -> int:
    """Bisection on x in [0, FIXED_SCALE] for a change of sign of f; the two ends' signs differ."""
    lo, hi = 0, FIXED_SCALE
    lo_sign = _sign(f(lo))
    while hi - lo > 1:
        mid = (lo + hi) // 2
        s = _sign(f(mid))
        if s == 0:
            return mid
        if s == lo_sign:
            lo = mid
        else:
            hi = mid
    return lo


def money_weighted_return(flows: Sequence[DatedFlow]) -> XirrResult:
    """XIRR: the annual rate r at which sum(amount / (1 + r)^(days / 365)) = 0,
    days counted from the earliest flow. Solved by bisection in 18-place fixed
    point on the per-day discount factor, then settled to 12 places and rounded
    half away from zero to 9 places and to a basis point."""
    net = _net_flows(flows)
    total = sum(a for _, a in net)
    rate = 0
    if total != 0:
        last = net[-1][0]
        positive_side = _sign(net[0][1]) != _sign(total)
        negative_side = _sign(net[-1][1]) != _sign(total)
        if positive_side == negative_side:
            raise ValueError(_NO_SINGLE_RATE)
        if positive_side:
            # v = (1 + r)^(-1/365) in (0, 1).
            v = _bisect(lambda x: sum(a * pow_fixed(x, t) for t, a in net))
            growth = pow_fixed(v, 365)
            if growth == 0:
                raise ValueError("the rate of return is too large to compute")
            rate = FIXED_SCALE * FIXED_SCALE // growth - FIXED_SCALE
        else:
            # u = (1 + r)^(1/365) in (0, 1); the sum is multiplied through by u^last.
            u = _bisect(lambda x: sum(a * pow_fixed(x, last - t) for t, a in net))
            rate = pow_fixed(u, 365) - FIXED_SCALE
    settled = _round_half_away(rate, 10 ** 6)
    bp = _round_half_away(settled * 10000, 10 ** 12)
    nine = _round_half_away(settled, 1000)
    magnitude = abs(nine)
    return XirrResult(
        basis_points=bp,
        rate="%s%d.%09d" % ("-" if nine < 0 else "", magnitude // 10 ** 9, magnitude % 10 ** 9),
    )