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

Add parametrization of quadratic utility

Browse Source
This commit is contained in:
phschiele
2021-05-18 20:22:03 +02:00
parent 65c790057b
commit 538087eb61
4 changed files with 53 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

48
pypfopt/base_optimizer.py
View File

@@ -7,13 +7,16 @@ Additionally, we define a general utility function ``portfolio_performance`` to
evaluate return and risk for a given set of portfolio weights. evaluate return and risk for a given set of portfolio weights.
""" """
import collections import collections
import copy
import json import json
import warnings import warnings
from collections.abc import Iterable
from typing import List
import numpy as np import numpy as np
import pandas as pd import pandas as pd
import cvxpy as cp import cvxpy as cp
import scipy.optimize as sco import scipy.optimize as sco
from . import objective_functions from . import objective_functions
from . import exceptions from . import exceptions
@@ -223,24 +226,26 @@ class BaseConvexOptimizer(BaseOptimizer):
def is_parameter_defined(self, parameter_name: str) -> bool: def is_parameter_defined(self, parameter_name: str) -> bool:
is_defined = False is_defined = False
for const in self._constraints: objective_and_constraints = self._constraints + [self._objective] if self._objective is not None else self._constraints
for arg in const.args: for expr in objective_and_constraints:
if isinstance(arg, cp.Parameter): params = [arg for arg in get_all_args(expr) if isinstance(arg, cp.Parameter)]
if arg.name() == parameter_name and not is_defined: for param in params:
is_defined = True if param.name() == parameter_name and not is_defined:
elif arg.name() == parameter_name and is_defined: is_defined = True
raise Exception('Parameter name defined multiple times') elif param.name() == parameter_name and is_defined:
raise Exception('Parameter name defined multiple times')
return is_defined return is_defined
def update_parameter_value(self, parameter_name: str, new_value: float) -> None: def update_parameter_value(self, parameter_name: str, new_value: float) -> None:
assert self.is_parameter_defined(parameter_name) assert self.is_parameter_defined(parameter_name)
was_updated = False was_updated = False
for const in self._constraints: objective_and_constraints = self._constraints + [self._objective] if self._objective is not None else self._constraints
for arg in const.args: for expr in objective_and_constraints:
if isinstance(arg, cp.Parameter): params = [arg for arg in get_all_args(expr) if isinstance(arg, cp.Parameter)]
if arg.name() == parameter_name: for param in params:
arg.value = new_value if param.name() == parameter_name:
was_updated = True param.value = new_value
was_updated = True
assert was_updated assert was_updated
def _solve_cvxpy_opt_problem(self): def _solve_cvxpy_opt_problem(self):
@@ -526,3 +531,18 @@ def portfolio_performance(
if verbose: if verbose:
print("Annual volatility: {:.1f}%".format(100 * sigma)) print("Annual volatility: {:.1f}%".format(100 * sigma))
return None, sigma, None return None, sigma, None
def get_all_args(expression: cp.Expression) -> List[cp.Expression]:
if expression.args == []:
return [expression]
else:
return list(flatten([get_all_args(arg) for arg in expression.args]))
def flatten(l: Iterable) -> Iterable:
for el in l:
if isinstance(el, Iterable) and not isinstance(el, (str, bytes)):
yield from flatten(el)
else:
yield el

22
pypfopt/efficient_frontier/efficient_frontier.py
View File

@@ -299,13 +299,17 @@ class EfficientFrontier(base_optimizer.BaseConvexOptimizer):
if risk_aversion <= 0: if risk_aversion <= 0:
raise ValueError("risk aversion coefficient must be greater than zero") raise ValueError("risk aversion coefficient must be greater than zero")
self._objective = objective_functions.quadratic_utility( update_existing_parameter = self.is_parameter_defined('risk_aversion')
self._w, self.expected_returns, self.cov_matrix, risk_aversion=risk_aversion if update_existing_parameter:
) self.update_parameter_value('risk_aversion', risk_aversion)
for obj in self._additional_objectives: else:
self._objective += obj self._objective = objective_functions.quadratic_utility(
self._w, self.expected_returns, self.cov_matrix, risk_aversion=risk_aversion
)
for obj in self._additional_objectives:
self._objective += obj
self._make_weight_sum_constraint(market_neutral) self._make_weight_sum_constraint(market_neutral)
return self._solve_cvxpy_opt_problem() return self._solve_cvxpy_opt_problem()
def efficient_risk(self, target_volatility, market_neutral=False): def efficient_risk(self, target_volatility, market_neutral=False):
@@ -376,9 +380,9 @@ class EfficientFrontier(base_optimizer.BaseConvexOptimizer):
"target_return must be lower than the maximum possible return" "target_return must be lower than the maximum possible return"
) )
update_existing_parameter = self.is_parameter_defined('target_return_par') update_existing_parameter = self.is_parameter_defined('target_return')
if update_existing_parameter: if update_existing_parameter:
self.update_parameter_value('target_return_par', target_return) self.update_parameter_value('target_return', target_return)
else: else:
self._objective = objective_functions.portfolio_variance( self._objective = objective_functions.portfolio_variance(
self._w, self.cov_matrix self._w, self.cov_matrix
@@ -390,7 +394,7 @@ class EfficientFrontier(base_optimizer.BaseConvexOptimizer):
for obj in self._additional_objectives: for obj in self._additional_objectives:
self._objective += obj self._objective += obj
target_return_par = cp.Parameter(name='target_return_par', value=target_return) target_return_par = cp.Parameter(name='target_return', value=target_return)
self._constraints.append(ret >= target_return_par) self._constraints.append(ret >= target_return_par)
self._make_weight_sum_constraint(market_neutral) self._make_weight_sum_constraint(market_neutral)
return self._solve_cvxpy_opt_problem() return self._solve_cvxpy_opt_problem()

3
pypfopt/objective_functions.py
View File

@@ -158,7 +158,8 @@ def quadratic_utility(w, expected_returns, cov_matrix, risk_aversion, negative=T
mu = w @ expected_returns mu = w @ expected_returns
variance = cp.quad_form(w, cov_matrix) variance = cp.quad_form(w, cov_matrix)
utility = mu - 0.5 * risk_aversion * variance risk_aversion_par = cp.Parameter(value=risk_aversion, name='risk_aversion', nonneg=True)
utility = mu - 0.5 * risk_aversion_par * variance
return _objective_value(w, sign * utility) return _objective_value(w, sign * utility)

10
pypfopt/plotting.py
View File

@@ -168,15 +168,13 @@ def _plot_ef(ef, ef_param, ef_param_range, ax, show_assets):
# Create a portfolio for each value of ef_param_range # Create a portfolio for each value of ef_param_range
for param_value in ef_param_range: for param_value in ef_param_range:
ef_i = copy.deepcopy(ef)
try: try:
if ef_param == "utility": if ef_param == "utility":
ef_i.max_quadratic_utility(param_value) ef.max_quadratic_utility(param_value)
elif ef_param == "risk": elif ef_param == "risk":
ef_i.efficient_risk(param_value) ef.efficient_risk(param_value)
elif ef_param == "return": elif ef_param == "return":
ef_i.efficient_return(param_value) ef.efficient_return(param_value)
else: else:
raise NotImplementedError( raise NotImplementedError(
"ef_param should be one of {'utility', 'risk', 'return'}" "ef_param should be one of {'utility', 'risk', 'return'}"
@@ -184,7 +182,7 @@ def _plot_ef(ef, ef_param, ef_param_range, ax, show_assets):
except exceptions.OptimizationError: except exceptions.OptimizationError:
continue continue
ret, sigma, _ = ef_i.portfolio_performance() ret, sigma, _ = ef.portfolio_performance()
mus.append(ret) mus.append(ret)
sigmas.append(sigma) sigmas.append(sigma)