Skip to content

API Reference

The public API of pyhrp. Import the main entry points directly from the top-level package:

from pyhrp.hrp import hrp, build_tree, Dendrogram
from pyhrp.algos import risk_parity, one_over_n
from pyhrp.cluster import Cluster, Portfolio

Hierarchical Risk Parity (HRP) allocation entry points.

This module exposes the top-level allocation functions and re-exports the supporting building blocks so the public pyhrp.hrp API is unchanged: - hrp: Compute HRP portfolio weights from prices - schur_hrp: Compute Schur Complementary Allocation weights from prices - build_tree: Build a hierarchical cluster tree (see :mod:pyhrp.dendrogram) - compute_returns: Simple returns from prices (see :mod:pyhrp.covariance) - compute_cov / compute_corr: Second-moment estimators (see :mod:pyhrp.covariance) - Dendrogram: Clustering result container (see :mod:pyhrp.dendrogram)

Dendrogram dataclass

Container for hierarchical clustering dendrogram data and visualization.

This class stores the results of hierarchical clustering and provides methods for accessing and visualizing the dendrogram structure.

Attributes:

Name Type Description
root Cluster

The root node of the hierarchical clustering tree
assets list[str]

Names of assets included in the clustering
linkage ndarray | None

Linkage matrix in scipy format for plotting
distance DataFrame | None

Distance matrix used for clustering
method str | None

Linkage method used for clustering
Source code in src/pyhrp/dendrogram.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
@dataclass(frozen=True)
class Dendrogram:
    """Container for hierarchical clustering dendrogram data and visualization.

    This class stores the results of hierarchical clustering and provides methods
    for accessing and visualizing the dendrogram structure.

    Attributes:
        root (Cluster): The root node of the hierarchical clustering tree
        assets (list[str]): Names of assets included in the clustering
        linkage (np.ndarray | None): Linkage matrix in scipy format for plotting
        distance (pl.DataFrame | None): Distance matrix used for clustering
        method (str | None): Linkage method used for clustering
    """

    root: Cluster
    assets: list[str]
    distance: pl.DataFrame | None = None
    linkage: np.ndarray | None = None
    method: str | None = None

    def __post_init__(self) -> None:
        """Validate dataclass fields after initialization.

        Ensures that the optional distance matrix, when provided, is a polars
        DataFrame with columns aligned to the asset list, and verifies that the
        number of leaves in the cluster tree matches the number of assets.
        """
        if self.distance is not None:
            if not isinstance(self.distance, pl.DataFrame):
                msg = "distance must be a polars DataFrame."
                raise TypeError(msg)

            if self.distance.columns != list(self.assets):
                msg = "Distance matrix index/columns must align with assets."
                raise ValueError(msg)

        if len(self.root.leaves) != len(self.assets):
            msg = "Number of leaves does not match number of assets."
            raise ValueError(msg)

    def plot(self, **kwargs: object) -> go.Figure:
        """Build and return a plotly dendrogram figure.

        Delegates to :func:`pyhrp.plot.plot_dendrogram`; the plotly dependency
        is imported lazily so importing the allocation core stays plotly-free.
        """
        from .plot import plot_dendrogram

        return plot_dendrogram(self, **kwargs)

    def one_over_n(self) -> Generator[tuple[int, Portfolio]]:
        """Yield the hierarchical 1/N portfolios level by level for this tree.

        Container-level convenience wrapper around :func:`pyhrp.algos.one_over_n`.

        Yields:
            tuple[int, Portfolio]: The level number and the equal-weight portfolio
            at that level.
        """
        yield from _one_over_n(self.root, self.assets)

    @property
    def ids(self) -> list[int]:
        """Node values in the order left -> right as they appear in the dendrogram."""
        return [node.value for node in self.root.leaves]

    @property
    def names(self) -> list[str]:
        """The asset names as induced by the order of ids."""
        return [self.assets[i] for i in self.ids]

ids property

Node values in the order left -> right as they appear in the dendrogram.

names property

The asset names as induced by the order of ids.

__post_init__()

Validate dataclass fields after initialization.

Ensures that the optional distance matrix, when provided, is a polars DataFrame with columns aligned to the asset list, and verifies that the number of leaves in the cluster tree matches the number of assets.

Source code in src/pyhrp/dendrogram.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __post_init__(self) -> None:
    """Validate dataclass fields after initialization.

    Ensures that the optional distance matrix, when provided, is a polars
    DataFrame with columns aligned to the asset list, and verifies that the
    number of leaves in the cluster tree matches the number of assets.
    """
    if self.distance is not None:
        if not isinstance(self.distance, pl.DataFrame):
            msg = "distance must be a polars DataFrame."
            raise TypeError(msg)

        if self.distance.columns != list(self.assets):
            msg = "Distance matrix index/columns must align with assets."
            raise ValueError(msg)

    if len(self.root.leaves) != len(self.assets):
        msg = "Number of leaves does not match number of assets."
        raise ValueError(msg)

one_over_n()

Yield the hierarchical 1/N portfolios level by level for this tree.

Container-level convenience wrapper around :func:pyhrp.algos.one_over_n.

Yields:

Type Description
Generator[tuple[int, Portfolio]]

tuple[int, Portfolio]: The level number and the equal-weight portfolio
Generator[tuple[int, Portfolio]]

at that level.
Source code in src/pyhrp/dendrogram.py 
80
81
82
83
84
85
86
87
88
89
def one_over_n(self) -> Generator[tuple[int, Portfolio]]:
    """Yield the hierarchical 1/N portfolios level by level for this tree.

    Container-level convenience wrapper around :func:`pyhrp.algos.one_over_n`.

    Yields:
        tuple[int, Portfolio]: The level number and the equal-weight portfolio
        at that level.
    """
    yield from _one_over_n(self.root, self.assets)

plot(**kwargs)

Build and return a plotly dendrogram figure.

Delegates to :func:pyhrp.plot.plot_dendrogram; the plotly dependency is imported lazily so importing the allocation core stays plotly-free.

Source code in src/pyhrp/dendrogram.py 
70
71
72
73
74
75
76
77
78
def plot(self, **kwargs: object) -> go.Figure:
    """Build and return a plotly dendrogram figure.

    Delegates to :func:`pyhrp.plot.plot_dendrogram`; the plotly dependency
    is imported lazily so importing the allocation core stays plotly-free.
    """
    from .plot import plot_dendrogram

    return plot_dendrogram(self, **kwargs)

