invest.rebalance Unreviewed

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, Sequence, Set

from .invest_rebalance_types import RebalanceHolding, RebalancePlan, RebalanceTrade
from .money_amount import Money, money  ← from money.amount ^1.0.0 · built alongside by fune


def _is_whole(value: object) -> bool:
    return isinstance(value, int) and not isinstance(value, bool)


def rebalance(holdings: Sequence[RebalanceHolding], cash: Money, minimum_trade: Money) -> RebalancePlan:
    """Whole-unit trades back to target weights. Target units are the nearest
    whole unit (a tie stays nearer the units held), trades below the minimum
    are dropped, and buys are then trimmed one unit at a time, from the holding
    that would end furthest above its target, until the cash covers them."""
    if len(holdings) == 0:
        raise ValueError("holdings must not be empty")
    currency = cash.currency
    if minimum_trade.currency != currency:
        raise ValueError("currency mismatch: %s and %s" % (currency, minimum_trade.currency))
    if not _is_whole(cash.minor) or cash.minor < 0:
        raise ValueError("cash must be whole minor units, 0 or more; received %s" % (cash.minor,))
    if not _is_whole(minimum_trade.minor) or minimum_trade.minor < 0:
        raise ValueError("minimumTrade must be whole minor units, 0 or more; received %s" % (minimum_trade.minor,))
    seen: Set[str] = set()
    total = cash.minor
    targets = 0
    for h in holdings:
        if h.id in seen:
            raise ValueError('duplicate holding id "%s"' % h.id)
        seen.add(h.id)
        if h.unit_price.currency != currency:
            raise ValueError("currency mismatch: %s and %s" % (currency, h.unit_price.currency))
        if not _is_whole(h.units) or h.units < 0:
            raise ValueError('units must be a whole number, 0 or more; received %s for "%s"' % (h.units, h.id))
        if not _is_whole(h.unit_price.minor) or h.unit_price.minor <= 0:
            raise ValueError(
                'unitPrice must be whole minor units greater than zero; received %s for "%s"' % (h.unit_price.minor, h.id)
            )
        t = h.target_basis_points
        if not _is_whole(t) or t < 0 or t > 10000:
            raise ValueError('targetBasisPoints must be a whole number from 0 to 10000; received %s for "%s"' % (t, h.id))
        targets += t
        total += h.units * h.unit_price.minor
    if targets > 10000:
        raise ValueError("targets must sum to at most 10000 basis points, received %d" % targets)
    if total == 0:
        raise ValueError("the portfolio total must be greater than zero")

    minimum = minimum_trade.minor
    trades: List[int] = []
    for h in holdings:
        price = h.unit_price.minor
        n = total * h.target_basis_points
        d = 10000 * price
        target, rem = divmod(n, d)
        if 2 * rem > d or (2 * rem == d and h.units > target):
            target += 1
        trade = target - h.units
        trades.append(0 if abs(trade) * price < minimum else trade)

    def spent() -> int:
        return sum(t * h.unit_price.minor for t, h in zip(trades, holdings))

    cash_after = cash.minor - spent()
    while cash_after < 0:
        pick = -1
        worst = 0
        for i, h in enumerate(holdings):
            if trades[i] <= 0:
                continue
            excess = (h.units + trades[i]) * h.unit_price.minor * 10000 - total * h.target_basis_points
            if pick < 0 or excess > worst:
                pick = i
                worst = excess
        trades[pick] -= 1
        if trades[pick] > 0 and trades[pick] * holdings[pick].unit_price.minor < minimum:
            trades[pick] = 0
        cash_after = cash.minor - spent()

    bought = 0
    sold = 0
    rows: List[RebalanceTrade] = []
    for t, h in zip(trades, holdings):
        price = h.unit_price.minor
        value = t * price
        if value > 0:
            bought += value
        else:
            sold -= value
        units_after = h.units + t
        value_after = units_after * price
        weight, rem = divmod(value_after * 10000, total)
        if 2 * rem >= total:
            weight += 1
        rows.append(
            RebalanceTrade(
                id=h.id,
                units=t,
                value=money(value, currency),
                units_after=units_after,
                value_after=money(value_after, currency),
                weight_after_basis_points=weight,
            )
        )
    return RebalancePlan(
        trades=rows,
        total_value=money(total, currency),
        bought=money(bought, currency),
        sold=money(sold, currency),
        cash_after=money(cash_after, currency),
    )