Coverage for src/pyhrp/algos.py: 100%
84 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-09-15 05:20 +0000
1"""Portfolio optimization algorithms for hierarchical risk parity.
3This module implements various portfolio optimization algorithms:
4- risk_parity: The main hierarchical risk parity algorithm
5- schur_risk_parity: Schur Complementary Allocation (Cotton, arXiv:2411.05807)
6- one_over_n: A simple equal-weight allocation strategy
8Allocator contract
9------------------
10All three allocators take the same inputs — a ``Cluster`` tree (``root``) plus
11the asset names — and none of them changes the *shape* of that tree: no node is
12added, removed or re-parented. They differ in where the weights land.
14``risk_parity`` and ``schur_risk_parity`` share the recursive ``_allocate_with``
15scaffolding and write **into the tree they are given**: every node's
16``portfolio`` is replaced rather than accumulated into, and the same root object
17is returned. Replacing makes them idempotent — re-running on an already weighted
18tree, with a different covariance matrix or gamma, gives the same answer as
19running on a fresh one — but the caller's tree *is* modified. Note in particular
20that ``Dendrogram`` is a frozen dataclass while the ``Cluster`` it holds is not:
21passing ``dendrogram.root`` to either allocator rewrites the portfolios inside
22that dendrogram. Pass ``copy.deepcopy(dendrogram.root)`` to keep the original
23tree unweighted.
25``one_over_n`` differs on both counts. It accumulates into a local buffer, so it
26leaves the input tree untouched entirely, and its *output* is a generator
27yielding the equal-weight portfolio one tree level at a time (see its
28docstring), because its purpose is to expose the allocation as it deepens rather
29than a single final result.
30"""
32from __future__ import annotations
34from collections.abc import Callable, Generator
35from copy import deepcopy
37import numpy as np
38import polars as pl
40from .cluster import Cluster, Portfolio
42__all__ = ["one_over_n", "risk_parity", "schur_risk_parity"]
45def risk_parity(root: Cluster, cov: pl.DataFrame) -> Cluster:
46 """Compute hierarchical risk parity weights for a cluster tree.
48 This is the main algorithm for hierarchical risk parity. It recursively
49 traverses the cluster tree and assigns weights to each node based on
50 the risk parity principle.
52 Note:
53 The tree is modified in place: the portfolio of every node is rebuilt
54 from scratch and ``root`` itself is returned, so the function is
55 idempotent and a tree can be reused with a different covariance matrix.
56 Deep-copy the tree first if you need to keep it unweighted.
58 Args:
59 root (Cluster): The root node of the cluster tree
60 cov (pl.DataFrame): Covariance matrix of asset returns
62 Returns:
63 Cluster: The same root node, with portfolio weights assigned
65 Examples:
66 >>> import polars as pl
67 >>> from pyhrp.cluster import Cluster
68 >>> from pyhrp.algos import risk_parity
69 >>> cov = pl.DataFrame({"A": [4.0, 0.0], "B": [0.0, 1.0]})
70 >>> root = Cluster(2, left=Cluster(0), right=Cluster(1))
71 >>> cluster = risk_parity(root=root, cov=cov)
72 >>> round(cluster.portfolio["B"], 1)
73 0.8
74 """
76 def node_variances(left: Cluster, right: Cluster, cov_np: np.ndarray, index: dict[str, int]) -> tuple[float, float]:
77 """Plain block variance of each child sub-portfolio."""
78 return (
79 _block_variance(left.portfolio, cov_np, index),
80 _block_variance(right.portfolio, cov_np, index),
81 )
83 return _allocate_with(root, cov, node_variances)
86def schur_risk_parity(root: Cluster, cov: pl.DataFrame, gamma: float = 0.5) -> Cluster:
87 """Compute Schur Complementary Allocation weights for a cluster tree.
89 An extension of HRP introduced by Peter Cotton (arXiv:2411.05807) that augments
90 sub-covariance matrices with off-diagonal block information via Schur complements.
91 At gamma=0 this recovers standard HRP; at gamma=1 it recovers the minimum-variance
92 portfolio through the same recursive structure.
94 Note:
95 The tree is modified in place: the portfolio of every node is rebuilt
96 from scratch and ``root`` itself is returned, so the function is
97 idempotent and a tree can be reused with a different covariance matrix
98 or gamma. Deep-copy the tree first if you need to keep it unweighted.
100 Args:
101 root (Cluster): The root node of the cluster tree
102 cov (pl.DataFrame): Covariance matrix of asset returns
103 gamma (float): Interpolation parameter in [0, 1]. 0 = HRP, 1 = minimum variance.
105 Returns:
106 Cluster: The same root node, with portfolio weights assigned
108 Raises:
109 ValueError: If gamma is outside the interval [0, 1].
111 Examples:
112 >>> import polars as pl
113 >>> from pyhrp.cluster import Cluster
114 >>> from pyhrp.algos import schur_risk_parity
115 >>> cov = pl.DataFrame({"A": [4.0, 0.0], "B": [0.0, 1.0]})
116 >>> root = Cluster(2, left=Cluster(0), right=Cluster(1))
117 >>> cluster = schur_risk_parity(root=root, cov=cov, gamma=0.5)
118 >>> round(cluster.portfolio["B"], 1)
119 0.8
120 """
121 if not 0.0 <= gamma <= 1.0:
122 msg = f"gamma must be in [0, 1], got {gamma}"
123 raise ValueError(msg)
125 def node_variances(left: Cluster, right: Cluster, cov_np: np.ndarray, index: dict[str, int]) -> tuple[float, float]:
126 """Schur-augmented block variance of each child, conditioned on the other."""
127 li = [index[a] for a in left.portfolio.assets]
128 ri = [index[a] for a in right.portfolio.assets]
130 a_mat = cov_np[np.ix_(li, li)]
131 b_mat = cov_np[np.ix_(li, ri)]
132 d_mat = cov_np[np.ix_(ri, ri)]
134 w_left = np.array([left.portfolio[a] for a in left.portfolio.assets])
135 w_right = np.array([right.portfolio[a] for a in right.portfolio.assets])
137 # Schur-augmented blocks: condition each group on the other
138 a_aug = a_mat - gamma * (b_mat @ _solve(d_mat, b_mat.T))
139 d_aug = d_mat - gamma * (b_mat.T @ _solve(a_mat, b_mat))
141 v_left = float(w_left @ a_aug @ w_left)
142 v_right = float(w_right @ d_aug @ w_right)
143 return v_left, v_right
145 return _allocate_with(root, cov, node_variances)
148# Given a node's two children (plus the precomputed covariance array and the
149# column->row index), return the (v_left, v_right) risk pair used to split it.
150NodeVariances = Callable[[Cluster, Cluster, np.ndarray, dict[str, int]], tuple[float, float]]
153def _allocate_with(root: Cluster, cov: pl.DataFrame, node_variances: NodeVariances) -> Cluster:
154 """Shared scaffolding for the recursive risk-based allocators.
156 Builds the numpy covariance array and column index once, then walks the tree
157 bottom-up, splitting each node's weight between its children inversely to the
158 ``(v_left, v_right)`` pair supplied by ``node_variances``. The only thing that
159 distinguishes ``risk_parity`` from ``schur_risk_parity`` is that per-node
160 variance rule; everything else — the ``cov``/``index`` setup, the combine
161 wrapper, and the rebuild-from-scratch traversal — lives here.
163 Args:
164 root (Cluster): The root node of the cluster tree.
165 cov (pl.DataFrame): Covariance matrix of asset returns.
166 node_variances (NodeVariances): Per-node rule mapping a node's left/right
167 children (and the precomputed covariance array and column index) to
168 the ``(v_left, v_right)`` risk pair used to split that node.
170 Returns:
171 Cluster: The root node with portfolio weights assigned.
172 """
173 cov_np = cov.to_numpy()
174 index = {name: i for i, name in enumerate(cov.columns)}
176 def combine(cluster: Cluster) -> Cluster:
177 """Combine the child portfolios of a cluster via an inverse-variance split."""
178 left, right = cluster._child_clusters()
179 v_left, v_right = node_variances(left, right, cov_np, index)
180 return _split(cluster, v_left, v_right)
182 return _allocate(root, cov.columns, combine)
185def _allocate(root: Cluster, assets: list[str], combine: Callable[[Cluster], Cluster]) -> Cluster:
186 """Traverse the tree bottom-up, assigning leaf portfolios and combining children.
188 Every node's portfolio is replaced, never accumulated into, which keeps
189 repeated allocations on the same tree idempotent.
191 Args:
192 root (Cluster): The (sub)tree to allocate weights for
193 assets (list[str]): Asset names; a leaf's value indexes into this list
194 combine (Callable[[Cluster], Cluster]): Combines the two child portfolios
195 of a node into the node's own portfolio
197 Returns:
198 Cluster: The input node with portfolio weights assigned
199 """
200 # Iterative post-order rather than recursion: a chain-degenerate tree is as deep
201 # as the universe is wide, and recursing here capped the number of assets that
202 # could be allocated. Pass one collects nodes parent-before-child (validating
203 # each non-leaf as it goes, so a malformed tree still raises before any weight is
204 # written); reversing that order guarantees both children hold a portfolio before
205 # their parent combines them.
206 order: list[Cluster] = []
207 stack: list[Cluster] = [root]
208 while stack:
209 node = stack.pop()
210 order.append(node)
211 if not node.is_leaf:
212 left, right = node._child_clusters()
213 stack.append(left)
214 stack.append(right)
216 for node in reversed(order):
217 if node.is_leaf:
218 node.portfolio = Portfolio()
219 node.portfolio[assets[int(node.value)]] = 1.0
220 else:
221 # combine() replaces node.portfolio in place; the return value is the
222 # same node, which is why the recursive form's reassignment of
223 # root.left/root.right was always a no-op.
224 combine(node)
226 return root
229def _block_variance(portfolio: Portfolio, cov_np: np.ndarray, index: dict[str, int]) -> float:
230 """Compute the variance of a portfolio against a precomputed covariance array."""
231 assets = portfolio.assets
232 idx = [index[a] for a in assets]
233 w = np.array([portfolio[a] for a in assets])
234 return float(w @ cov_np[np.ix_(idx, idx)] @ w)
237def _split(cluster: Cluster, v_left: float, v_right: float) -> Cluster:
238 """Distribute weight between the two children inversely proportional to risk.
240 The split satisfies v_left * alpha_left == v_right * alpha_right with
241 alpha_left + alpha_right == 1. If both variances are zero (e.g. riskless
242 sub-portfolios), the weight is split equally.
244 Args:
245 cluster (Cluster): The parent cluster with left and right children
246 v_left (float): Variance of the left sub-portfolio
247 v_right (float): Variance of the right sub-portfolio
249 Returns:
250 Cluster: The parent cluster with portfolio weights assigned
251 """
252 left, right = cluster._child_clusters()
253 total = v_left + v_right
254 alpha_left = v_right / total if total > 0 else 0.5
255 alpha_right = 1.0 - alpha_left
257 cluster.portfolio = Portfolio()
258 for asset, weight in left.portfolio.weights.items():
259 cluster.portfolio[asset] = alpha_left * weight
260 for asset, weight in right.portfolio.weights.items():
261 cluster.portfolio[asset] = alpha_right * weight
263 return cluster
266def _solve(m: np.ndarray, b: np.ndarray) -> np.ndarray:
267 """Solve m @ x = b, falling back to least squares for singular matrices.
269 Covariance blocks of collinear assets are singular; the minimum-norm
270 least-squares solution keeps the Schur augmentation well-defined there.
271 """
272 try:
273 return np.linalg.solve(m, b)
274 except np.linalg.LinAlgError:
275 return np.asarray(np.linalg.lstsq(m, b, rcond=None)[0])
278def one_over_n(root: Cluster, assets: list[str]) -> Generator[tuple[int, Portfolio]]:
279 """Generate 1/N (equal-weight) portfolios one tree level at a time.
281 This implements a hierarchical 1/N strategy where weights are distributed
282 equally among the leaves of each cluster, and the weight budget halves at
283 each successive level of the tree.
285 Unlike :func:`risk_parity` and :func:`schur_risk_parity` — which rebuild a
286 single final allocation and return the root ``Cluster`` — this allocator is
287 intentionally a **generator**: its purpose is to expose the equal-weight
288 allocation as the tree deepens, yielding one portfolio per level. It shares
289 the sibling input contract (a ``Cluster`` tree plus the asset names) and, like
290 them, does not mutate the tree: weights accumulate in a local buffer, so a
291 leaf that terminates at a shallow level keeps its weight in the deeper levels
292 (each yielded portfolio is therefore a complete allocation over all assets),
293 and re-running on the same tree yields an identical sequence.
295 Args:
296 root (Cluster): The root node of the cluster tree.
297 assets (list[str]): Asset names; a leaf's value indexes into this list.
299 Yields:
300 tuple[int, Portfolio]: The level number and the (cumulative) equal-weight
301 portfolio at that level.
303 Examples:
304 >>> import polars as pl
305 >>> from pyhrp.hrp import build_tree
306 >>> from pyhrp.algos import one_over_n
307 >>> cor = pl.DataFrame({"A": [1.0, 0.3], "B": [0.3, 1.0]})
308 >>> dg = build_tree(cor, method="ward")
309 >>> levels = list(one_over_n(dg.root, dg.assets))
310 >>> len(levels) > 0
311 True
312 """
313 # Accumulate into a local buffer so the input tree is never mutated.
314 portfolio = Portfolio()
316 # Initial weight to distribute
317 w: float = 1.0
319 # Process each level of the tree
320 for n, level in enumerate(root.levels):
321 for node in level:
322 # Distribute weight equally among all leaves in this node
323 for leaf in node.leaves:
324 portfolio[assets[leaf.value]] = w / node.leaf_count
326 # Reduce weight for the next level
327 w *= 0.5
329 # Yield the current level number and a deep copy of the portfolio
330 yield n, deepcopy(portfolio)