build_tree(cor, method='ward', bisection=False)

Build hierarchical cluster tree from correlation matrix.

This function converts a correlation matrix to a distance matrix, performs hierarchical clustering, and returns a Dendrogram object containing the resulting tree structure.

Parameters:

Name Type Description Default
cor DataFrame

Correlation matrix of asset returns (columns are assets)

 required
method Literal['single', 'complete', 'average', 'ward']

Linkage method for hierarchical clustering - "single": minimum distance between points (nearest neighbor) - "complete": maximum distance between points (furthest neighbor) - "average": average distance between all points - "ward": Ward variance minimization

 'ward'
bisection bool

Whether to use bisection method for tree construction

 False

Returns:

Name Type Description
Dendrogram Dendrogram

Object containing the hierarchical clustering tree, with: - root: Root cluster node - linkage: Linkage matrix for plotting - assets: List of assets - method: Clustering method used - distance: Distance matrix

Examples:

>>> import polars as pl
>>> from pyhrp.dendrogram import build_tree
>>> cor = pl.DataFrame({"A": [1.0, 0.5], "B": [0.5, 1.0]})
>>> dg = build_tree(cor, method="ward")
>>> dg.root.leaf_count
2
Source code in src/pyhrp/dendrogram.py 
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
def build_tree(
    cor: pl.DataFrame, method: Literal["single", "complete", "average", "ward"] = "ward", bisection: bool = False
) -> Dendrogram:
    """Build hierarchical cluster tree from correlation matrix.

    This function converts a correlation matrix to a distance matrix, performs
    hierarchical clustering, and returns a Dendrogram object containing the
    resulting tree structure.

    Args:
        cor (pl.DataFrame): Correlation matrix of asset returns (columns are assets)
        method (Literal["single", "complete", "average", "ward"]): Linkage method for hierarchical clustering
            - "single": minimum distance between points (nearest neighbor)
            - "complete": maximum distance between points (furthest neighbor)
            - "average": average distance between all points
            - "ward": Ward variance minimization
        bisection (bool): Whether to use bisection method for tree construction

    Returns:
        Dendrogram: Object containing the hierarchical clustering tree, with:
            - root: Root cluster node
            - linkage: Linkage matrix for plotting
            - assets: List of assets
            - method: Clustering method used
            - distance: Distance matrix

    Examples:
        >>> import polars as pl
        >>> from pyhrp.dendrogram import build_tree
        >>> cor = pl.DataFrame({"A": [1.0, 0.5], "B": [0.5, 1.0]})
        >>> dg = build_tree(cor, method="ward")
        >>> dg.root.leaf_count
        2
    """
    _validate_correlation_matrix(cor)
    dist = _compute_distance_matrix(cor)
    links = sch.linkage(ssd.squareform(dist.to_numpy(), checks=False), method=method)

    root = _to_cluster(sch.to_tree(links, rd=False))

    # Apply bisection if requested
    if bisection:
        # Rebuild tree using bisection
        leaf_ids: list[int] = [int(node.value) for node in root.leaves]
        root, _ = _bisect_tree(ids=leaf_ids, next_id=max(leaf_ids))
        links = np.array(_get_linkage(root, dist.to_numpy()))

    return Dendrogram(root=root, linkage=links, method=method, distance=dist, assets=cor.columns)

check_finite_matrix(matrix, name='matrix')

Raise if a matrix contains NaN or null entries.

A covariance matrix with non-finite entries makes every downstream weight meaningless; this guard fails loudly instead of propagating garbage.

Parameters:

Name Type Description Default
matrix DataFrame

Square matrix (columns are assets)

 required
name str

Human-readable name used in the error message

 'matrix'

Returns:

Type Description
DataFrame

pl.DataFrame: The input matrix, unchanged

Raises:

Type Description
ValueError

If any entry of matrix is NaN or null.
Source code in src/pyhrp/covariance.py 
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def check_finite_matrix(matrix: pl.DataFrame, name: str = "matrix") -> pl.DataFrame:
    """Raise if a matrix contains NaN or null entries.

    A covariance matrix with non-finite entries makes every downstream weight
    meaningless; this guard fails loudly instead of propagating garbage.

    Args:
        matrix (pl.DataFrame): Square matrix (columns are assets)
        name (str): Human-readable name used in the error message

    Returns:
        pl.DataFrame: The input matrix, unchanged

    Raises:
        ValueError: If any entry of ``matrix`` is NaN or null.
    """
    bad = {
        col: int(count) + int(matrix[col].is_nan().sum())
        for col, count in matrix.null_count().row(0, named=True).items()
        if count > 0 or bool(matrix[col].is_nan().any())
    }
    if bad:
        detail = ", ".join(f"{c} ({n})" for c, n in sorted(bad.items()))
        msg = f"{name} contains NaN/null entries in column(s): {detail}"
        raise ValueError(msg)
    return matrix

compute_corr(df)

Compute correlation matrix from a DataFrame of returns.

Source code in src/pyhrp/covariance.py 
116
117
118
119
120
def compute_corr(df: pl.DataFrame) -> pl.DataFrame:
    """Compute correlation matrix from a DataFrame of returns."""
    cols = df.columns
    corr = np.atleast_2d(np.corrcoef(df.to_numpy().T))
    return pl.DataFrame(dict(zip(cols, corr, strict=True)))

compute_cov(df)

Compute covariance matrix from a DataFrame of returns.

Source code in src/pyhrp/covariance.py 
109
110
111
112
113
def compute_cov(df: pl.DataFrame) -> pl.DataFrame:
    """Compute covariance matrix from a DataFrame of returns."""
    cols = df.columns
    cov = np.atleast_2d(np.cov(df.to_numpy().T))
    return pl.DataFrame(dict(zip(cols, cov, strict=True)))

compute_returns(prices)

Compute simple returns from prices.

Drops leading all-null rows produced by pct_change and fills remaining nulls/NaNs (e.g. from missing prices) with zero returns.

Warning

Filling with zero is a dirty helper, not a data-cleaning strategy: a suspended asset or an asset listed mid-sample contributes fabricated zero returns that bias variance down and correlations toward zero. A :class:UserWarning is emitted whenever this path fires. Callers working with messy universes should clean the data first and compose the pipeline manually::

from pyhrp import build_tree, risk_parity
from pyhrp import compute_corr, compute_cov

