diff --git a/pypfopt/efficient_frontier/efficient_frontier.py b/pypfopt/efficient_frontier/efficient_frontier.py index 04e9d627..47628488 100644 --- a/pypfopt/efficient_frontier/efficient_frontier.py +++ b/pypfopt/efficient_frontier/efficient_frontier.py @@ -3,6 +3,7 @@ classical mean-variance optimal portfolios for a variety of objectives and constraints """ +from typing import Any import warnings import cvxpy as cp @@ -56,13 +57,13 @@ class EfficientFrontier(BaseConvexOptimizer): def __init__( self, - expected_returns, - cov_matrix, - weight_bounds=(0, 1), - solver=None, - verbose=False, - solver_options=None, - ): + expected_returns: pd.Series | list | np.ndarray | None, + cov_matrix: pd.DataFrame | np.ndarray, + weight_bounds: tuple | list = (0, 1), + solver: str | None = None, + verbose: bool = False, + solver_options: dict[str, Any] | None = None, + ) -> None: """ Parameters ---------- @@ -89,30 +90,20 @@ def __init__( if ``expected_returns`` is not a series, list or array TypeError if ``cov_matrix`` is not a dataframe or array + + Notes + ----- + Asset labels are aligned automatically only when ``expected_returns`` is a + Series and ``cov_matrix`` is a DataFrame. When pandas and array-like inputs + are mixed, their values are matched positionally, so they must use the same + asset order. """ - # Inputs - self.cov_matrix = self._validate_cov_matrix(cov_matrix) - self.expected_returns = self._validate_expected_returns(expected_returns) + self.expected_returns, self.cov_matrix, tickers = ( + self._validate_and_format_inputs(expected_returns, cov_matrix) + ) self._max_return_value = None self._market_neutral = None - if self.expected_returns is None: - num_assets = len(cov_matrix) - else: - num_assets = len(expected_returns) - - # Labels - if isinstance(expected_returns, pd.Series): - tickers = list(expected_returns.index) - elif isinstance(cov_matrix, pd.DataFrame): - tickers = list(cov_matrix.columns) - else: # use integer labels - tickers = list(range(num_assets)) - - if expected_returns is not None and cov_matrix is not None: - if cov_matrix.shape != (num_assets, num_assets): - raise ValueError("Covariance matrix does not match expected returns") - super().__init__( len(tickers), tickers, @@ -123,28 +114,84 @@ def __init__( ) @staticmethod - def _validate_expected_returns(expected_returns): - if expected_returns is None: - return None - elif isinstance(expected_returns, pd.Series): - return expected_returns.values - elif isinstance(expected_returns, list): - return np.array(expected_returns) - elif isinstance(expected_returns, np.ndarray): - return expected_returns.ravel() + def _validate_and_format_inputs( + expected_returns: pd.Series | list | np.ndarray | None, + cov_matrix: pd.DataFrame | np.ndarray | None, + ) -> tuple[np.ndarray | None, np.ndarray, list[Any]]: + """Validate inputs and align labeled data to a common asset order.""" + cov_array, cov_tickers = EfficientFrontier._validate_and_format_cov_matrix( + cov_matrix + ) + expected_array, expected_tickers = ( + EfficientFrontier._validate_and_format_expected_returns(expected_returns) + ) + + if expected_array is not None and cov_array.shape[0] != len(expected_array): + raise ValueError("Covariance matrix does not match expected returns") + + # Labeled pandas inputs can be aligned safely. Mixed pandas/array inputs + # remain positional because their alignment cannot be verified. + if expected_tickers is not None and cov_tickers is not None: + labels_match = ( + expected_tickers.is_unique + and expected_tickers.isin(cov_tickers).all() + ) + if not labels_match: + raise ValueError( + "Covariance matrix labels do not match expected returns" + ) + # Reorder both covariance axes to match the expected-return order. + positions = cov_tickers.get_indexer(expected_tickers) + cov_array = cov_array[np.ix_(positions, positions)] + + if expected_tickers is not None: + tickers = list(expected_tickers) + elif cov_tickers is not None: + tickers = list(cov_tickers) else: - raise TypeError("expected_returns is not a series, list or array") + tickers = list(range(cov_array.shape[0])) + + return expected_array, cov_array, tickers @staticmethod - def _validate_cov_matrix(cov_matrix): + def _validate_and_format_cov_matrix( + cov_matrix: pd.DataFrame | np.ndarray | None, + ) -> tuple[np.ndarray, pd.Index | None]: + """Validate a covariance matrix and normalize DataFrame row order.""" if cov_matrix is None: raise ValueError("cov_matrix must be provided") - elif isinstance(cov_matrix, pd.DataFrame): - return cov_matrix.values - elif isinstance(cov_matrix, np.ndarray): - return cov_matrix - else: + if not isinstance(cov_matrix, (pd.DataFrame, np.ndarray)): raise TypeError("cov_matrix is not a dataframe or array") + if cov_matrix.ndim != 2 or cov_matrix.shape[0] != cov_matrix.shape[1]: + raise ValueError("cov_matrix must be a square matrix") + + if isinstance(cov_matrix, np.ndarray): + return cov_matrix, None + if not cov_matrix.index.is_unique or not cov_matrix.columns.is_unique: + raise ValueError("Covariance matrix labels must be unique") + if not cov_matrix.index.isin(cov_matrix.columns).all(): + raise ValueError( + "Covariance matrix index and columns must contain the same labels" + ) + + cov_tickers = cov_matrix.columns + cov_array = cov_matrix.loc[cov_tickers, cov_tickers].values + return cov_array, cov_tickers + + @staticmethod + def _validate_and_format_expected_returns( + expected_returns: pd.Series | list | np.ndarray | None, + ) -> tuple[np.ndarray | None, pd.Index | None]: + """Convert expected returns to a flat array and preserve Series labels.""" + if expected_returns is None: + return None, None + if isinstance(expected_returns, pd.Series): + return expected_returns.values, expected_returns.index + if isinstance(expected_returns, list): + return np.array(expected_returns), None + if isinstance(expected_returns, np.ndarray): + return expected_returns.ravel(), None + raise TypeError("expected_returns is not a series, list or array") def _validate_returns(self, returns): """ @@ -217,7 +264,7 @@ def _max_return(self, return_value=True): Returns ------- OrderedDict - asset weights for the return-minimising portfolio + asset weights for the return-maximising portfolio """ if self.expected_returns is None: raise ValueError("no expected returns provided") @@ -232,16 +279,17 @@ def _max_return(self, return_value=True): if return_value: return -self._opt.value - else: - return res + return res def max_sharpe(self, risk_free_rate=0.0): """ - Maximise the Sharpe Ratio. The result is also referred to as the tangency portfolio, - as it is the portfolio for which the capital market line is tangent to the efficient frontier. + Maximise the Sharpe Ratio. The result is also referred to as the tangency + portfolio, as it is the portfolio for which the capital market line is tangent + to the efficient frontier. - This is a convex optimization problem after making a certain variable substitution. See - `Cornuejols and Tutuncu (2006) `_ for more. + This is a convex optimization problem after making a certain variable + substitution. See `Cornuejols and Tutuncu (2006) + `_ for more. Parameters ---------- @@ -265,7 +313,8 @@ def max_sharpe(self, risk_free_rate=0.0): if max(self.expected_returns) <= risk_free_rate: raise ValueError( - "at least one of the assets must have an expected return exceeding the risk-free rate" + "at least one of the assets must have an expected return exceeding " + "the risk-free rate" ) self._risk_free_rate = risk_free_rate @@ -277,9 +326,10 @@ def max_sharpe(self, risk_free_rate=0.0): # Note: objectives are not scaled by k. Hence there are subtle differences # between how these objectives work for max_sharpe vs min_volatility - if len(self._additional_objectives) > 0: + if self._additional_objectives: warnings.warn( - "max_sharpe transforms the optimization problem so additional objectives may not work as expected." + "max_sharpe transforms the optimization problem so additional " + "objectives may not work as expected." ) for obj in self._additional_objectives: self._objective += obj @@ -375,8 +425,6 @@ def efficient_risk(self, target_volatility, market_neutral=False): if ``target_volatility`` is not a positive float ValueError if no portfolio can be found with volatility equal to ``target_volatility`` - ValueError - if ``risk_free_rate`` is non-numeric Returns ------- @@ -390,9 +438,8 @@ def efficient_risk(self, target_volatility, market_neutral=False): if target_volatility < global_min_volatility: raise ValueError( - "The minimum volatility is {:.3f}. Please use a higher target_volatility".format( - global_min_volatility - ) + f"The minimum volatility is {global_min_volatility:.3f}. " + "Please use a higher target_volatility" ) update_existing_parameter = self.is_parameter_defined("target_variance") @@ -441,9 +488,9 @@ def efficient_return(self, target_return, market_neutral=False): """ if not isinstance(target_return, float): raise ValueError("target_return should be a float") - if not self._max_return_value: + if self._max_return_value is None: a = self.deepcopy() - self._max_return_value = a._max_return() + self._max_return_value = a._max_return() # pylint: disable=protected-access if target_return > self._max_return_value: raise ValueError( "target_return must be lower than the maximum possible return" diff --git a/tests/test_efficient_frontier.py b/tests/test_efficient_frontier.py index 038788e0..67efca84 100644 --- a/tests/test_efficient_frontier.py +++ b/tests/test_efficient_frontier.py @@ -82,6 +82,205 @@ def test_efficient_frontier_expected_returns_list(): ) +class TestValidateAndFormatExpectedReturns: + # Direct access is intentional: these tests define the private helper's contract. + # pylint: disable=protected-access + + def test_accepts_none(self): + formatted_returns, tickers = ( + EfficientFrontier._validate_and_format_expected_returns(None) + ) + + assert formatted_returns is None + assert tickers is None + + def test_converts_list_to_array(self): + formatted_returns, tickers = ( + EfficientFrontier._validate_and_format_expected_returns([0.1, 0.2]) + ) + + np.testing.assert_array_equal(formatted_returns, np.array([0.1, 0.2])) + assert tickers is None + + def test_flattens_numpy_array(self): + formatted_returns, tickers = ( + EfficientFrontier._validate_and_format_expected_returns( + np.array([[0.1, 0.2]]) + ) + ) + + np.testing.assert_array_equal(formatted_returns, np.array([0.1, 0.2])) + assert tickers is None + + def test_extracts_series_values_and_tickers(self): + returns_series = pd.Series([0.1, 0.2], index=["A", "B"]) + + formatted_returns, tickers = ( + EfficientFrontier._validate_and_format_expected_returns(returns_series) + ) + + np.testing.assert_array_equal(formatted_returns, np.array([0.1, 0.2])) + pd.testing.assert_index_equal(tickers, pd.Index(["A", "B"])) + + def test_rejects_unsupported_type(self): + with pytest.raises( + TypeError, match="expected_returns is not a series, list or array" + ): + EfficientFrontier._validate_and_format_expected_returns(0.02) + + +class TestValidateAndFormatCovMatrix: + # Direct access is intentional: these tests define the private helper's contract. + # pylint: disable=protected-access + + def test_accepts_numpy_array_without_tickers(self): + cov_matrix = np.array([[0.1, 0.2], [0.3, 0.4]]) + + formatted_matrix, tickers = ( + EfficientFrontier._validate_and_format_cov_matrix(cov_matrix) + ) + + assert formatted_matrix is cov_matrix + assert tickers is None + + def test_aligns_dataframe_rows_to_column_order(self): + cov_matrix = pd.DataFrame( + [[0.3, 0.4], [0.1, 0.2]], + index=["B", "A"], + columns=["A", "B"], + ) + + formatted_matrix, tickers = ( + EfficientFrontier._validate_and_format_cov_matrix(cov_matrix) + ) + + np.testing.assert_array_equal( + formatted_matrix, np.array([[0.1, 0.2], [0.3, 0.4]]) + ) + pd.testing.assert_index_equal(tickers, pd.Index(["A", "B"])) + + def test_rejects_none(self): + with pytest.raises(ValueError, match="cov_matrix must be provided"): + EfficientFrontier._validate_and_format_cov_matrix(None) + + def test_rejects_unsupported_type(self): + with pytest.raises( + TypeError, match="cov_matrix is not a dataframe or array" + ): + EfficientFrontier._validate_and_format_cov_matrix(0.01) + + def test_rejects_one_dimensional_array(self): + with pytest.raises(ValueError, match="cov_matrix must be a square matrix"): + EfficientFrontier._validate_and_format_cov_matrix(np.ones(2)) + + @pytest.mark.parametrize( + "cov_matrix", + [np.ones((2, 3)), pd.DataFrame(np.ones((2, 3)))], + ) + def test_rejects_non_square_matrix(self, cov_matrix): + with pytest.raises(ValueError, match="cov_matrix must be a square matrix"): + EfficientFrontier._validate_and_format_cov_matrix(cov_matrix) + + @pytest.mark.parametrize( + "index, columns", + [ + (["A", "A"], ["A", "B"]), + (["A", "B"], ["A", "A"]), + ], + ) + def test_rejects_duplicate_dataframe_labels(self, index, columns): + cov_matrix = pd.DataFrame(np.eye(2), index=index, columns=columns) + + with pytest.raises(ValueError, match="Covariance matrix labels must be unique"): + EfficientFrontier._validate_and_format_cov_matrix(cov_matrix) + + def test_rejects_different_dataframe_axis_labels(self): + cov_matrix = pd.DataFrame( + np.eye(2), index=["A", "B"], columns=["A", "C"] + ) + + with pytest.raises(ValueError, match="must contain the same labels"): + EfficientFrontier._validate_and_format_cov_matrix(cov_matrix) + + +class TestValidateAndFormatInputs: + # Direct access is intentional: these tests define the private helper's contract. + # pylint: disable=protected-access + + def test_aligns_covariance_to_expected_return_order(self): + expected_returns = pd.Series([0.2, 0.1], index=["B", "A"]) + cov_matrix = pd.DataFrame( + [[0.1, 0.2], [0.3, 0.4]], + index=["A", "B"], + columns=["A", "B"], + ) + + formatted_returns, formatted_covariance, tickers = ( + EfficientFrontier._validate_and_format_inputs( + expected_returns, cov_matrix + ) + ) + + np.testing.assert_array_equal(formatted_returns, np.array([0.2, 0.1])) + np.testing.assert_array_equal( + formatted_covariance, np.array([[0.4, 0.3], [0.2, 0.1]]) + ) + assert tickers == ["B", "A"] + + def test_uses_expected_return_tickers_with_array_covariance(self): + expected_returns = pd.Series([0.1, 0.2], index=["A", "B"]) + + _, _, tickers = EfficientFrontier._validate_and_format_inputs( + expected_returns, np.eye(2) + ) + + assert tickers == ["A", "B"] + + def test_uses_covariance_tickers_without_expected_returns(self): + cov_matrix = pd.DataFrame( + np.eye(2), index=["A", "B"], columns=["A", "B"] + ) + + _, _, tickers = EfficientFrontier._validate_and_format_inputs(None, cov_matrix) + + assert tickers == ["A", "B"] + + def test_uses_positional_tickers_for_array_inputs(self): + _, _, tickers = EfficientFrontier._validate_and_format_inputs( + np.array([0.1, 0.2]), np.eye(2) + ) + + assert tickers == [0, 1] + + def test_rejects_dimension_mismatch(self): + with pytest.raises(ValueError, match="does not match expected returns"): + EfficientFrontier._validate_and_format_inputs( + np.array([0.1, 0.2, 0.3]), np.eye(2) + ) + + def test_rejects_different_labels(self): + expected_returns = pd.Series([0.1, 0.2], index=["A", "C"]) + cov_matrix = pd.DataFrame( + np.eye(2), index=["A", "B"], columns=["A", "B"] + ) + + with pytest.raises(ValueError, match="labels do not match expected returns"): + EfficientFrontier._validate_and_format_inputs( + expected_returns, cov_matrix + ) + + def test_rejects_duplicate_expected_return_labels(self): + expected_returns = pd.Series([0.1, 0.2], index=["A", "A"]) + cov_matrix = pd.DataFrame( + np.eye(2), index=["A", "B"], columns=["A", "B"] + ) + + with pytest.raises(ValueError, match="labels do not match expected returns"): + EfficientFrontier._validate_and_format_inputs( + expected_returns, cov_matrix + ) + + def test_portfolio_performance(): ef = setup_efficient_frontier() with pytest.raises(ValueError): @@ -107,6 +306,53 @@ def test_min_volatility(): ) +def test_min_volatility_aligns_covariance_labels(): + mean_returns = pd.Series({"A": 0.1, "B": 0.2}) + cov_matrix = pd.DataFrame( + [[0.01, 0.0], [0.0, 1.0]], + index=["A", "B"], + columns=["A", "B"], + ) + + expected = EfficientFrontier(mean_returns, cov_matrix).min_volatility() + reordered_cov = cov_matrix.loc[["B", "A"], ["B", "A"]] + actual = EfficientFrontier(mean_returns, reordered_cov).min_volatility() + + pd.testing.assert_series_equal( + pd.Series(actual), pd.Series(expected), check_names=False + ) + + +def test_covariance_dataframe_aligns_index_to_columns(): + cov_matrix = pd.DataFrame( + [[0.01, 0.002], [0.002, 1.0]], + index=["A", "B"], + columns=["A", "B"], + ) + reordered_index = cov_matrix.loc[["B", "A"]] + + ef = EfficientFrontier(None, reordered_index) + + assert ef.tickers == ["A", "B"] + np.testing.assert_array_equal(ef.cov_matrix, cov_matrix.values) + + +@pytest.mark.parametrize( + "mean_returns", + [pd.Series({"A": 0.1, "B": 0.2, "C": 0.3}), np.array([0.1, 0.2, 0.3])], +) +@pytest.mark.parametrize( + "cov_matrix", + [ + pd.DataFrame(np.eye(2), index=["A", "B"], columns=["A", "B"]), + np.eye(2), + ], +) +def test_covariance_matrix_must_match_expected_returns(mean_returns, cov_matrix): + with pytest.raises(ValueError, match="Covariance matrix does not match"): + EfficientFrontier(mean_returns, cov_matrix) + + @pytest.mark.skipif( not _check_soft_dependencies(["ecos"], severity="none"), reason="skip test if ecos is not installed in environment", @@ -1085,16 +1331,16 @@ def test_efficient_return_error(): ef.efficient_return(max_ret + 0.01) -def test_efficient_frontier_error(): +def test_efficient_return_reuses_zero_max_return(monkeypatch): ef = setup_efficient_frontier() - with pytest.raises(ValueError): - EfficientFrontier(ef.expected_returns[:-1], ef.cov_matrix) - with pytest.raises(TypeError): - EfficientFrontier(0.02, ef.cov_matrix) - with pytest.raises(ValueError): - EfficientFrontier(ef.expected_returns, None) - with pytest.raises(TypeError): - EfficientFrontier(ef.expected_returns, 0.01) + monkeypatch.setattr(ef, "_max_return_value", 0.0) + monkeypatch.setattr( + ef, + "_max_return", + lambda: pytest.fail("cached maximum return was recomputed"), + ) + + ef.efficient_return(0.0) @pytest.mark.skipif(