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 .invest_time_weighted_return_types import TwrResult, ValuationPoint
from .math_fractional_power import FIXED_SCALE, fractional_power_fixed ← from math.fractional-power ^1.0.0 · built alongside by fune
_MAX_SPAN_DAYS = 36500
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 _whole(minor: int) -> int:
if isinstance(minor, bool) or not isinstance(minor, int):
raise ValueError("amounts must be whole minor units, received %r" % (minor,))
return minor
def time_weighted_return(points: Sequence[ValuationPoint]) -> TwrResult:
"""True time-weighted return: each sub-period's growth, value_i over the
previous value plus the previous flow, linked geometrically as an exact
fraction and rounded once to basis points, half away from zero."""
if len(points) < 2:
raise ValueError("at least two valuation points are needed")
currency = points[0].value.currency
days: List[int] = []
for point in points:
for amount in (point.value, point.flow):
if amount.currency != currency:
raise ValueError("currency mismatch: %s and %s" % (currency, amount.currency))
if _whole(point.value.minor) < 0:
raise ValueError("valuations must not be negative, received %d on %s" % (point.value.minor, point.date))
_whole(point.flow.minor)
day = epoch_day_from_iso(point.date)
if days and day <= days[-1]:
raise ValueError("dates must be strictly increasing, one point per date")
days.append(day)
span = days[-1] - days[0]
if span > _MAX_SPAN_DAYS:
raise ValueError("points must fall within %d days" % _MAX_SPAN_DAYS)
numerator, denominator = 1, 1
sub_period_basis_points: List[int] = []
for i in range(1, len(points)):
invested = points[i - 1].value.minor + points[i - 1].flow.minor
if invested <= 0:
raise ValueError("nothing is invested at the start of the sub-period from %s" % points[i - 1].date)
value = points[i].value.minor
sub_period_basis_points.append(_round_half_away((value - invested) * 10000, invested))
numerator *= value
denominator *= invested
annualised: Optional[int] = None
if span >= 365:
if numerator == 0:
annualised = -10000
else:
growth = fractional_power_fixed(numerator * FIXED_SCALE // denominator, 365, span)
annualised = _round_half_away((growth - FIXED_SCALE) * 10000, FIXED_SCALE)
return TwrResult(
basis_points=_round_half_away((numerator - denominator) * 10000, denominator),
sub_period_basis_points=sub_period_basis_points,
days=span,
annualised_basis_points=annualised,
)