returns = prices.select(pl.all().pct_change()).drop_nulls()
root = risk_parity(root=build_tree(compute_corr(returns)).root,
                   cov=compute_cov(returns))

Parameters:

Name Type Description Default
prices DataFrame

Asset price time series (columns are assets, rows are dates)

 required

Returns:

Type Description
DataFrame

pl.DataFrame: Simple returns, one row shorter than prices

Examples:

>>> import polars as pl
>>> from pyhrp.covariance import compute_returns
>>> prices = pl.DataFrame({"A": [100.0, 110.0, 99.0]})
>>> compute_returns(prices)['A'].to_list()
[0.1, -0.1]
Source code in src/pyhrp/covariance.py 
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def compute_returns(prices: pl.DataFrame) -> pl.DataFrame:
    r"""Compute simple returns from prices.

    Drops leading all-null rows produced by pct_change and fills remaining
    nulls/NaNs (e.g. from missing prices) with zero returns.

    Warning:
        Filling with zero is a dirty helper, not a data-cleaning strategy: a
        suspended asset or an asset listed mid-sample contributes fabricated
        zero returns that bias variance down and correlations toward zero.
        A :class:`UserWarning` is emitted whenever this path fires. Callers
        working with messy universes should clean the data first and compose
        the pipeline manually::

            from pyhrp import build_tree, risk_parity
            from pyhrp import compute_corr, compute_cov

            returns = prices.select(pl.all().pct_change()).drop_nulls()
            root = risk_parity(root=build_tree(compute_corr(returns)).root,
                               cov=compute_cov(returns))

    Args:
        prices (pl.DataFrame): Asset price time series (columns are assets, rows are dates)

    Returns:
        pl.DataFrame: Simple returns, one row shorter than ``prices``

    Examples:
        >>> import polars as pl
        >>> from pyhrp.covariance import compute_returns
        >>> prices = pl.DataFrame({"A": [100.0, 110.0, 99.0]})
        >>> compute_returns(prices)['A'].to_list()
        [0.1, -0.1]
    """
    raw = prices.select(pl.all().pct_change()).filter(pl.any_horizontal(pl.all().is_not_null()))
    _warn_on_missing_returns(raw)
    return raw.fill_null(0.0).fill_nan(0.0)

hrp(prices, node=None, method='ward', bisection=False)

Compute the hierarchical risk parity portfolio weights.

This is the main entry point for the HRP algorithm. It calculates returns from prices, builds a hierarchical clustering tree if not provided, and applies risk parity weights.

Note

A node passed in is weighted in place and returned; see the allocator contract in :mod:pyhrp.algos. Passing dendrogram.root therefore rewrites the portfolios inside that Dendrogram, even though the Dendrogram itself is a frozen dataclass.

Parameters:

Name Type Description Default
prices DataFrame

Asset price time series (columns are assets, rows are dates)

 required
node Cluster

Root node of the hierarchical clustering tree. If None, a tree will be built from the correlation matrix.

 None
method Literal['single', 'complete', 'average', 'ward']

Linkage method to use for distance calculation - "single": minimum distance between points (nearest neighbor) - "complete": maximum distance between points (furthest neighbor) - "average": average distance between all points - "ward": Ward variance minimization

 'ward'
bisection bool

Whether to use bisection method for tree construction

 False

Returns:

Name Type Description
Cluster Cluster

The root cluster with portfolio weights assigned according to HRP
Source code in src/pyhrp/hrp.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def hrp(
    prices: pl.DataFrame,
    node: Cluster | None = None,
    method: Literal["single", "complete", "average", "ward"] = "ward",
    bisection: bool = False,
) -> Cluster:
    """Compute the hierarchical risk parity portfolio weights.

    This is the main entry point for the HRP algorithm. It calculates returns from prices,
    builds a hierarchical clustering tree if not provided, and applies risk parity weights.

    Note:
        A ``node`` passed in is weighted in place and returned; see the allocator
        contract in :mod:`pyhrp.algos`. Passing ``dendrogram.root`` therefore
        rewrites the portfolios inside that ``Dendrogram``, even though the
        ``Dendrogram`` itself is a frozen dataclass.

    Args:
        prices (pl.DataFrame): Asset price time series (columns are assets, rows are dates)
        node (Cluster, optional): Root node of the hierarchical clustering tree.
            If None, a tree will be built from the correlation matrix.
        method (Literal["single", "complete", "average", "ward"]): Linkage method to use for distance calculation
            - "single": minimum distance between points (nearest neighbor)
            - "complete": maximum distance between points (furthest neighbor)
            - "average": average distance between all points
            - "ward": Ward variance minimization
        bisection (bool): Whether to use bisection method for tree construction

    Returns:
        Cluster: The root cluster with portfolio weights assigned according to HRP
    """
    returns = compute_returns(prices)
    cov = check_finite_matrix(compute_cov(returns), name="covariance matrix")
    cor = compute_corr(returns)
    node = node or build_tree(cor, method=method, bisection=bisection).root

    return risk_parity(root=node, cov=cov)

schur_hrp(prices, node=None, method='ward', bisection=False, gamma=0.5)

Compute Schur Complementary Allocation portfolio weights.

Extends HRP by augmenting each sub-covariance block with off-diagonal information via Schur complements before splitting risk between clusters. Introduced by Peter Cotton (arXiv:2411.05807). At gamma=0 this is identical to HRP; at gamma=1 it recovers the global minimum-variance portfolio through the same recursive hierarchy.

Note

A node passed in is weighted in place and returned; see the allocator contract in :mod:pyhrp.algos. Passing dendrogram.root therefore rewrites the portfolios inside that Dendrogram, even though the Dendrogram itself is a frozen dataclass.

Parameters:

Name Type Description Default
prices DataFrame

Asset price time series (columns are assets, rows are dates)

 required
node Cluster

Root node of the hierarchical clustering tree. If None, a tree will be built from the correlation matrix.

 None
method Literal['single', 'complete', 'average', 'ward']

Linkage method for clustering

 'ward'
bisection bool

Whether to use bisection method for tree construction

 False
gamma float

Schur interpolation parameter in [0, 1]. 0 recovers standard HRP; 1 recovers minimum-variance portfolio.

 0.5

Returns:

Name Type Description
Cluster Cluster

