Coverage for src / pyhrp / algos.py: 90%

1"""Portfolio optimization algorithms for hierarchical risk parity. 

2 

3This module implements various portfolio optimization algorithms: 

4- risk_parity: The main hierarchical risk parity algorithm 

5- one_over_n: A simple equal-weight allocation strategy 

6""" 

7 

8from __future__ import annotations 

9 

10from collections.abc import Generator 

11from copy import deepcopy 

12from typing import Any 

13 

14import pandas as pd 

15 

16from .cluster import Cluster, Portfolio 

17 

18 

19def risk_parity(root: Cluster, cov: pd.DataFrame) -> Cluster: 

20 """Compute hierarchical risk parity weights for a cluster tree. 

21 

22 This is the main algorithm for hierarchical risk parity. It recursively 

23 traverses the cluster tree and assigns weights to each node based on 

24 the risk parity principle. 

25 

26 Args: 

27 root (Cluster): The root node of the cluster tree 

28 cov (pd.DataFrame): Covariance matrix of asset returns 

29 

30 Returns: 

31 Cluster: The root node with portfolio weights assigned 

32 """ 

33 if root.is_leaf: 

34 # a node is a leaf if has no further relatives downstream. 

35 asset = cov.keys().to_list()[int(root.value)] 

36 root.portfolio[asset] = 1.0 

37 return root 

38 

39 # drill down on the left 

40 if not isinstance(root.left, Cluster): 

41 raise TypeError("Expected left child to be a Cluster") # noqa: TRY003 

42 if not isinstance(root.right, Cluster): 

43 raise TypeError("Expected right child to be a Cluster") # noqa: TRY003 

44 root.left = risk_parity(root.left, cov) 

45 # drill down on the right 

46 root.right = risk_parity(root.right, cov) 

47 

48 # combine left and right into a new cluster 

49 return _parity(root, cov=cov) 

50 

51 

52def _parity(cluster: Cluster, cov: pd.DataFrame) -> Cluster: 

53 """Compute risk parity weights for a parent cluster from its children. 

54 

55 This function implements the core risk parity principle: allocating weights 

56 inversely proportional to risk, so that each sub-portfolio contributes 

57 equally to the total portfolio risk. 

58 

59 Args: 

60 cluster (Cluster): The parent cluster with left and right children 

61 cov (pd.DataFrame): Covariance matrix of asset returns 

62 

63 Returns: 

64 Cluster: The parent cluster with portfolio weights assigned 

65 """ 

66 # Calculate variances of left and right sub-portfolios 

67 if not isinstance(cluster.left, Cluster): 

68 raise TypeError("Expected left child to be a Cluster") # noqa: TRY003 

69 if not isinstance(cluster.right, Cluster): 

70 raise TypeError("Expected right child to be a Cluster") # noqa: TRY003 

71 v_left = cluster.left.portfolio.variance(cov) 

72 v_right = cluster.right.portfolio.variance(cov) 

73 

74 # Calculate weights inversely proportional to risk 

75 # such that v_left * alpha_left == v_right * alpha_right and alpha_left + alpha_right = 1 

76 alpha_left = v_right / (v_left + v_right) 

77 alpha_right = v_left / (v_left + v_right) 

78 

79 # Combine assets from left and right clusters with their adjusted weights 

80 assets = { 

81 **(alpha_left * cluster.left.portfolio.weights).to_dict(), 

82 **(alpha_right * cluster.right.portfolio.weights).to_dict(), 

83 } 

84 

85 # Assign the combined weights to the parent cluster's portfolio 

86 for asset, weight in assets.items(): 

87 cluster.portfolio[asset] = weight 

88 

89 return cluster 

90 

91 

92def one_over_n(dendrogram: Any) -> Generator[tuple[int, Portfolio]]: 

93 """Generate portfolios using the 1/N (equal weight) strategy at each tree level. 

94 

95 This function implements a hierarchical 1/N strategy where weights are 

96 distributed equally among assets within each cluster at each level of the tree. 

97 The weight assigned to each cluster decreases by half at each level. 

98 

99 Args: 

100 dendrogram: A dendrogram object containing the hierarchical clustering tree 

101 and the list of assets 

102 

103 Yields: 

104 tuple[int, Portfolio]: A tuple containing the level number and the portfolio 

105 at that level 

106 """ 

107 root = dendrogram.root 

108 assets = dendrogram.assets 

109 

110 # Initial weight to distribute 

111 w: float = 1.0 

112 

113 # Process each level of the tree 

114 for n, level in enumerate(root.levels): 

115 for node in level: 

116 # Distribute weight equally among all leaves in this node 

117 for leaf in node.leaves: 

118 root.portfolio[assets[leaf.value]] = w / node.leaf_count 

119 

120 # Reduce weight for the next level 

121 w *= 0.5 

122 

123 # Yield the current level number and a deep copy of the portfolio 

124 yield n, deepcopy(root.portfolio)