Coverage for src/jquantstats/_rebalance.py: 100%

1"""Path-dependent simulation of a portfolio rebalanced to sparse target weights. 

2 

3`Portfolio.from_weights` takes target weights that are set on rebalancing dates 

4only, and this module turns them into the units actually held. Between two 

5rebalancings the book holds *units*, not weights, so the realised weights drift 

6with prices. At a rebalancing the units are reset to ``nav * w / price``, where 

7``nav`` is the value of the drifted book on that row, and the trade is charged 

8a cost proportional to its traded notional, paid out of cash. 

9 

10Because the units set at a rebalancing depend on the nav reached there, and the 

11cost paid there feeds into every nav after it, this is a genuine recursion in 

12nav and cash and is simulated one row at a time. The design follows 

13``qis.backtest_rebalanced_portfolio``: target weights are applied to the 

14pre-trade nav, and costs are ``cost_bps / 10_000 * price * |Δunits|`` per asset. 

15""" 

16 

17from __future__ import annotations 

18 

19from typing import NamedTuple 

20 

21import numpy as np 

22 

23from .exceptions import RebalancePriceError 

24 

25 

26class RebalancingPath(NamedTuple): 

27 """Row-by-row state of a simulated rebalanced portfolio. 

28 

29 Attributes: 

30 units: Units held per asset after trading on each row, shape ``(t, n)``. 

31 costs: Trading cost charged per asset on each row, shape ``(t, n)``. 

32 cash: Cash balance after trading on each row, shape ``(t,)``. 

33 nav: Net asset value after trading on each row, shape ``(t,)``. 

34 """ 

35 

36 units: np.ndarray 

37 costs: np.ndarray 

38 cash: np.ndarray 

39 nav: np.ndarray 

40 

41 

42def simulate_rebalancing( 

43 prices: np.ndarray, 

44 weights: np.ndarray, 

45 is_rebalancing: np.ndarray, 

46 aum: float, 

47 assets: list[str], 

48 cost_bps: float = 0.0, 

49 compounding: bool = True, 

50) -> RebalancingPath: 

51 """Simulate a book that trades to target weights on rebalancing rows only. 

52 

53 On a rebalancing row ``t`` the target units are ``base * weights[t] / 

54 prices[t]``, where ``base`` is the pre-trade nav when *compounding* and the 

55 constant *aum* otherwise. On every other row the previous units are held. 

56 The trade cost ``cost_bps / 10_000 * prices[t] * |Δunits|`` is paid out of 

57 cash, so it lowers the nav every later rebalancing is sized off. 

58 

59 A price that is missing between two rebalancings drops that leg from the 

60 nav for the row, as the units are still held. On a rebalancing row, 

61 however, every asset that is held or targeted must have a finite, positive 

62 price: the trade could not be valued otherwise. 

63 

64 Args: 

65 prices: Price per asset, shape ``(t, n)``. 

66 weights: Target weights, shape ``(t, n)``. Rows where *is_rebalancing* 

67 is False are ignored. 

68 is_rebalancing: Boolean flag per row, shape ``(t,)``. 

69 aum: Starting capital, held as cash until the first rebalancing. 

70 assets: Asset names in column order, used in error messages. 

71 cost_bps: One-way cost in basis points of traded notional. 

72 compounding: Size trades off the current nav when True, off the 

73 constant *aum* when False. 

74 

75 Returns: 

76 RebalancingPath: Units, per-asset costs, cash and nav for every row. 

77 

78 Raises: 

79 RebalancePriceError: If an asset that is held or targeted has no 

80 finite, positive price on a rebalancing row. 

81 """ 

82 n_rows, n_assets = prices.shape 

83 rate = cost_bps / 10_000.0 

84 units = np.zeros((n_rows, n_assets)) 

85 costs = np.zeros((n_rows, n_assets)) 

86 cash = np.zeros(n_rows) 

87 nav = np.zeros(n_rows) 

88 

89 held = np.zeros(n_assets) 

90 balance = float(aum) 

91 for t in range(n_rows): 

92 price = prices[t] 

93 if is_rebalancing[t]: 

94 involved = (weights[t] != 0.0) | (held != 0.0) 

95 unpriced = involved & ~(np.isfinite(price) & (price > 0.0)) 

96 if unpriced.any(): 

97 raise RebalancePriceError(t, [assets[i] for i in np.flatnonzero(unpriced)]) 

98 pre_trade_nav = balance + np.nansum(held * price) 

99 base = pre_trade_nav if compounding else aum 

100 target = np.divide(base * weights[t], price, out=np.zeros(n_assets), where=weights[t] != 0.0) 

101 traded = np.multiply(target - held, price, out=np.zeros(n_assets), where=involved) 

102 costs[t] = rate * np.abs(traded) 

103 balance -= traded.sum() + costs[t].sum() 

104 held = target 

105 units[t] = held 

106 cash[t] = balance 

107 nav[t] = balance + np.nansum(held * price) 

108 

109 return RebalancingPath(units=units, costs=costs, cash=cash, nav=nav)