The root cluster with portfolio weights assigned
Source code in src/pyhrp/hrp.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def schur_hrp(
    prices: pl.DataFrame,
    node: Cluster | None = None,
    method: Literal["single", "complete", "average", "ward"] = "ward",
    bisection: bool = False,
    gamma: float = 0.5,
) -> Cluster:
    """Compute Schur Complementary Allocation portfolio weights.

    Extends HRP by augmenting each sub-covariance block with off-diagonal information
    via Schur complements before splitting risk between clusters. Introduced by Peter Cotton
    (arXiv:2411.05807). At gamma=0 this is identical to HRP; at gamma=1 it recovers the
    global minimum-variance portfolio through the same recursive hierarchy.

    Note:
        A ``node`` passed in is weighted in place and returned; see the allocator
        contract in :mod:`pyhrp.algos`. Passing ``dendrogram.root`` therefore
        rewrites the portfolios inside that ``Dendrogram``, even though the
        ``Dendrogram`` itself is a frozen dataclass.

    Args:
        prices (pl.DataFrame): Asset price time series (columns are assets, rows are dates)
        node (Cluster, optional): Root node of the hierarchical clustering tree.
            If None, a tree will be built from the correlation matrix.
        method (Literal["single", "complete", "average", "ward"]): Linkage method for clustering
        bisection (bool): Whether to use bisection method for tree construction
        gamma (float): Schur interpolation parameter in [0, 1].
            0 recovers standard HRP; 1 recovers minimum-variance portfolio.

    Returns:
        Cluster: The root cluster with portfolio weights assigned
    """
    returns = compute_returns(prices)
    cov = check_finite_matrix(compute_cov(returns), name="covariance matrix")
    cor = compute_corr(returns)
    node = node or build_tree(cor, method=method, bisection=bisection).root

    return schur_risk_parity(root=node, cov=cov, gamma=gamma)

Portfolio optimization algorithms for hierarchical risk parity.

This module implements various portfolio optimization algorithms: - risk_parity: The main hierarchical risk parity algorithm - schur_risk_parity: Schur Complementary Allocation (Cotton, arXiv:2411.05807) - one_over_n: A simple equal-weight allocation strategy

Allocator contract

All three allocators take the same inputs — a Cluster tree (root) plus the asset names — and none of them changes the shape of that tree: no node is added, removed or re-parented. They differ in where the weights land.

risk_parity and schur_risk_parity share the recursive _allocate_with scaffolding and write into the tree they are given: every node's portfolio is replaced rather than accumulated into, and the same root object is returned. Replacing makes them idempotent — re-running on an already weighted tree, with a different covariance matrix or gamma, gives the same answer as running on a fresh one — but the caller's tree is modified. Note in particular that Dendrogram is a frozen dataclass while the Cluster it holds is not: passing dendrogram.root to either allocator rewrites the portfolios inside that dendrogram. Pass copy.deepcopy(dendrogram.root) to keep the original tree unweighted.

one_over_n differs on both counts. It accumulates into a local buffer, so it leaves the input tree untouched entirely, and its output is a generator yielding the equal-weight portfolio one tree level at a time (see its docstring), because its purpose is to expose the allocation as it deepens rather than a single final result.

one_over_n(root, assets)

Generate 1/N (equal-weight) portfolios one tree level at a time.

This implements a hierarchical 1/N strategy where weights are distributed equally among the leaves of each cluster, and the weight budget halves at each successive level of the tree.

Unlike :func:risk_parity and :func:schur_risk_parity — which rebuild a single final allocation and return the root Cluster — this allocator is intentionally a generator: its purpose is to expose the equal-weight allocation as the tree deepens, yielding one portfolio per level. It shares the sibling input contract (a Cluster tree plus the asset names) and, like them, does not mutate the tree: weights accumulate in a local buffer, so a leaf that terminates at a shallow level keeps its weight in the deeper levels (each yielded portfolio is therefore a complete allocation over all assets), and re-running on the same tree yields an identical sequence.

Parameters:

Name Type Description Default
root Cluster

The root node of the cluster tree.

 required
assets list[str]

Asset names; a leaf's value indexes into this list.

 required

Yields:

Type Description
Generator[tuple[int, Portfolio]]

tuple[int, Portfolio]: The level number and the (cumulative) equal-weight
Generator[tuple[int, Portfolio]]

portfolio at that level.

Examples:

>>> import polars as pl
>>> from pyhrp.hrp import build_tree
>>> from pyhrp.algos import one_over_n
>>> cor = pl.DataFrame({"A": [1.0, 0.3], "B": [0.3, 1.0]})
>>> dg = build_tree(cor, method="ward")
>>> levels = list(one_over_n(dg.root, dg.assets))
>>> len(levels) > 0
True
Source code in src/pyhrp/algos.py 
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
def one_over_n(root: Cluster, assets: list[str]) -> Generator[tuple[int, Portfolio]]:
    """Generate 1/N (equal-weight) portfolios one tree level at a time.

    This implements a hierarchical 1/N strategy where weights are distributed
    equally among the leaves of each cluster, and the weight budget halves at
    each successive level of the tree.

    Unlike :func:`risk_parity` and :func:`schur_risk_parity` — which rebuild a
    single final allocation and return the root ``Cluster`` — this allocator is
    intentionally a **generator**: its purpose is to expose the equal-weight
    allocation as the tree deepens, yielding one portfolio per level. It shares
    the sibling input contract (a ``Cluster`` tree plus the asset names) and, like
    them, does not mutate the tree: weights accumulate in a local buffer, so a
    leaf that terminates at a shallow level keeps its weight in the deeper levels
    (each yielded portfolio is therefore a complete allocation over all assets),
    and re-running on the same tree yields an identical sequence.

    Args:
        root (Cluster): The root node of the cluster tree.
        assets (list[str]): Asset names; a leaf's value indexes into this list.

    Yields:
        tuple[int, Portfolio]: The level number and the (cumulative) equal-weight
        portfolio at that level.

    Examples:
        >>> import polars as pl
        >>> from pyhrp.hrp import build_tree
        >>> from pyhrp.algos import one_over_n
        >>> cor = pl.DataFrame({"A": [1.0, 0.3], "B": [0.3, 1.0]})
        >>> dg = build_tree(cor, method="ward")
        >>> levels = list(one_over_n(dg.root, dg.assets))
        >>> len(levels) > 0
        True
    """
    # Accumulate into a local buffer so the input tree is never mutated.
    portfolio = Portfolio()

    # Initial weight to distribute
    w: float = 1.0

    # Process each level of the tree
    for n, level in enumerate(root.levels):
        for node in level:
            # Distribute weight equally among all leaves in this node
            for leaf in node.leaves:
                portfolio[assets[leaf.value]] = w / node.leaf_count

        # Reduce weight for the next level
        w *= 0.5

        # Yield the current level number and a deep copy of the portfolio
        yield n, deepcopy(portfolio)

