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

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()[root.value] 

36 root.portfolio[asset] = 1.0 

37 return root 

38 

39 # drill down on the left 

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

41 # drill down on the right 

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

43 

44 # combine left and right into a new cluster 

45 return _parity(root, cov=cov) 

46 

47 

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

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

50 

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

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

53 equally to the total portfolio risk. 

54 

55 Args: 

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

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

58 

59 Returns: 

60 Cluster: The parent cluster with portfolio weights assigned 

61 """ 

62 # Calculate variances of left and right sub-portfolios 

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

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

65 

66 # Calculate weights inversely proportional to risk 

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

68 alpha_left = v_right / (v_left + v_right) 

69 alpha_right = v_left / (v_left + v_right) 

70 

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

72 assets = { 

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

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

75 } 

76 

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

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

79 cluster.portfolio[asset] = weight 

80 

81 return cluster 

82 

83 

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

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

86 

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

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

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

90 

91 Args: 

92 dendrogram: A dendrogram object containing the hierarchical clustering tree 

93 and the list of assets 

94 

95 Yields: 

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

97 at that level 

98 """ 

99 root = dendrogram.root 

100 assets = dendrogram.assets 

101 

102 # Initial weight to distribute 

103 w = 1 

104 

105 # Process each level of the tree 

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

107 for node in level: 

108 # Distribute weight equally among all leaves in this node 

109 for leaf in node.leaves: 

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

111 

112 # Reduce weight for the next level 

113 w *= 0.5 

114 

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

116 yield n, deepcopy(root.portfolio)