Coverage for src / monkeys / portfolio.py: 100%

1"""Portfolio simulation module for random allocation strategies. 

2 

3This module provides tools for simulating random portfolio allocation strategies 

4("monkeys") that generate random weights and track portfolio performance. 

5""" 

6 

7from __future__ import annotations 

8 

9from typing import TYPE_CHECKING 

10 

11import numpy as np 

12 

13if TYPE_CHECKING: 

14 from numpy.typing import NDArray 

15 

16 

17def generate_weight_history( 

18 n_assets: int, 

19 n_periods: int, 

20 seed: int | None = None, 

21 probabilities: NDArray[np.float64] | None = None, 

22) -> NDArray[np.float64]: 

23 """Generate a history of random portfolio weights over multiple periods. 

24 

25 Simulates a monkey rebalancing the portfolio each period with new random 

26 weights. 

27 

28 Args: 

29 n_assets: Number of assets in the portfolio. Must be positive. 

30 n_periods: Number of time periods to simulate. Must be positive. 

31 seed: Random seed for reproducibility. 

32 probabilities: Selection probabilities for each asset. 

33 

34 Returns: 

35 2D array of shape (n_periods, n_assets) where each row sums to 1.0. 

36 

37 Raises: 

38 ValueError: If n_assets < 1 or n_periods < 1. 

39 

40 Examples: 

41 >>> history = generate_weight_history(3, 10, seed=42) 

42 >>> history.shape 

43 (10, 3) 

44 >>> np.allclose(history.sum(axis=1), 1.0) 

45 True 

46 """ 

47 if n_periods < 1: 

48 msg = f"n_periods must be positive, got {n_periods}" 

49 raise ValueError(msg) 

50 

51 rng = np.random.default_rng(seed) 

52 

53 # Generate all random samples at once for efficiency 

54 if probabilities is None: 

55 probabilities = np.ones(n_assets) / n_assets 

56 elif not np.isclose(probabilities.sum(), 1.0): 

57 msg = f"probabilities must sum to 1.0, got {probabilities.sum()}" 

58 raise ValueError(msg) 

59 

60 uniform_samples = rng.uniform(0, 1, (n_periods, n_assets)) 

61 raw_weights = probabilities * uniform_samples 

62 

63 # Normalize each row 

64 row_sums = raw_weights.sum(axis=1, keepdims=True) 

65 weights: NDArray[np.float64] = raw_weights / row_sums 

66 

67 return weights 

68 

69 

70def simulate_portfolio_returns( 

71 returns: NDArray[np.float64], 

72 seed: int | None = None, 

73 probabilities: NDArray[np.float64] | None = None, 

74 rebalance_frequency: int = 1, 

75) -> NDArray[np.float64]: 

76 """Simulate portfolio returns using random weight allocation. 

77 

78 Simulates a monkey managing a portfolio over time, rebalancing at the 

79 specified frequency with random weights. 

80 

81 Args: 

82 returns: 2D array of asset returns with shape (n_periods, n_assets). 

83 seed: Random seed for reproducibility. 

84 probabilities: Selection probabilities for each asset. 

85 rebalance_frequency: Number of periods between rebalancing. Default is 1 

86 (rebalance every period). 

87 

88 Returns: 

89 1D array of portfolio returns for each period. 

90 

91 Raises: 

92 ValueError: If returns is not 2D or rebalance_frequency < 1. 

93 

94 Examples: 

95 >>> asset_returns = np.array([ 

96 ... [0.01, 0.02, -0.01], 

97 ... [0.02, -0.01, 0.03], 

98 ... [-0.01, 0.01, 0.02], 

99 ... ]) 

100 >>> portfolio_returns = simulate_portfolio_returns(asset_returns, seed=42) 

101 >>> len(portfolio_returns) 

102 3 

103 """ 

104 if returns.ndim != 2: 

105 msg = f"returns must be 2D array, got {returns.ndim}D" 

106 raise ValueError(msg) 

107 

108 if rebalance_frequency < 1: 

109 msg = f"rebalance_frequency must be positive, got {rebalance_frequency}" 

110 raise ValueError(msg) 

111 

112 n_periods, n_assets = returns.shape 

113 rng = np.random.default_rng(seed) 

114 

115 if probabilities is None: 

116 probabilities = np.ones(n_assets) / n_assets 

117 

118 portfolio_returns = np.zeros(n_periods) 

119 current_weights = None 

120 

121 for t in range(n_periods): 

122 # Rebalance if needed 

123 if t % rebalance_frequency == 0 or current_weights is None: 

124 uniform_samples = rng.uniform(0, 1, n_assets) 

125 raw_weights = probabilities * uniform_samples 

126 current_weights = raw_weights / raw_weights.sum() 

127 

128 portfolio_returns[t] = np.dot(current_weights, returns[t]) 

129 

130 return portfolio_returns