risk_parity(root, cov)

Compute hierarchical risk parity weights for a cluster tree.

This is the main algorithm for hierarchical risk parity. It recursively traverses the cluster tree and assigns weights to each node based on the risk parity principle.

Note

The tree is modified in place: the portfolio of every node is rebuilt from scratch and root itself is returned, so the function is idempotent and a tree can be reused with a different covariance matrix. Deep-copy the tree first if you need to keep it unweighted.

Parameters:

Name Type Description Default
root Cluster

The root node of the cluster tree

 required
cov DataFrame

Covariance matrix of asset returns

 required

Returns:

Name Type Description
Cluster Cluster

The same root node, with portfolio weights assigned

Examples:

>>> import polars as pl
>>> from pyhrp.cluster import Cluster
>>> from pyhrp.algos import risk_parity
>>> cov = pl.DataFrame({"A": [4.0, 0.0], "B": [0.0, 1.0]})
>>> root = Cluster(2, left=Cluster(0), right=Cluster(1))
>>> cluster = risk_parity(root=root, cov=cov)
>>> round(cluster.portfolio["B"], 1)
0.8
Source code in src/pyhrp/algos.py 
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def risk_parity(root: Cluster, cov: pl.DataFrame) -> Cluster:
    """Compute hierarchical risk parity weights for a cluster tree.

    This is the main algorithm for hierarchical risk parity. It recursively
    traverses the cluster tree and assigns weights to each node based on
    the risk parity principle.

    Note:
        The tree is modified in place: the portfolio of every node is rebuilt
        from scratch and ``root`` itself is returned, so the function is
        idempotent and a tree can be reused with a different covariance matrix.
        Deep-copy the tree first if you need to keep it unweighted.

    Args:
        root (Cluster): The root node of the cluster tree
        cov (pl.DataFrame): Covariance matrix of asset returns

    Returns:
        Cluster: The same root node, with portfolio weights assigned

    Examples:
        >>> import polars as pl
        >>> from pyhrp.cluster import Cluster
        >>> from pyhrp.algos import risk_parity
        >>> cov = pl.DataFrame({"A": [4.0, 0.0], "B": [0.0, 1.0]})
        >>> root = Cluster(2, left=Cluster(0), right=Cluster(1))
        >>> cluster = risk_parity(root=root, cov=cov)
        >>> round(cluster.portfolio["B"], 1)
        0.8
    """

    def node_variances(left: Cluster, right: Cluster, cov_np: np.ndarray, index: dict[str, int]) -> tuple[float, float]:
        """Plain block variance of each child sub-portfolio."""
        return (
            _block_variance(left.portfolio, cov_np, index),
            _block_variance(right.portfolio, cov_np, index),
        )

    return _allocate_with(root, cov, node_variances)

schur_risk_parity(root, cov, gamma=0.5)

Compute Schur Complementary Allocation weights for a cluster tree.

An extension of HRP introduced by Peter Cotton (arXiv:2411.05807) that augments sub-covariance matrices with off-diagonal block information via Schur complements. At gamma=0 this recovers standard HRP; at gamma=1 it recovers the minimum-variance portfolio through the same recursive structure.

Note

The tree is modified in place: the portfolio of every node is rebuilt from scratch and root itself is returned, so the function is idempotent and a tree can be reused with a different covariance matrix or gamma. Deep-copy the tree first if you need to keep it unweighted.

Parameters:

Name Type Description Default
root Cluster

The root node of the cluster tree

 required
cov DataFrame

Covariance matrix of asset returns

 required
gamma float

Interpolation parameter in [0, 1]. 0 = HRP, 1 = minimum variance.

 0.5

Returns:

Name Type Description
Cluster Cluster

The same root node, with portfolio weights assigned

Raises:

Type Description
ValueError

If gamma is outside the interval [0, 1].

Examples:

>>> import polars as pl
>>> from pyhrp.cluster import Cluster
>>> from pyhrp.algos import schur_risk_parity
>>> cov = pl.DataFrame({"A": [4.0, 0.0], "B": [0.0, 1.0]})
>>> root = Cluster(2, left=Cluster(0), right=Cluster(1))
>>> cluster = schur_risk_parity(root=root, cov=cov, gamma=0.5)
>>> round(cluster.portfolio["B"], 1)
0.8
Source code in src/pyhrp/algos.py 
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
def schur_risk_parity(root: Cluster, cov: pl.DataFrame, gamma: float = 0.5) -> Cluster:
    """Compute Schur Complementary Allocation weights for a cluster tree.

    An extension of HRP introduced by Peter Cotton (arXiv:2411.05807) that augments
    sub-covariance matrices with off-diagonal block information via Schur complements.
    At gamma=0 this recovers standard HRP; at gamma=1 it recovers the minimum-variance
    portfolio through the same recursive structure.

    Note:
        The tree is modified in place: the portfolio of every node is rebuilt
        from scratch and ``root`` itself is returned, so the function is
        idempotent and a tree can be reused with a different covariance matrix
        or gamma. Deep-copy the tree first if you need to keep it unweighted.

    Args:
        root (Cluster): The root node of the cluster tree
        cov (pl.DataFrame): Covariance matrix of asset returns
        gamma (float): Interpolation parameter in [0, 1]. 0 = HRP, 1 = minimum variance.

    Returns:
        Cluster: The same root node, with portfolio weights assigned

    Raises:
        ValueError: If gamma is outside the interval [0, 1].

    Examples:
        >>> import polars as pl
        >>> from pyhrp.cluster import Cluster
        >>> from pyhrp.algos import schur_risk_parity
        >>> cov = pl.DataFrame({"A": [4.0, 0.0], "B": [0.0, 1.0]})
        >>> root = Cluster(2, left=Cluster(0), right=Cluster(1))
        >>> cluster = schur_risk_parity(root=root, cov=cov, gamma=0.5)
        >>> round(cluster.portfolio["B"], 1)
        0.8
    """
    if not 0.0 <= gamma <= 1.0:
        msg = f"gamma must be in [0, 1], got {gamma}"
        raise ValueError(msg)

    def node_variances(left: Cluster, right: Cluster, cov_np: np.ndarray, index: dict[str, int]) -> tuple[float, float]:
        """Schur-augmented block variance of each child, conditioned on the other."""
        li = [index[a] for a in left.portfolio.assets]
        ri = [index[a] for a in right.portfolio.assets]

        a_mat = cov_np[np.ix_(li, li)]
        b_mat = cov_np[np.ix_(li, ri)]
        d_mat = cov_np[np.ix_(ri, ri)]

        w_left = np.array([left.portfolio[a] for a in left.portfolio.assets])
        w_right = np.array([right.portfolio[a] for a in right.portfolio.assets])

        # Schur-augmented blocks: condition each group on the other
        a_aug = a_mat - gamma * (b_mat @ _solve(d_mat, b_mat.T))
        d_aug = d_mat - gamma * (b_mat.T @ _solve(a_mat, b_mat))

        v_left = float(w_left @ a_aug @ w_left)
        v_right = float(w_right @ d_aug @ w_right)
        return v_left, v_right

    return _allocate_with(root, cov, node_variances)

