"""Nested Clusters Optimization estimator."""
# Copyright (c) 2023-2026
# Author: Hugo Delatte <hugo.delatte@skfoliolabs.com>
# SPDX-License-Identifier: BSD-3-Clause
# Implementation derived from:
# Riskfolio-Lib, Copyright (c) 2020-2023, Dany Cajas, Licensed under BSD 3 clause.
# scikit-learn, Copyright (c) 2007-2010 David Cournapeau, Fabian Pedregosa, Olivier
# Grisel Licensed under BSD 3 clause.
from __future__ import annotations
from copy import deepcopy
from typing import Any
import numpy as np
import pandas as pd
import sklearn as sk
import sklearn.base as skb
import sklearn.model_selection as sks
import sklearn.utils.metadata_routing as skm
import sklearn.utils.parallel as skp
import sklearn.utils.validation as skv
import skfolio.typing as skt
from skfolio.cluster import HierarchicalClustering
from skfolio.distance import BaseDistance, PearsonDistance
from skfolio.exceptions import OptimizationError
from skfolio.measures import RatioMeasure
from skfolio.model_selection import BaseCombinatorialCV, cross_val_predict
from skfolio.optimization._base import BaseOptimization, _check_finite_weights
from skfolio.optimization.convex import MeanRisk
from skfolio.typing import ArrayLike
from skfolio.utils.tools import check_estimator, fit_single_estimator
[docs]
class NestedClustersOptimization(BaseOptimization):
"""Nested Clusters Optimization estimator.
Nested Clusters Optimization (NCO) is a portfolio optimization method developed by
Marcos Lopez de Prado.
It uses a distance matrix to compute clusters using a clustering algorithm (
Hierarchical Tree Clustering, KMeans, etc.). For each cluster, the inner-cluster
weights are computed by fitting the inner-estimator on each cluster using the whole
training data. Then the outer-cluster weights are computed by training the
outer-estimator using out-of-sample estimates of the inner-estimators with
cross-validation. Finally, the final assets weights are the dot-product of the
inner-weights and outer-weights.
.. note ::
The original paper uses KMeans as the clustering algorithm, minimum Variance for
the inner-estimator and equal-weighted for the outer-estimator. Here we
generalize it to all `sklearn` and `skfolio` clustering algorithms
(HierarchicalClustering, KMeans, etc.), all portfolio optimizations
(Mean-Variance, HRP, etc.) and risk measures (Variance, CVaR, etc.).
To avoid data leakage at the outer-estimator, we use out-of-sample estimates to
fit the outer estimator.
Parameters
----------
inner_estimator : BaseOptimization, optional
:ref:`Optimization estimator <optimization>` used to estimate the inner-weights
(also called intra-weights) which are the assets weights inside each cluster.
The default `None` is to use :class:`~skfolio.optimization.MeanRisk`.
outer_estimator : BaseOptimization, optional
:ref:`Optimization estimator <optimization>` used to estimate the outer-weights
(also called inter-weights) which are the weights applied to each cluster.
The default `None` is to use :class:`~skfolio.optimization.MeanRisk`.
distance_estimator : BaseDistance, optional
:ref:`Distance estimator <distance>`.
The distance estimator is used to estimate the codependence and the distance
matrix needed for the computation of the linkage matrix.
The default (`None`) is to use :class:`~skfolio.distance.PearsonDistance`.
clustering_estimator : BaseEstimator, optional
Clustering estimator. Must expose a `labels_` attribute after fitting.
The clustering estimator is used to compute the clusters of the assets based on
the distance matrix. The default (`None`) is to use
:class:`~skfolio.cluster.HierarchicalClustering`.
.. note ::
Clustering estimators from `sklearn` are also supported. For example:
`sklearn.cluster.KMeans`.
cv : BaseCrossValidator | BaseCombinatorialCV | int | "ignore", optional
Determines the cross-validation splitting strategy.
The default (`None`) is to use the 5-fold cross validation `KFold()`.
It is applied to the inner-estimators. Its out-of-sample outputs are used to
train the outer-estimator.
Possible inputs for `cv` are:
* "ignore": no cross-validation is used (note that it will likely lead to data leakage with a high risk of overfitting)
* Integer, to specify the number of folds in a :class:`sklearn.model_selection.KFold`
* An object to be used as a cross-validation generator
* An iterable yielding train, test splits
* A :class:`~skfolio.model_selection.CombinatorialPurgedCV`
If a `CombinatorialCV` cross-validator is used, each cluster out-of-sample
outputs becomes a collection of multiple paths instead of one single path. The
selected out-of-sample path among this collection of paths is chosen according
to the `quantile` and `quantile_measure` parameters.
n_jobs : int, optional
The number of jobs to run in parallel for `fit` of all `estimators`.
The value `-1` means using all processors.
The default (`None`) means 1 unless in a `joblib.parallel_backend` context.
quantile : float, default=0.5
Quantile for a given measure (`quantile_measure`) of the out-of-sample
inner-estimator paths when the `cv` parameter is a
:class:`~skfolio.model_selection.CombinatorialPurgedCV` cross-validator.
The default value is `0.5` corresponding to the path with the median measure.
(see `cv`)
quantile_measure : PerfMeasure or RatioMeasure or RiskMeasure or ExtraRiskMeasure, default=RatioMeasure.SHARPE_RATIO
Measure used for the quantile path selection (see `quantile` and `cv`).
The default is `RatioMeasure.SHARPE_RATIO`.
verbose : int, default=0
The verbosity level. The default value is `0`.
portfolio_params : dict, optional
Portfolio parameters forwarded to the resulting `Portfolio` in `predict`.
If not provided and if available on the estimator, the following
attributes are propagated to the portfolio by default: `name` and
`previous_weights`.
fallback : BaseOptimization | "previous_weights" | list[BaseOptimization | "previous_weights"], optional
Fallback estimator or a list of estimators to try, in order, when the primary
optimization raises during `fit`. Alternatively, use `"previous_weights"` (alone
or in a list) to fall back to the estimator's `previous_weights`. When a
fallback succeeds, its fitted `weights_` are copied back to the primary
estimator so that `fit` still returns the original instance. For traceability,
`fallback_` stores the successful estimator (or the string `"previous_weights"`)
and `fallback_chain_` stores each attempt with the associated outcome. See
:ref:`optimization_fallbacks`.
previous_weights : float | dict[str, float] | array-like of shape (n_assets,), optional
When `fallback="previous_weights"`, failures will fall back to these weights
if provided.
raise_on_failure : bool, default=True
Controls error handling when fitting fails and no fallback succeeds. If True,
the estimator raises the final error. If False, the estimator emits a warning
and sets `weights_` to None, so subsequent calls to `predict` return a
:class:`~skfolio.portfolio.FailedPortfolio`. During `fit`, `raise_on_failure`
applies to any fitting error, including errors raised by the prior estimator.
See :ref:`optimization_failure_handling`.
Attributes
----------
weights_ : ndarray of shape (n_assets,)
Weights of the assets.
distance_estimator_ : BaseDistance
Fitted `distance_estimator`.
inner_estimators_ : list[BaseOptimization]
List of fitted `inner_estimator`. One per cluster for clusters containing more
than one asset.
outer_estimator_ : BaseOptimization
Fitted `outer_estimator`.
clustering_estimator_ : BaseEstimator
Fitted `clustering_estimator`.
n_features_in_ : int
Number of assets seen during `fit`.
feature_names_in_ : ndarray of shape (`n_features_in_`,)
Names of assets seen during `fit`. Defined only when `X`
has assets names that are all strings.
fallback_ : BaseOptimization | "previous_weights" | None
The fallback estimator instance, or the string `"previous_weights"`, that
produced the final result. `None` if no fallback was used.
fallback_chain_ : list[tuple[str, str]] | None
Sequence describing the optimization fallback attempts. Each element is a
pair `(estimator_repr, outcome)` where `estimator_repr` is the string
representation of the primary estimator or a fallback (e.g. `"EqualWeighted()"`,
`"previous_weights"`), and `outcome` is `"success"` if that step produced
a valid solution, otherwise the stringified error message. For successful
fits without any fallback, this is `None`.
error_ : str | list[str | None] | None
For a single portfolio, this is the recorded error message, or None after a
successful allocation or fallback. For multiple portfolios, it is a list with
one entry per row of `weights_`, containing an error message for each failed
portfolio and None for each successful portfolio.
Notes
-----
All estimators should specify all parameters as explicit keyword arguments in
`__init__` (no `*args` or `**kwargs`), following scikit-learn conventions.
Examples
--------
For complete tutorials on nested clusters optimization, see the
:ref:`cluster_examples` gallery.
>>> from skfolio import RiskMeasure
>>> from skfolio.datasets import load_sp500_dataset
>>> from skfolio.optimization import (
... MeanRisk,
... NestedClustersOptimization,
... ObjectiveFunction,
... RiskBudgeting,
... )
>>> from skfolio.preprocessing import prices_to_returns
>>>
>>> # Load historical prices and convert them to returns
>>> prices = load_sp500_dataset()
>>> X = prices_to_returns(prices)
>>>
>>> # Maximize the Sharpe ratio within each cluster and minimize the CVaR
>>> # computed from the out-of-sample predicted returns of the clusters
>>> inner_estimator = MeanRisk(
... objective_function=ObjectiveFunction.MAXIMIZE_RATIO,
... risk_measure=RiskMeasure.STANDARD_DEVIATION,
... )
>>> outer_estimator = RiskBudgeting(risk_measure=RiskMeasure.CVAR)
>>> model = NestedClustersOptimization(
... inner_estimator=inner_estimator,
... outer_estimator=outer_estimator,
... )
>>> model.fit(X)
NestedClustersOptimization(...)
>>> print(model.weights_)
[0.0327 0.0029 0. ... 0.0864 0.0236 0.0736]
>>> print(model.clustering_estimator_.labels_)
[4 4 3 4 2 3 4 1 3 0 1 1 4 0 1 0 2 1 4 2]
References
----------
.. [1] "Building diversified portfolios that outperform out of sample",
The Journal of Portfolio Management,
Marcos López de Prado (2016)
.. [2] "A robust estimator of the efficient frontier",
SSRN Electronic Journal,
Marcos López de Prado (2019)
.. [3] "Machine Learning for Asset Managers",
Elements in Quantitative Finance. Cambridge University Press,
Marcos López de Prado (2020)
"""
inner_estimators_: list[BaseOptimization]
outer_estimator_: BaseOptimization
distance_estimator_: BaseDistance
clustering_estimator_: skb.BaseEstimator
def __init__(
self,
inner_estimator: BaseOptimization | None = None,
outer_estimator: BaseOptimization | None = None,
distance_estimator: BaseDistance | None = None,
clustering_estimator: skb.BaseEstimator | None = None,
cv: sks.BaseCrossValidator | BaseCombinatorialCV | str | int | None = None,
quantile: float = 0.5,
quantile_measure: skt.Measure = RatioMeasure.SHARPE_RATIO,
n_jobs: int | None = None,
verbose: int = 0,
portfolio_params: dict | None = None,
fallback: skt.Fallback = None,
previous_weights: skt.MultiInput | None = None,
raise_on_failure: bool = True,
) -> None:
super().__init__(
portfolio_params=portfolio_params,
fallback=fallback,
previous_weights=previous_weights,
raise_on_failure=raise_on_failure,
)
self.distance_estimator = distance_estimator
self.clustering_estimator = clustering_estimator
self.inner_estimator = inner_estimator
self.outer_estimator = outer_estimator
self.cv = cv
self.quantile = quantile
self.quantile_measure = quantile_measure
self.n_jobs = n_jobs
self.verbose = verbose
[docs]
def fit(
self, X: ArrayLike, y: ArrayLike | None = None, **fit_params: Any
) -> NestedClustersOptimization:
"""Fit the Nested Clusters Optimization estimator.
Parameters
----------
X : array-like of shape (n_observations, n_assets)
Price returns of the assets.
y : array-like of shape (n_observations, n_targets), optional
Price returns of factors or a target benchmark.
The default is `None`.
**fit_params : dict
Parameters to pass to the underlying estimators.
Only available if `enable_metadata_routing=True`, which can be
set by using `sklearn.set_config(enable_metadata_routing=True)`.
See :ref:`Metadata Routing User Guide <metadata_routing>` for
more details.
Returns
-------
self : NestedClustersOptimization
Fitted estimator.
"""
routed_params = skm.process_routing(self, "fit", **fit_params)
self.distance_estimator_ = check_estimator(
self.distance_estimator,
default=PearsonDistance(),
check_type=BaseDistance,
)
self.clustering_estimator_ = check_estimator(
self.clustering_estimator,
default=HierarchicalClustering(),
check_type=skb.BaseEstimator,
)
self.outer_estimator_ = check_estimator(
self.outer_estimator,
default=MeanRisk(),
check_type=BaseOptimization,
)
_inner_estimator = check_estimator(
self.inner_estimator,
default=MeanRisk(),
check_type=BaseOptimization,
)
self.distance_estimator_.fit(X, y, **routed_params.distance_estimator.fit)
distance = self.distance_estimator_.distance_
n_assets = distance.shape[0]
# To keep the asset_names --> used for visualisation
if isinstance(X, pd.DataFrame):
distance = pd.DataFrame(distance, columns=X.columns)
self.clustering_estimator_.fit(
X=distance, y=None, **routed_params.clustering_estimator.fit
)
labels = self.clustering_estimator_.labels_
n_clusters = max(labels) + 1
clusters = [np.argwhere(labels == i).flatten() for i in range(n_clusters)]
# Intra cluster weights
# Fit the inner estimator on the whole training data. Those
# base estimators will be used to retrieve the inner weights.
# They are exposed publicly.
fitted_inner_estimators = skp.Parallel(n_jobs=self.n_jobs)(
skp.delayed(fit_single_estimator)(
sk.clone(_inner_estimator),
X,
y,
routed_params.inner_estimator.fit,
indices=cluster_ids,
axis=1,
)
for cluster_ids in clusters
if len(cluster_ids) != 1
)
fitted_inner_estimators = iter(fitted_inner_estimators)
self.inner_estimators_ = []
inner_weights = []
for cluster_ids in clusters:
w = np.zeros(n_assets)
# For single assets, we don't run the inner optimization estimator.
if len(cluster_ids) == 1:
w[cluster_ids] = 1
else:
fitted_inner_estimator = next(fitted_inner_estimators)
self.inner_estimators_.append(fitted_inner_estimator)
if fitted_inner_estimator.weights_ is None:
raise OptimizationError(
"An inner estimator returned no allocation weights."
)
w[cluster_ids] = fitted_inner_estimator.weights_
inner_weights.append(w)
inner_weights = np.array(inner_weights)
assert not any(fitted_inner_estimators), (
"fitted_inner_estimator iterator must be empty"
)
# Outer cluster weights
# To train the outer-estimator using the most data as possible, we use
# a cross-validation to obtain the output of the cluster estimators.
# To ensure that the data provided to each estimator are the same,
# we need to set the random state of the cv if there is one and we
# need to take a copy.
if self.cv == "ignore":
cv_predictions = None
test_indices = slice(None)
else:
cv = sks.check_cv(self.cv)
if hasattr(cv, "random_state") and cv.random_state is None:
cv.random_state = np.random.RandomState()
cv_predictions = skp.Parallel(n_jobs=self.n_jobs)(
skp.delayed(cross_val_predict)(
sk.clone(_inner_estimator),
X,
y,
cv=deepcopy(cv),
n_jobs=self.n_jobs,
verbose=self.verbose,
column_indices=cluster_ids,
method="predict",
params=routed_params.inner_estimator.fit,
)
for cluster_ids in clusters
if len(cluster_ids) != 1
)
cv_predictions = iter(cv_predictions)
if isinstance(self.cv, BaseCombinatorialCV):
test_indices = slice(None)
else:
test_indices = np.sort(
np.concatenate([test for _, test in cv.split(X, y)])
)
# We validate and convert to numpy array only after inner-estimator fitting to
# keep the assets names in case they are used in the estimator.
if y is not None:
X, y = skv.validate_data(self, X, y)
y_pred = y[test_indices]
else:
X = skv.validate_data(self, X)
y_pred = None
X_pred = []
fitted_inner_estimators = iter(self.inner_estimators_)
for cluster_ids in clusters:
if len(cluster_ids) == 1:
pred = X[test_indices, cluster_ids[0]]
else:
if cv_predictions is None:
fitted_inner_estimator = next(fitted_inner_estimators)
pred = fitted_inner_estimator.predict(X[test_indices, cluster_ids])
else:
pred = next(cv_predictions)
if isinstance(self.cv, BaseCombinatorialCV):
pred = pred.quantile(
measure=self.quantile_measure, q=self.quantile
)
X_pred.append(np.asarray(pred))
X_pred = np.array(X_pred).T
if cv_predictions is None:
assert not any(fitted_inner_estimators), (
"fitted_inner_estimator iterator must be empty"
)
else:
assert not any(cv_predictions), "cv_predictions iterator must be empty"
fit_single_estimator(self.outer_estimator_, X_pred, y_pred, fit_params={})
outer_weights = self.outer_estimator_.weights_
if outer_weights is None:
raise OptimizationError(
"The outer estimator returned no allocation weights."
)
weights = outer_weights @ inner_weights
_check_finite_weights(weights)
self.weights_ = weights
return self
