alihan/PyPortfolioOpt
1
0
Fork 0
mirror of https://github.com/robertmartin8/PyPortfolioOpt.git synced 2022-11-27 18:02:41 +03:00
Code Issues Packages Projects Releases Wiki Activity

migrated max_sharpe to cvxpy

Browse Source
This commit is contained in:
robertmartin8
2020-03-15 12:59:01 +00:00
parent a6b8df3bcb
commit 8b337fa3af
6 changed files with 390 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
pypfopt/base_optimizer.py
View File

@@ -128,7 +128,10 @@ class BaseConvexOptimizer(BaseOptimizer):
self._w = cp.Variable(n_assets)
self._objective = None
self._additional_objectives = []
self._additional_constraints_raw = []
self._constraints = []
self._lower_bounds = None
self._upper_bounds = None
self._map_bounds_to_constraints(weight_bounds)
def _map_bounds_to_constraints(self, test_bounds):
@@ -147,8 +150,8 @@ class BaseConvexOptimizer(BaseOptimizer):
test_bounds[0], (float, int)
):
bounds = np.array(test_bounds, dtype=np.float)
lower = np.nan_to_num(bounds[:, 0], nan=-np.inf)
upper = np.nan_to_num(bounds[:, 1], nan=np.inf)
self._lower_bounds = np.nan_to_num(bounds[:, 0], nan=-np.inf)
self._upper_bounds = np.nan_to_num(bounds[:, 1], nan=np.inf)
else:
# Otherwise this must be a pair.
if len(test_bounds) != 2 or not isinstance(test_bounds, (tuple, list)):
@@ -161,13 +164,15 @@ class BaseConvexOptimizer(BaseOptimizer):
# Replace None values with the appropriate infinity.
if np.isscalar(lower) or lower is None:
lower = -np.inf if lower is None else lower
self._lower_bounds = np.array([lower] * self.n_assets)
upper = np.inf if upper is None else upper
self._upper_bounds = np.array([upper] * self.n_assets)
else:
lower = np.nan_to_num(lower, nan=-np.inf)
upper = np.nan_to_num(upper, nan=np.inf)
self._lower_bounds = np.nan_to_num(lower, nan=-np.inf)
self._upper_bounds = np.nan_to_num(upper, nan=np.inf)
self._constraints.append(self._w >= lower)
self._constraints.append(self._w <= upper)
self._constraints.append(self._w >= self._lower_bounds)
self._constraints.append(self._w <= self._upper_bounds)
@staticmethod
def _make_scipy_bounds():
@@ -178,7 +183,7 @@ class BaseConvexOptimizer(BaseOptimizer):
def portfolio_performance(
expected_returns, cov_matrix, weights, verbose=False, risk_free_rate=0.02
weights, expected_returns, cov_matrix, verbose=False, risk_free_rate=0.02
):
"""
After optimising, calculate (and optionally print) the performance of the optimal
@@ -186,9 +191,9 @@ def portfolio_performance(
:param expected_returns: expected returns for each asset. Set to None if
optimising for volatility only.
:type expected_returns: pd.Series, list, np.ndarray
:type expected_returns: np.ndarray or pd.Series
:param cov_matrix: covariance of returns for each asset
:type cov_matrix: pd.DataFrame or np.array
:type cov_matrix: np.array or pd.DataFrame
:param weights: weights or assets
:type weights: list, np.array or dict, optional
:param verbose: whether performance should be printed, defaults to False
@@ -218,11 +223,23 @@ def portfolio_performance(
raise ValueError("Weights is None")
sigma = np.sqrt(objective_functions.portfolio_variance(new_weights, cov_matrix))
mu = new_weights.dot(expected_returns)
sharpe = -objective_functions.negative_sharpe(
new_weights, expected_returns, cov_matrix, risk_free_rate=risk_free_rate
mu = objective_functions.portfolio_return(
new_weights, expected_returns, negative=False
)
# new_weights.dot(expected_returns)
# sharpe = -objective_functions.negative_sharpe(
# new_weights, expected_returns, cov_matrix, risk_free_rate=risk_free_rate
# )
sharpe = objective_functions.sharpe_ratio(
new_weights,
expected_returns,
cov_matrix,
risk_free_rate=risk_free_rate,
negative=False,
)
if verbose:
print("Expected annual return: {:.1f}%".format(100 * mu))
print("Annual volatility: {:.1f}%".format(100 * sigma))

138
pypfopt/efficient_frontier.py
View File

@@ -111,6 +111,22 @@ class EfficientFrontier(base_optimizer.BaseConvexOptimizer):
else:
raise TypeError("cov_matrix is not a series, list or array")
def _solve_cvxpy_opt_problem(self):
"""
Helper method to solve the cvxpy problem and check output,
once objectives and constraints have been defined
:raises exceptions.OptimizationError: if problem is not solvable by cvxpy
"""
try:
opt = cp.Problem(cp.Minimize(self._objective), self._constraints)
except TypeError:
raise exceptions.OptimizationError
opt.solve()
if opt.status != "optimal":
raise exceptions.OptimizationError
self.weights = self._w.value.round(16) + 0.0 # +0.0 removes signed zero
def add_objective(self, new_objective, **kwargs):
"""
Add a new term into the objective function. This term must be convex,
@@ -144,16 +160,38 @@ class EfficientFrontier(base_optimizer.BaseConvexOptimizer):
"""
if not callable(new_constraint):
raise TypeError("New constraint must be provided as a lambda function")
# Save raw constraint (needed for e.g max_sharpe)
self._additional_constraints_raw.append(new_constraint)
# Add constraint
self._constraints.append(new_constraint(self._w))
def convex_optimize(custom_objective, constraints):
def convex_optimize(self, custom_objective, constraints):
# TODO: fix
# genera convex optimistion
pass
def nonconvex_optimize(custom_objective, constraints):
# opt using scip
# args = (self.cov_matrix, self.gamma)
def nonconvex_optimize(self, custom_objective=None, constraints=None):
#  TODO: fix
# opt using scipy
args = (self.cov_matrix,)
initial_guess = np.array([1 / self.n_assets] * self.n_assets)
result = sco.minimize(
objective_functions.volatility,
x0=initial_guess,
args=args,
method="SLSQP",
bounds=[(0, 1)] * 20,
constraints=[{"type": "eq", "fun": lambda x: np.sum(x) - 1}],
)
self.weights = result["x"]
#  max sharpe
# args = (self.expected_returns, self.cov_matrix, self.gamma, risk_free_rate)
# result = sco.minimize(
# objective_functions.volatility,
# objective_functions.negative_sharpe,
# x0=self.initial_guess,
# args=args,
# method=self.opt_method,
@@ -161,34 +199,7 @@ class EfficientFrontier(base_optimizer.BaseConvexOptimizer):
# constraints=self.constraints,
# )
# self.weights = result["x"]
pass
def max_sharpe(self, risk_free_rate=0.02):
"""
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.
:param risk_free_rate: risk-free rate of borrowing/lending, defaults to 0.02.
The period of the risk-free rate should correspond to the
frequency of expected returns.
:type risk_free_rate: float, optional
:raises ValueError: if ``risk_free_rate`` is non-numeric
:return: asset weights for the Sharpe-maximising portfolio
:rtype: dict
"""
if not isinstance(risk_free_rate, (int, float)):
raise ValueError("risk_free_rate should be numeric")
args = (self.expected_returns, self.cov_matrix, self.gamma, risk_free_rate)
result = sco.minimize(
objective_functions.negative_sharpe,
x0=self.initial_guess,
args=args,
method=self.opt_method,
bounds=self.bounds,
constraints=self.constraints,
)
self.weights = result["x"]
return dict(zip(self.tickers, self.weights))
def min_volatility(self):
@@ -206,15 +217,59 @@ class EfficientFrontier(base_optimizer.BaseConvexOptimizer):
self._constraints.append(cp.sum(self._w) == 1)
try:
opt = cp.Problem(cp.Minimize(self._objective), self._constraints)
except TypeError:
raise exceptions.OptimizationError
self._solve_cvxpy_opt_problem()
return dict(zip(self.tickers, self.weights))
opt.solve()
if opt.status != "optimal":
raise exceptions.OptimizationError
self.weights = self._w.value.round(20)
def max_sharpe(self, risk_free_rate=0.02):
"""
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 optimisation problem after making a certain variable substitution. See
`Cornuejols and Tutuncu 2006 <http://web.math.ku.dk/~rolf/CT_FinOpt.pdf>`_ for more.
:param risk_free_rate: risk-free rate of borrowing/lending, defaults to 0.02.
The period of the risk-free rate should correspond to the
frequency of expected returns.
:type risk_free_rate: float, optional
:raises ValueError: if ``risk_free_rate`` is non-numeric
:return: asset weights for the Sharpe-maximising portfolio
:rtype: dict
"""
if not isinstance(risk_free_rate, (int, float)):
raise ValueError("risk_free_rate should be numeric")
# max_sharpe requires us to make a variable transformation.
# Here we treat w as the transformed variable.
self._objective = cp.quad_form(self._w, self.cov_matrix)
k = cp.Variable()
# Note: objectives are not scaled by k. Hence there are subtle differences
# between how these objectives work for max_sharpe vs min_volatility
for obj in self._additional_objectives:
self._objective += obj
# Overwrite original constraints with suitable constraints
# for the transformed max_sharpe problem
self._constraints = [
(self.expected_returns - risk_free_rate).T * self._w == 1,
cp.sum(self._w) == k,
k >= 0,
]
#  Rebuild original constraints with scaling factor
for raw_constr in self._additional_constraints_raw:
self._constraints.append(raw_constr(self.w / k))
# Sharpe ratio is invariant w.r.t scaled weights, so we must
# replace infinities and negative infinities
new_lower_bound = np.nan_to_num(self._lower_bounds, neginf=-1)
new_upper_bound = np.nan_to_num(self._upper_bounds, posinf=1)
self._constraints.append(self._w >= k * new_lower_bound)
self._constraints.append(self._w <= k * new_upper_bound)
self._solve_cvxpy_opt_problem()
# Inverse-transform
self.weights = (self._w.value / k.value).round(16) + 0.0
return dict(zip(self.tickers, self.weights))
def max_unconstrained_utility(self, risk_aversion=1):
@@ -242,6 +297,7 @@ class EfficientFrontier(base_optimizer.BaseConvexOptimizer):
self.weights = np.linalg.solve(A, b)
return dict(zip(self.tickers, self.weights))
# TODO: roll custom_objective into nonconvex_optimizer
def custom_objective(self, objective_function, *args):
"""
Optimise some objective function. While an implicit requirement is that the function
@@ -402,9 +458,9 @@ class EfficientFrontier(base_optimizer.BaseConvexOptimizer):
:rtype: (float, float, float)
"""
return base_optimizer.portfolio_performance(
self.weights,
self.expected_returns,
self.cov_matrix,
self.weights,
verbose,
risk_free_rate,
)

93
pypfopt/objective_functions.py
View File

@@ -37,7 +37,9 @@ def _objective_value(w, obj):
:rtype: float OR cp.Expression
"""
if isinstance(w, np.ndarray):
if np.isscalar(obj.value):
if np.isscalar(obj):
return obj
elif np.isscalar(obj.value):
return obj.value
else:
return obj.value.item()
@@ -46,32 +48,78 @@ def _objective_value(w, obj):
def portfolio_variance(w, cov_matrix):
if isinstance(w, pd.Series):
w = w.values
"""
Total portfolio variance (i.e square volatility).
:param w: asset weights in the portfolio
:type w: np.ndarray OR cp.Variable
:param cov_matrix: covariance matrix
:type cov_matrix: np.ndarray
:return: value of the objective function OR objective function expression
:rtype: float OR cp.Expression
"""
variance = cp.quad_form(w, cov_matrix)
return _objective_value(w, variance)
def L2_reg(w, gamma=1):
if isinstance(w, pd.Series):
w = w.values
L2_reg = gamma * cp.sum_squares(w)
return _objective_value(w, L2_reg)
def negative_mean_return(weights, expected_returns):
def portfolio_return(w, expected_returns, negative=True):
"""
Calculate the negative mean return of a portfolio
Calculate the (negative) mean return of a portfolio
:param weights: asset weights of the portfolio
:type weights: np.ndarray
:param w: asset weights in the portfolio
:type w: np.ndarray OR cp.Variable
:param expected_returns: expected return of each asset
:type expected_returns: pd.Series
:type expected_returns: np.ndarray
:param negative: whether quantity should be made negative (so we can minimise)
:type negative: boolean
:return: negative mean return
:rtype: float
"""
return -weights.dot(expected_returns)
sign = -1 if negative else 1
mu = sign * (w @ expected_returns)
return _objective_value(w, mu)
def sharpe_ratio(w, expected_returns, cov_matrix, risk_free_rate=0.02, negative=True):
"""
Calculate the (negative) Sharpe ratio of a portfolio
:param w: asset weights in the portfolio
:type w: np.ndarray
:param expected_returns: expected return of each asset
:type expected_returns: np.ndarray
:param cov_matrix: the covariance matrix of asset returns
:type cov_matrix: pd.DataFrame
:param risk_free_rate: risk-free rate of borrowing/lending, defaults to 0.02.
The period of the risk-free rate should correspond to the
frequency of expected returns.
:type risk_free_rate: float, optional
:param negative: whether quantity should be made negative (so we can minimise)
:type negative: boolean
:return: (negative) Sharpe ratio
:rtype: float
"""
mu = w @ expected_returns
sigma = cp.sqrt(cp.quad_form(w, cov_matrix))
sign = -1 if negative else 1
sharpe = sign * (mu - risk_free_rate) / sigma
return _objective_value(w, sharpe)
def L2_reg(w, gamma=1):
"""
"L2 regularisation", i.e gamma * ||w||^2
:param w: weights
:type w: np.ndarray OR cp.Variable
:param gamma: L2 regularisation parameter, defaults to 1. Increase if you want more
non-negligible weights
:type gamma: float, optional
:return: value of the objective function OR objective function expression
:rtype: float OR cp.Expression
"""
L2_reg = gamma * cp.sum_squares(w)
return _objective_value(w, L2_reg)
def negative_sharpe(
@@ -96,10 +144,7 @@ def negative_sharpe(
:return: negative Sharpe ratio
:rtype: float
"""
mu = weights.dot(expected_returns)
sigma = np.sqrt(np.dot(weights, np.dot(cov_matrix, weights.T)))
L2_reg = gamma * (weights ** 2).sum()
return -(mu - risk_free_rate) / sigma + L2_reg
pass
def volatility(weights, cov_matrix, gamma=0):
@@ -118,9 +163,7 @@ def volatility(weights, cov_matrix, gamma=0):
:return: portfolio variance
:rtype: float
"""
L2_reg = gamma * (weights ** 2).sum()
portfolio_volatility = np.dot(weights.T, np.dot(cov_matrix, weights))
return portfolio_volatility + L2_reg
pass
def negative_quadratic_utility(
@@ -140,9 +183,7 @@ def negative_quadratic_utility(
:type gamma: float, optional
"""
L2_reg = gamma * (weights ** 2).sum()
mu = weights.dot(expected_returns)
portfolio_volatility = np.dot(weights.T, np.dot(cov_matrix, weights))
return -(mu - 0.5 * risk_aversion * portfolio_volatility) + L2_reg
pass
# def negative_cvar(weights, returns, s=10000, beta=0.95, random_state=None):

18
tests/test_base_optimizer.py
View File

@@ -12,7 +12,7 @@ def test_custom_upper_bound():
*setup_efficient_frontier(data_only=True), weight_bounds=(0, 0.10)
)
ef.min_volatility()
ef.portfolio_performance()
np.testing.assert_allclose(ef._lower_bounds, np.array([0] * ef.n_assets))
assert ef.weights.max() <= 0.1
np.testing.assert_almost_equal(ef.weights.sum(), 1)
@@ -52,11 +52,27 @@ def test_custom_bounds_different_values():
)
def test_weight_bounds_minus_one_to_one():
ef = EfficientFrontier(
*setup_efficient_frontier(data_only=True), weight_bounds=(-1, 1)
)
assert ef.max_sharpe()
assert ef.min_volatility()
# TODO: fix
# assert ef.efficient_return(0.05)
# assert ef.efficient_risk(0.20)
def test_bound_input_types():
bounds = [0.01, 0.13]
ef = EfficientFrontier(
*setup_efficient_frontier(data_only=True), weight_bounds=bounds
)
assert ef
np.testing.assert_allclose(ef._lower_bounds, np.array([0.01] * ef.n_assets))
np.testing.assert_allclose(ef._upper_bounds, np.array([0.13] * ef.n_assets))
lb = np.array([0.01, 0.02] * 10)
ub = np.array([0.07, 0.2] * 10)
assert EfficientFrontier(

320
tests/test_efficient_frontier.py
View File

@@ -2,10 +2,12 @@ import warnings
import numpy as np
import pandas as pd
import pytest
import scipy.optimize as sco
from pypfopt import EfficientFrontier
from tests.utilities_for_tests import get_data, setup_efficient_frontier
from pypfopt import risk_models
from pypfopt import objective_functions
from tests.utilities_for_tests import get_data, setup_efficient_frontier
def test_data_source():
@@ -29,77 +31,183 @@ def test_returns_dataframe():
def test_efficient_frontier_inheritance():
ef = setup_efficient_frontier()
assert ef.clean_weights
assert isinstance(ef.initial_guess, np.ndarray)
assert isinstance(ef.constraints, list)
assert ef.n_assets
assert ef.tickers
assert isinstance(ef._constraints, list)
assert isinstance(ef._lower_bounds, np.ndarray)
assert isinstance(ef._upper_bounds, np.ndarray)
# def test_max_sharpe_input_errors():
# with pytest.raises(ValueError):
# ef = EfficientFrontier(*setup_efficient_frontier(data_only=True), gamma="2")
# with warnings.catch_warnings(record=True) as w:
# ef = EfficientFrontier(*setup_efficient_frontier(data_only=True), gamma=-1)
# assert len(w) == 1
# assert issubclass(w[0].category, UserWarning)
# assert str(w[0].message) == "in most cases, gamma should be positive"
# with pytest.raises(ValueError):
# ef.max_sharpe(risk_free_rate="0.2")
def test_portfolio_performance():
ef = setup_efficient_frontier()
with pytest.raises(ValueError):
ef.portfolio_performance()
ef.min_volatility()
perf = ef.portfolio_performance()
assert isinstance(perf, tuple)
assert len(perf) == 3
assert isinstance(perf[0], float)
# def test_portfolio_performance():
# ef = setup_efficient_frontier()
# with pytest.raises(ValueError):
# ef.portfolio_performance()
# ef.max_sharpe()
# assert ef.portfolio_performance()
def test_min_volatility():
ef = setup_efficient_frontier()
w = ef.min_volatility()
assert isinstance(w, dict)
assert set(w.keys()) == set(ef.tickers)
np.testing.assert_almost_equal(ef.weights.sum(), 1)
assert all([i >= 0 for i in w.values()])
# TODO fix
np.testing.assert_allclose(
ef.portfolio_performance(),
(0.17931232481259154, 0.15915084514118694, 1.00101463282373),
)
# def test_max_sharpe_long_only():
# ef = setup_efficient_frontier()
# w = ef.max_sharpe()
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# assert all([i >= 0 for i in w.values()])
def test_min_volatility_short():
ef = EfficientFrontier(
*setup_efficient_frontier(data_only=True), weight_bounds=(None, None)
)
w = ef.min_volatility()
assert isinstance(w, dict)
assert set(w.keys()) == set(ef.tickers)
np.testing.assert_almost_equal(ef.weights.sum(), 1)
np.testing.assert_allclose(
ef.portfolio_performance(),
(0.1721356467349655, 0.1555915367269669, 0.9777887019776287),
)
# np.testing.assert_allclose(
# ef.portfolio_performance(),
# (0.3303554227420522, 0.21671629569400466, 1.4320816150358278),
# )
# Shorting should reduce volatility
volatility = ef.portfolio_performance()[1]
ef_long_only = setup_efficient_frontier()
ef_long_only.min_volatility()
long_only_volatility = ef_long_only.portfolio_performance()[1]
assert volatility < long_only_volatility
# def test_max_sharpe_short():
# ef = EfficientFrontier(
# *setup_efficient_frontier(data_only=True), weight_bounds=(None, None)
# )
# w = ef.max_sharpe()
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# np.testing.assert_allclose(
# ef.portfolio_performance(),
# (0.4072375737868628, 0.24823079606119094, 1.5599900573634125),
# )
# sharpe = ef.portfolio_performance()[2]
def test_min_volatility_L2_reg():
ef = setup_efficient_frontier()
ef.add_objective(objective_functions.L2_reg, gamma=5)
weights = ef.min_volatility()
assert isinstance(weights, dict)
assert set(weights.keys()) == set(ef.tickers)
np.testing.assert_almost_equal(ef.weights.sum(), 1)
assert all([i >= 0 for i in weights.values()])
# ef_long_only = setup_efficient_frontier()
# ef_long_only.max_sharpe()
# long_only_sharpe = ef_long_only.portfolio_performance()[2]
ef2 = setup_efficient_frontier()
ef2.min_volatility()
# assert sharpe > long_only_sharpe
# L2_reg should pull close to equal weight
equal_weight = np.full((ef.n_assets,), 1 / ef.n_assets)
assert (
np.abs(equal_weight - ef.weights).sum()
< np.abs(equal_weight - ef2.weights).sum()
)
np.testing.assert_allclose(
ef.portfolio_performance(),
(0.2382083649754719, 0.20795460936504614, 1.049307662098637),
)
# def test_weight_bounds_minus_one_to_one():
# ef = EfficientFrontier(
# *setup_efficient_frontier(data_only=True), weight_bounds=(-1, 1)
# )
# assert ef.max_sharpe()
# assert ef.min_volatility()
# assert ef.efficient_return(0.05)
# assert ef.efficient_risk(0.20)
def test_min_volatility_L2_reg_many_values():
ef = setup_efficient_frontier()
ef.min_volatility()
# Count the number of weights more 1%
initial_number = sum(ef.weights > 0.01)
for _ in range(10):
ef.add_objective(objective_functions.L2_reg, gamma=0.05)
ef.min_volatility()
np.testing.assert_almost_equal(ef.weights.sum(), 1)
new_number = sum(ef.weights > 0.01)
# Higher gamma should reduce the number of small weights
assert new_number >= initial_number
initial_number = new_number
def test_min_volatility_L2_reg_limit_case():
ef = setup_efficient_frontier()
ef.add_objective(objective_functions.L2_reg, gamma=1e10)
ef.min_volatility()
equal_weights = np.array([1 / ef.n_assets] * ef.n_assets)
np.testing.assert_array_almost_equal(ef.weights, equal_weights)
def test_min_volatility_cvxpy_vs_scipy():
# cvxpy
ef = setup_efficient_frontier()
ef.min_volatility()
w1 = ef.weights
# scipy
args = (ef.cov_matrix,)
initial_guess = np.array([1 / ef.n_assets] * ef.n_assets)
result = sco.minimize(
objective_functions.volatility,
x0=initial_guess,
args=args,
method="SLSQP",
bounds=[(0, 1)] * 20,
constraints=[{"type": "eq", "fun": lambda x: np.sum(x) - 1}],
)
w2 = result["x"]
cvxpy_var = objective_functions.portfolio_variance(w1, ef.cov_matrix)
scipy_var = objective_functions.portfolio_variance(w2, ef.cov_matrix)
assert cvxpy_var <= scipy_var
def test_max_sharpe_long_only():
ef = setup_efficient_frontier()
w = ef.max_sharpe()
assert isinstance(w, dict)
assert set(w.keys()) == set(ef.tickers)
np.testing.assert_almost_equal(ef.weights.sum(), 1)
assert all([i >= 0 for i in w.values()])
np.testing.assert_allclose(
ef.portfolio_performance(),
(0.33035037367760506, 0.21671276571944567, 1.4320816434015786),
)
def test_max_sharpe_long_weight_bounds():
ef = EfficientFrontier(
*setup_efficient_frontier(data_only=True), weight_bounds=(0.03, 0.13)
)
ef.max_sharpe()
np.testing.assert_almost_equal(ef.weights.sum(), 1)
assert ef.weights.min() >= 0.03
assert ef.weights.max() <= 0.13
bounds = [(0.01, 0.13), (0.02, 0.11)] * 10
ef = EfficientFrontier(
*setup_efficient_frontier(data_only=True), weight_bounds=bounds
)
ef.max_sharpe()
assert (0.01 <= ef.weights[::2]).all() and (ef.weights[::2] <= 0.13).all()
assert (0.02 <= ef.weights[1::2]).all() and (ef.weights[1::2] <= 0.11).all()
def test_max_sharpe_short():
ef = EfficientFrontier(
*setup_efficient_frontier(data_only=True), weight_bounds=(None, None)
)
w = ef.max_sharpe()
assert isinstance(w, dict)
assert set(w.keys()) == set(ef.tickers)
np.testing.assert_almost_equal(ef.weights.sum(), 1)
np.testing.assert_allclose(
ef.portfolio_performance(),
(0.4072439477276246, 0.24823487545231313, 1.5599900981762558),
)
sharpe = ef.portfolio_performance()[2]
ef_long_only = setup_efficient_frontier()
ef_long_only.max_sharpe()
long_only_sharpe = ef_long_only.portfolio_performance()[2]
assert sharpe > long_only_sharpe
# def test_max_sharpe_L2_reg():
@@ -108,7 +216,6 @@ def test_efficient_frontier_inheritance():
# w = ef.max_sharpe()
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# assert all([i >= 0 for i in w.values()])
@@ -166,7 +273,6 @@ def test_efficient_frontier_inheritance():
# w = ef.max_sharpe()
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# np.testing.assert_allclose(
# ef.portfolio_performance(),
@@ -194,7 +300,6 @@ def test_efficient_frontier_inheritance():
# w = ef.max_unconstrained_utility(2)
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_allclose(
# ef.portfolio_performance(),
# (1.3507326549906276, 0.8218067458322021, 1.6192768698230409),
@@ -215,88 +320,11 @@ def test_efficient_frontier_inheritance():
# ef.max_unconstrained_utility(-1)
def test_min_volatility():
ef = setup_efficient_frontier()
w = ef.min_volatility()
assert isinstance(w, dict)
assert set(w.keys()) == set(ef.tickers)
np.testing.assert_almost_equal(ef.weights.sum(), 1)
assert all([i >= 0 for i in w.values()])
# TODO fix
np.testing.assert_allclose(
ef.portfolio_performance(),
(0.1791557243114251, 0.15915426422116669, 1.0000091740567905),
)
def test_min_volatility_short():
ef = EfficientFrontier(
*setup_efficient_frontier(data_only=True), weight_bounds=(None, None)
)
w = ef.min_volatility()
assert isinstance(w, dict)
assert set(w.keys()) == set(ef.tickers)
np.testing.assert_almost_equal(ef.weights.sum(), 1)
np.testing.assert_allclose(
ef.portfolio_performance(),
(0.1719799152621441, 0.1555954785460613, 0.9767630568850568),
)
# Shorting should reduce volatility
volatility = ef.portfolio_performance()[1]
ef_long_only = setup_efficient_frontier()
ef_long_only.min_volatility()
long_only_volatility = ef_long_only.portfolio_performance()[1]
assert volatility < long_only_volatility
def test_min_volatility_L2_reg():
ef = setup_efficient_frontier()
ef.add_objective(objective_functions.L2_reg, gamma=5)
weights = ef.min_volatility()
assert isinstance(weights, dict)
assert set(weights.keys()) == set(ef.tickers)
np.testing.assert_almost_equal(ef.weights.sum(), 1)
assert all([i >= 0 for i in weights.values()])
ef2 = setup_efficient_frontier()
ef2.min_volatility()
# L2_reg should pull close to equal weight
equal_weight = np.full((ef.n_assets,), 1 / ef.n_assets)
assert (
np.abs(equal_weight - ef.weights).sum()
< np.abs(equal_weight - ef2.weights).sum()
)
np.testing.assert_allclose(
ef.portfolio_performance(),
(0.23136193240984504, 0.1955259140191799, 1.0809919159314694),
)
def test_min_volatility_L2_reg_many_values():
ef = setup_efficient_frontier()
ef.min_volatility()
# Count the number of weights more 1%
initial_number = sum(ef.weights > 0.01)
for _ in range(10):
ef.add_objective(objective_functions.L2_reg, gamma=0.05)
ef.min_volatility()
np.testing.assert_almost_equal(ef.weights.sum(), 1)
new_number = sum(ef.weights > 0.01)
# Higher gamma should reduce the number of small weights
assert new_number >= initial_number
initial_number = new_number
# def test_efficient_risk():
# ef = setup_efficient_frontier()
# w = ef.efficient_risk(0.19)
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# assert all([i >= 0 for i in w.values()])
# np.testing.assert_allclose(
@@ -331,7 +359,6 @@ def test_min_volatility_L2_reg_many_values():
# w = ef.efficient_risk(0.19)
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# np.testing.assert_allclose(
# ef.portfolio_performance(),
@@ -353,7 +380,6 @@ def test_min_volatility_L2_reg_many_values():
# w = ef.efficient_risk(0.19)
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# assert all([i >= 0 for i in w.values()])
@@ -386,7 +412,6 @@ def test_min_volatility_L2_reg_many_values():
# w = ef.efficient_risk(0.19, market_neutral=True)
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 0)
# assert (ef.weights < 1).all() and (ef.weights > -1).all()
# np.testing.assert_allclose(
@@ -419,7 +444,6 @@ def test_min_volatility_L2_reg_many_values():
# w = ef.efficient_return(0.25)
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# assert all([i >= 0 for i in w.values()])
# np.testing.assert_allclose(
@@ -454,7 +478,6 @@ def test_min_volatility_L2_reg_many_values():
# w = ef.efficient_return(0.25)
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# np.testing.assert_allclose(
# ef.portfolio_performance(), (0.25, 0.1682647442258144, 1.3668935881968987)
@@ -474,7 +497,6 @@ def test_min_volatility_L2_reg_many_values():
# w = ef.efficient_return(0.25)
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# assert all([i >= 0 for i in w.values()])
# np.testing.assert_allclose(
@@ -505,7 +527,6 @@ def test_min_volatility_L2_reg_many_values():
# w = ef.efficient_return(0.25, market_neutral=True)
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 0)
# assert (ef.weights < 1).all() and (ef.weights > -1).all()
# np.testing.assert_almost_equal(
@@ -537,7 +558,6 @@ def test_min_volatility_L2_reg_many_values():
# w = ef.max_sharpe()
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# assert all([i >= 0 for i in w.values()])
# np.testing.assert_allclose(
@@ -555,7 +575,6 @@ def test_min_volatility_L2_reg_many_values():
# w = ef.max_sharpe()
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# np.testing.assert_allclose(
# ef.portfolio_performance(),
@@ -571,7 +590,6 @@ def test_min_volatility_L2_reg_many_values():
# w = ef.min_volatility()
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# assert all([i >= 0 for i in w.values()])
# np.testing.assert_allclose(
@@ -587,7 +605,6 @@ def test_min_volatility_L2_reg_many_values():
# w = ef.efficient_return(0.12)
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# assert all([i >= 0 for i in w.values()])
# np.testing.assert_allclose(
@@ -603,7 +620,6 @@ def test_min_volatility_L2_reg_many_values():
# w = ef.max_sharpe()
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# assert all([i >= 0 for i in w.values()])
# np.testing.assert_allclose(
@@ -620,7 +636,6 @@ def test_min_volatility_L2_reg_many_values():
# w = ef.min_volatility()
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 1)
# assert all([i >= 0 for i in w.values()])
# np.testing.assert_allclose(
@@ -638,7 +653,6 @@ def test_min_volatility_L2_reg_many_values():
# w = ef.efficient_risk(0.19, market_neutral=True)
# assert isinstance(w, dict)
# assert set(w.keys()) == set(ef.tickers)
# assert set(w.keys()) == set(ef.expected_returns.index)
# np.testing.assert_almost_equal(ef.weights.sum(), 0)
# assert (ef.weights < 1).all() and (ef.weights > -1).all()
# np.testing.assert_allclose(

26
tests/test_objective_functions.py
View File

@@ -6,22 +6,22 @@ from pypfopt.risk_models import sample_cov
from tests.utilities_for_tests import get_data
def test_negative_mean_return_dummy():
def test_portfolio_return_dummy():
w = np.array([0.3, 0.1, 0.2, 0.25, 0.15])
e_rets = pd.Series([0.19, 0.08, 0.09, 0.23, 0.17])
negative_mu = objective_functions.negative_mean_return(w, e_rets)
assert isinstance(negative_mu, float)
assert negative_mu < 0
np.testing.assert_almost_equal(negative_mu, -w.dot(e_rets))
np.testing.assert_almost_equal(negative_mu, -(w * e_rets).sum())
mu = objective_functions.portfolio_return(w, e_rets, negative=False)
assert isinstance(mu, float)
assert mu > 0
np.testing.assert_almost_equal(mu, w.dot(e_rets))
np.testing.assert_almost_equal(mu, (w * e_rets).sum())
def test_negative_mean_return_real():
def test_portfolio_return_real():
df = get_data()
e_rets = mean_historical_return(df)
w = np.array([1 / len(e_rets)] * len(e_rets))
negative_mu = objective_functions.negative_mean_return(w, e_rets)
negative_mu = objective_functions.portfolio_return(w, e_rets)
assert isinstance(negative_mu, float)
assert negative_mu < 0
assert negative_mu == -w.dot(e_rets)
@@ -29,24 +29,22 @@ def test_negative_mean_return_real():
np.testing.assert_almost_equal(-e_rets.sum() / len(e_rets), negative_mu)
def test_negative_sharpe():
def test_sharpe_ratio():
df = get_data()
e_rets = mean_historical_return(df)
S = sample_cov(df)
w = np.array([1 / len(e_rets)] * len(e_rets))
sharpe = objective_functions.negative_sharpe(w, e_rets, S)
sharpe = objective_functions.sharpe_ratio(w, e_rets, S)
assert isinstance(sharpe, float)
assert sharpe < 0
sigma = np.sqrt(np.dot(w, np.dot(S, w.T)))
negative_mu = objective_functions.negative_mean_return(w, e_rets)
negative_mu = objective_functions.portfolio_return(w, e_rets)
np.testing.assert_almost_equal(sharpe * sigma - 0.02, negative_mu)
# Risk free rate increasing should lead to negative Sharpe increasing.
assert sharpe < objective_functions.negative_sharpe(
w, e_rets, S, risk_free_rate=0.1
)
assert sharpe < objective_functions.sharpe_ratio(w, e_rets, S, risk_free_rate=0.1)
def test_negative_quadratic_utility():