Data structures for hierarchical risk parity portfolio optimization.

This module defines the core data structures used in the hierarchical risk parity algorithm: - Portfolio: Manages a collection of asset weights (strings identify assets) - Cluster: Represents a node in the hierarchical clustering tree

Cluster

Bases: Node[int]

Represents a cluster in the hierarchical clustering tree.

Clusters are the nodes of the graphs we build. Each cluster is aware of the left and the right cluster it is connecting to. Each cluster also has an associated portfolio.

Attributes:

Name Type Description
portfolio Portfolio

The portfolio associated with this cluster
Source code in src/pyhrp/cluster.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
class Cluster(Node[int]):
    """Represents a cluster in the hierarchical clustering tree.

    Clusters are the nodes of the graphs we build.
    Each cluster is aware of the left and the right cluster
    it is connecting to. Each cluster also has an associated portfolio.

    Attributes:
        portfolio (Portfolio): The portfolio associated with this cluster
    """

    def __init__(self, value: int, left: Cluster | None = None, right: Cluster | None = None) -> None:
        """Initialize a new Cluster.

        Args:
            value (int): The identifier for this cluster
            left (Cluster, optional): The left child cluster
            right (Cluster, optional): The right child cluster
        """
        super().__init__(value=value, left=left, right=right)
        self.portfolio = Portfolio()

    # Override narrows the return type to list[Cluster] and validates tree integrity;
    # the traversal order (left to right) matches Node.leaves.
    @property
    def leaves(self) -> list[Cluster]:
        """Get all reachable leaf nodes in left-to-right dendrogram order.

        Returns:
            list[Cluster]: List of all leaf nodes reachable from this cluster
        """
        # Iterative for the same reason as Node.leaves, and still validating each
        # non-leaf node through _child_clusters() so a malformed tree raises here
        # rather than yielding a silently short leaf list.
        result: list[Cluster] = []
        stack: list[Cluster] = [self]
        while stack:
            node = stack.pop()
            if node.is_leaf:
                result.append(node)
                continue
            left, right = node._child_clusters()
            # Right first, so the left subtree is emitted first.
            stack.append(right)
            stack.append(left)
        return result

    def _child_clusters(self) -> tuple[Cluster, Cluster]:
        """Return the validated (left, right) child clusters of a non-leaf node.

        Raises:
            ValueError: If either child is missing on a non-leaf cluster.
            TypeError: If either child is not a Cluster.
        """
        if self.left is None:
            msg = "Expected left child to exist for non-leaf cluster"
            raise ValueError(msg)
        if self.right is None:
            msg = "Expected right child to exist for non-leaf cluster"
            raise ValueError(msg)
        if not isinstance(self.left, Cluster):
            msg = f"Expected left child to be a Cluster for node {self.value}"
            raise TypeError(msg)
        if not isinstance(self.right, Cluster):
            msg = f"Expected right child to be a Cluster for node {self.value}"
            raise TypeError(msg)
        return self.left, self.right

leaves property

Get all reachable leaf nodes in left-to-right dendrogram order.

Returns:

Type Description
list[Cluster]

list[Cluster]: List of all leaf nodes reachable from this cluster

__init__(value, left=None, right=None)

Initialize a new Cluster.

Parameters:

Name Type Description Default
value int

The identifier for this cluster

 required
left Cluster

The left child cluster

 None
right Cluster

The right child cluster

 None
Source code in src/pyhrp/cluster.py 
130
131
132
133
134
135
136
137
138
139
def __init__(self, value: int, left: Cluster | None = None, right: Cluster | None = None) -> None:
    """Initialize a new Cluster.

    Args:
        value (int): The identifier for this cluster
        left (Cluster, optional): The left child cluster
        right (Cluster, optional): The right child cluster
    """
    super().__init__(value=value, left=left, right=right)
    self.portfolio = Portfolio()

Portfolio dataclass

Container for portfolio asset weights.

This lightweight class stores and manipulates a mapping from asset names to their portfolio weights, and provides convenience helpers for analysis and visualization.

Attributes:

Name Type Description
_weights dict[str, float]

Internal mapping from asset symbol to weight.
Source code in src/pyhrp/cluster.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
@dataclass
class Portfolio:
    """Container for portfolio asset weights.

    This lightweight class stores and manipulates a mapping from asset names to
    their portfolio weights, and provides convenience helpers for analysis and
    visualization.

    Attributes:
        _weights (dict[str, float]): Internal mapping from asset symbol to weight.
    """

    _weights: dict[str, float] = field(default_factory=dict)

    @property
    def assets(self) -> list[str]:
        """List of asset names present in the portfolio.

        Returns:
            list[str]: Asset identifiers in insertion order (Python 3.7+ dict order).
        """
        return list(self._weights.keys())

    def variance(self, cov: pl.DataFrame) -> float:
        """Calculate the variance of the portfolio.

        Args:
            cov (pl.DataFrame): Covariance matrix where columns and rows correspond
                to assets in the same order as columns list.

        Returns:
            float: Portfolio variance
        """
        assets = self.assets
        index = {name: i for i, name in enumerate(cov.columns)}
        row_indices = [index[a] for a in assets]
        cov_matrix = cov.to_numpy()
        c = cov_matrix[np.ix_(row_indices, row_indices)]
        w = np.array([self._weights[a] for a in assets])
        return float(w @ c @ w)

    def __getitem__(self, item: str) -> float:
        """Return the weight for a given asset.

        Args:
            item (str): Asset name/symbol.

        Returns:
            float: The weight associated with the asset.

        Raises:
            KeyError: If the asset is not present in the portfolio.
        """
        return self._weights[item]

    def __setitem__(self, key: str, value: float) -> None:
        """Set or update the weight for an asset.

        Args:
            key (str): Asset name/symbol.
            value (float): Portfolio weight for the asset.
        """
        self._weights[key] = value

    @property
    def weights(self) -> dict[str, float]:
        """Get all weights as a dict sorted alphabetically by asset name.

        Returns:
            dict[str, float]: Mapping from asset name to weight, sorted by name.
        """
        return dict(sorted(self._weights.items()))

    def plot(self, names: list[str]) -> go.Figure:
        """Plot the portfolio weights as a bar chart.

        Args:
            names (list[str]): List of asset names to include in the plot

        Returns:
            go.Figure: The plotly figure

        Note:
            The plotly dependency is imported lazily so importing the allocation
            core (which pulls in this module) stays plotly-free.
        """
        import plotly.graph_objects as go

        w = self.weights
        values = [w[n] for n in names]
        fig = go.Figure(go.Bar(x=names, y=values, marker_color="steelblue"))
        fig.update_layout(xaxis={"tickangle": -90})
        return fig

