API Reference¶

def hrp(
    prices: pl.DataFrame,
    node: Cluster | None = None,
    method: Method = "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
    """
    cov, node = _prepare(prices, node, method, bisection)
    return risk_parity(root=node, cov=cov)

def schur_hrp(
    prices: pl.DataFrame,
    node: Cluster | None = None,
    method: Method = "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
    """
    cov, node = _prepare(prices, node, method, bisection)
    return schur_risk_parity(root=node, cov=cov, gamma=gamma)

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)

def risk_parity(root: Cluster, cov: pl.DataFrame | CovarianceOperator, assets: Sequence[str] | None = None) -> 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.

    The covariance is reached only through block products, one per child of
    every node, so a factor or Gram :class:`~pyhrp.operators.CovarianceOperator`
    allocates without forming the ``n x n`` matrix.

    Args:
        root (Cluster): The root node of the cluster tree
        cov (pl.DataFrame | CovarianceOperator): Covariance matrix of asset returns,
            as a DataFrame or as an operator
        assets (Sequence[str], optional): Asset names for an operator ``cov``;
            defaults to ``"0", "1", ...``. A DataFrame names its own assets.

    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: Block, right: Block, op: CovarianceOperator) -> tuple[float, float]:
        """Plain block variance of each child sub-portfolio."""
        return _block_variance(left, op), _block_variance(right, op)

    return _allocate_with(root, cov, node_variances, assets)

def schur_risk_parity(
    root: Cluster,
    cov: pl.DataFrame | CovarianceOperator,
    gamma: float = 0.5,
    assets: Sequence[str] | None = None,
) -> 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.

    The Schur terms need, per node, the two cross-block products
    ``Sigma[L, R] w_R`` and ``Sigma[R, L] w_L`` and one solve against each
    child's principal block, which a factor operator does by Woodbury in
    ``O(|C| k**2)`` rather than ``O(|C|**3)``. The block quadratic forms
    ``B D^{-1} B^T`` are never formed: only their action on the child weights is.

    Args:
        root (Cluster): The root node of the cluster tree
        cov (pl.DataFrame | CovarianceOperator): Covariance matrix of asset returns,
            as a DataFrame or as an operator
        gamma (float): Interpolation parameter in [0, 1]. 0 = HRP, 1 = minimum variance.
        assets (Sequence[str], optional): Asset names for an operator ``cov``;
            defaults to ``"0", "1", ...``. A DataFrame names its own assets.

    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: Block, right: Block, op: CovarianceOperator) -> tuple[float, float]:
        """Schur-augmented block variance of each child, conditioned on the other."""
        (li, w_left), (ri, w_right) = left, right

        v_left = _block_variance(left, op)
        v_right = _block_variance(right, op)
        if gamma == 0.0:
            return v_left, v_right

        # Schur-augmented blocks A - gamma B D^{-1} B^T and D - gamma B^T A^{-1} B
        # (A = Sigma_LL, B = Sigma_LR, D = Sigma_RR), applied to the child weights
        # through one cross product and one block solve each.
        bt_w = op.block_matvec(ri, li, w_left)
        b_w = op.block_matvec(li, ri, w_right)
        return (
            _schur_variance(v_left, gamma * float(bt_w @ solve_block(op, ri, bt_w))),
            _schur_variance(v_right, gamma * float(b_w @ solve_block(op, li, b_w))),
        )

    return _allocate_with(root, cov, node_variances, assets)

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