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),
)