assets property

List of asset names present in the portfolio.

Returns:

Type Description
list[str]

list[str]: Asset identifiers in insertion order (Python 3.7+ dict order).

weights property

Get all weights as a dict sorted alphabetically by asset name.

Returns:

Type Description
dict[str, float]

dict[str, float]: Mapping from asset name to weight, sorted by name.

__getitem__(item)

Return the weight for a given asset.

Parameters:

Name Type Description Default
item str

Asset name/symbol.

 required

Returns:

Name Type Description
float float

The weight associated with the asset.

Raises:

Type Description
KeyError

If the asset is not present in the portfolio.
Source code in src/pyhrp/cluster.py 
65
66
67
68
69
70
71
72
73
74
75
76
77
def __getitem__(self, item: str) -> float:
    """Return the weight for a given asset.

    Args:
        item (str): Asset name/symbol.

    Returns:
        float: The weight associated with the asset.

    Raises:
        KeyError: If the asset is not present in the portfolio.
    """
    return self._weights[item]

__setitem__(key, value)

Set or update the weight for an asset.

Parameters:

Name Type Description Default
key str

Asset name/symbol.

 required
value float

Portfolio weight for the asset.

 required
Source code in src/pyhrp/cluster.py 
79
80
81
82
83
84
85
86
def __setitem__(self, key: str, value: float) -> None:
    """Set or update the weight for an asset.

    Args:
        key (str): Asset name/symbol.
        value (float): Portfolio weight for the asset.
    """
    self._weights[key] = value

plot(names)

Plot the portfolio weights as a bar chart.

Parameters:

Name Type Description Default
names list[str]

List of asset names to include in the plot

 required

Returns:

Type Description
Figure

go.Figure: The plotly figure
Note

The plotly dependency is imported lazily so importing the allocation core (which pulls in this module) stays plotly-free.

Source code in src/pyhrp/cluster.py 
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
def plot(self, names: list[str]) -> go.Figure:
    """Plot the portfolio weights as a bar chart.

    Args:
        names (list[str]): List of asset names to include in the plot

    Returns:
        go.Figure: The plotly figure

    Note:
        The plotly dependency is imported lazily so importing the allocation
        core (which pulls in this module) stays plotly-free.
    """
    import plotly.graph_objects as go

    w = self.weights
    values = [w[n] for n in names]
    fig = go.Figure(go.Bar(x=names, y=values, marker_color="steelblue"))
    fig.update_layout(xaxis={"tickangle": -90})
    return fig

variance(cov)

Calculate the variance of the portfolio.

Parameters:

Name Type Description Default
cov DataFrame

Covariance matrix where columns and rows correspond to assets in the same order as columns list.

 required

Returns:

Name Type Description
float float

Portfolio variance
Source code in src/pyhrp/cluster.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def variance(self, cov: pl.DataFrame) -> float:
    """Calculate the variance of the portfolio.

    Args:
        cov (pl.DataFrame): Covariance matrix where columns and rows correspond
            to assets in the same order as columns list.

    Returns:
        float: Portfolio variance
    """
    assets = self.assets
    index = {name: i for i, name in enumerate(cov.columns)}
    row_indices = [index[a] for a in assets]
    cov_matrix = cov.to_numpy()
    c = cov_matrix[np.ix_(row_indices, row_indices)]
    w = np.array([self._weights[a] for a in assets])
    return float(w @ c @ w)

Plotly visualization for hierarchical clustering dendrograms.

Kept separate from the allocation core (hrp.py) so that the optional plotting dependency (plotly) and the visualization logic do not couple the algorithm modules. Dendrogram.plot delegates here.

plot_dendrogram(dendrogram, **kwargs)

Build and return a plotly dendrogram figure for a Dendrogram.

Parameters:

Name Type Description Default
dendrogram Dendrogram

Clustering result carrying the linkage matrix and asset labels to render.

 required
**kwargs object

Extra keyword arguments forwarded to scipy.cluster.hierarchy.dendrogram (e.g. color_threshold).

 {}

Returns:

Type Description
Figure

go.Figure: A plotly figure drawing the dendrogram as line traces.

Raises:

Type Description
ValueError

If the dendrogram has no linkage matrix to plot.

Examples:

>>> import polars as pl
>>> from pyhrp.hrp import build_tree
>>> from pyhrp.plot import plot_dendrogram
>>> cor = pl.DataFrame({"A": [1.0, 0.5], "B": [0.5, 1.0]})
>>> fig = plot_dendrogram(build_tree(cor, method="ward"))
>>> type(fig).__name__
'Figure'
Source code in src/pyhrp/plot.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def plot_dendrogram(dendrogram: Dendrogram, **kwargs: object) -> go.Figure:
    """Build and return a plotly dendrogram figure for a Dendrogram.

    Args:
        dendrogram (Dendrogram): Clustering result carrying the linkage matrix
            and asset labels to render.
        **kwargs (object): Extra keyword arguments forwarded to
            ``scipy.cluster.hierarchy.dendrogram`` (e.g. ``color_threshold``).

    Returns:
        go.Figure: A plotly figure drawing the dendrogram as line traces.

    Raises:
        ValueError: If the dendrogram has no linkage matrix to plot.

    Examples:
        >>> import polars as pl
        >>> from pyhrp.hrp import build_tree
        >>> from pyhrp.plot import plot_dendrogram
        >>> cor = pl.DataFrame({"A": [1.0, 0.5], "B": [0.5, 1.0]})
        >>> fig = plot_dendrogram(build_tree(cor, method="ward"))
        >>> type(fig).__name__
        'Figure'
    """
    if dendrogram.linkage is None:
        msg = "Dendrogram has no linkage matrix to plot."
        raise ValueError(msg)
    ddata = sch.dendrogram(dendrogram.linkage, labels=dendrogram.assets, no_plot=True, **kwargs)
    fig = go.Figure()
    for xs, ys in zip(ddata["icoord"], ddata["dcoord"], strict=False):
        fig.add_trace(go.Scatter(x=xs, y=ys, mode="lines", line={"color": "steelblue"}, showlegend=False))
    n = len(dendrogram.assets)
    fig.update_layout(
        xaxis={
            "tickmode": "array",
            "tickvals": [5 + 10 * i for i in range(n)],
            "ticktext": ddata["ivl"],
            "tickangle": -90,
        },
    )
    return fig

A lightweight binary tree implementation to replace the binarytree dependency.

This module provides a simple Node class that can be used to create binary trees. It implements only the functionality needed by the pyhrp package.

Node

Bases: Generic[T]

A binary tree node with left and right children.

This class implements the minimal functionality needed from the binarytree.Node class that is used in the pyhrp package.

Attributes:

Name Type Description
value

The value of the node
left

The left child node
right

The right child node
Source code in src/pyhrp/treelib.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
class Node(Generic[T]):
    """A binary tree node with left and right children.

    This class implements the minimal functionality needed from the binarytree.Node class
    that is used in the pyhrp package.

    Attributes:
        value: The value of the node
        left: The left child node
        right: The right child node
    """

    def __init__(self, value: T, left: Node[T] | None = None, right: Node[T] | None = None) -> None:
        """Initialize a new Node.

        Args:
            value: The value of the node
            left: The left child node
            right: The right child node
        """
        self.value = value
        self.left = left
        self.right = right

    @property
    def is_leaf(self) -> bool:
        """Check if this node is a leaf node (has no children).

        Returns:
            bool: True if this is a leaf node, False otherwise
        """
        return self.left is None and self.right is None

    @property
    def leaves(self) -> Sequence[Node[T]]:
        """Get all leaf nodes in the tree rooted at this node.

        Returns:
            List[Node]: List of all leaf nodes
        """
        # Iterative depth-first walk with an explicit stack: the leaf order is the
        # same left-to-right order the recursive form produced, but the traversal
        # depth is bounded by the heap rather than by sys.getrecursionlimit(). A
        # chain-degenerate cluster tree is as deep as it is wide, so recursing here
        # put a ceiling on the number of assets the package could handle.
        result: list[Node[T]] = []
        stack: list[Node[T]] = [self]
        while stack:
            node = stack.pop()
            if node.is_leaf:
                result.append(node)
                continue
            # Right first, so the left subtree is popped and emitted first.
            if node.right is not None:
                stack.append(node.right)
            if node.left is not None:
                stack.append(node.left)

        return result

    @property
    def levels(self) -> list[list[Node[T]]]:
        """Get nodes by level in the tree.

        Returns:
            List[List[Node]]: List of lists of nodes at each level
        """
        result: list[list[Node[T]]] = []
        current_level: list[Node[T]] = [self]

        while current_level:
            result.append(current_level)
            next_level = []

            for node in current_level:
                if node.left:
                    next_level.append(node.left)
                if node.right:
                    next_level.append(node.right)

            current_level = next_level

        return result

    @property
    def leaf_count(self) -> int:
        """Count the number of leaf nodes in the tree.

        Returns:
            int: Number of leaf nodes
        """
        return len(self.leaves)

    @property
    def size(self) -> int:
        """Count the total number of nodes in the tree.

        Returns:
            int: Total number of nodes
        """
        # Counts via __iter__, which is already an iterative level-order walk, so
        # this inherits its freedom from the recursion limit.
        return sum(1 for _ in self)

    def __iter__(self) -> Iterator[Node[T]]:
        """Iterate through all nodes in the tree in level-order.

        Returns:
            Iterator[Node]: Iterator over all nodes
        """
        queue: deque[Node[T]] = deque([self])
        while queue:
            node = queue.popleft()
            yield node
            if node.left:
                queue.append(node.left)
            if node.right:
                queue.append(node.right)

is_leaf property

Check if this node is a leaf node (has no children).

Returns:

Name Type Description
bool bool

True if this is a leaf node, False otherwise

leaf_count property

Count the number of leaf nodes in the tree.

Returns:

Name Type Description
int int

Number of leaf nodes

leaves property

Get all leaf nodes in the tree rooted at this node.

Returns:

Type Description
Sequence[Node[T]]

List[Node]: List of all leaf nodes

levels property

Get nodes by level in the tree.

Returns:

Type Description
list[list[Node[T]]]

List[List[Node]]: List of lists of nodes at each level

size property

Count the total number of nodes in the tree.

Returns:

Name Type Description
int int

Total number of nodes

__init__(value, left=None, right=None)

Initialize a new Node.

Parameters:

Name Type Description Default
value T

The value of the node

 required
left Node[T] | None

The left child node

 None
right Node[T] | None

The right child node

 None
Source code in src/pyhrp/treelib.py 
33
34
35
36
37
38
39
40
41
42
43
def __init__(self, value: T, left: Node[T] | None = None, right: Node[T] | None = None) -> None:
    """Initialize a new Node.

    Args:
        value: The value of the node
        left: The left child node
        right: The right child node
    """
    self.value = value
    self.left = left
    self.right = right

__iter__()

Iterate through all nodes in the tree in level-order.

Returns:

Type Description
Iterator[Node[T]]

Iterator[Node]: Iterator over all nodes
Source code in src/pyhrp/treelib.py 
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def __iter__(self) -> Iterator[Node[T]]:
    """Iterate through all nodes in the tree in level-order.

    Returns:
        Iterator[Node]: Iterator over all nodes
    """
    queue: deque[Node[T]] = deque([self])
    while queue:
        node = queue.popleft()
        yield node
        if node.left:
            queue.append(node.left)
        if node.right:
            queue.append(node.right)