"""
The ``efficient_frontier`` submodule houses the EfficientFrontier class, which generates
classical mean-variance optimal portfolios for a variety of objectives and constraints
"""
import warnings
import cvxpy as cp
import numpy as np
import pandas as pd
from .. import base_optimizer, exceptions, objective_functions
[文档]
class EfficientFrontier(base_optimizer.BaseConvexOptimizer):
"""
一个 EfficientFrontier 对象（继承自 BaseConvexOptimizer）包含多个优化方法，可以用不同的参数调用（对应不同的目标函数）。
注意：如果要对目标/约束条件/边界/参数做任何改变，应该实例化一个新的 EfficientFrontier 对象。
可用变量:
- Inputs:
- ``n_assets`` - int
- ``tickers`` - str list
- ``bounds`` - float tuple 或 (float tuple) list
- ``cov_matrix`` - np.ndarray
- ``expected_returns`` - np.ndarray
- ``solver`` - str
- ``solver_options`` - {str: str} dict
- Output: ``weights`` - np.ndarray
Public methods:
- ``min_volatility()`` 优化最小波动率
- ``max_sharpe()`` 对最大夏普比率进行优化（又称切线组合）
- ``max_quadratic_utility()`` 在一定的风险规避条件下，使二次效用最大化。
- ``efficient_risk()`` 在给定的目标风险下，使收益最大
- ``efficient_return()`` 在给定的目标收益下，使风险最小
- ``add_objective()`` 为优化问题添加一个（凸）目标
- ``add_constraint()`` 给优化问题添加一个约束条件
- ``convex_objective()`` 解决一个带有线性约束的通用凸目标
- ``portfolio_performance()`` 计算优化组合的期望收益率、波动率和夏普比率。
- ``set_weights()`` 从一个权重字典中创建 self.weights (np.ndarray)
- ``clean_weights()`` 对权重进行四舍五入，并将接近零值的部分删除。
- ``save_weights_to_file()`` 将权重保存为 csv、json 或 txt 格式。
"""
[文档]
def __init__(
self,
expected_returns,
cov_matrix,
weight_bounds=(0, 1),
solver=None,
verbose=False,
solver_options=None,
):
"""
:param expected_returns: expected returns for each asset. Can be None if
optimising for volatility only (but not recommended).
:type expected_returns: pd.Series, list, np.ndarray
:param cov_matrix: covariance of returns for each asset. This **must** be
positive semidefinite, otherwise optimization will fail.
:type cov_matrix: pd.DataFrame or np.array
:param weight_bounds: minimum and maximum weight of each asset OR single min/max pair
if all identical, defaults to (0, 1). Must be changed to (-1, 1)
for portfolios with shorting.
:type weight_bounds: tuple OR tuple list, optional
:param solver: name of solver. list available solvers with: `cvxpy.installed_solvers()`
:type solver: str
:param verbose: whether performance and debugging info should be printed, defaults to False
:type verbose: bool, optional
:param solver_options: parameters for the given solver
:type solver_options: dict, optional
:raises TypeError: if ``expected_returns`` is not a series, list or array
:raises TypeError: if ``cov_matrix`` is not a dataframe or array
"""
# Inputs
self.cov_matrix = self._validate_cov_matrix(cov_matrix)
self.expected_returns = self._validate_expected_returns(expected_returns)
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,
weight_bounds,
solver=solver,
verbose=verbose,
solver_options=solver_options,
)
@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()
else:
raise TypeError("expected_returns is not a series, list or array")
@staticmethod
def _validate_cov_matrix(cov_matrix):
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:
raise TypeError("cov_matrix is not a dataframe or array")
def _validate_returns(self, returns):
"""
Helper method to validate daily returns (needed for some efficient frontiers)
"""
if not isinstance(returns, (pd.DataFrame, np.ndarray)):
raise TypeError("returns should be a pd.Dataframe or np.ndarray")
returns_df = pd.DataFrame(returns)
if returns_df.isnull().values.any():
warnings.warn(
"Removing NaNs from returns",
UserWarning,
)
returns_df = returns_df.dropna(axis=0, how="any")
if self.expected_returns is not None:
if returns_df.shape[1] != len(self.expected_returns):
raise ValueError(
"returns columns do not match expected_returns. Please check your tickers."
)
return returns_df
def _make_weight_sum_constraint(self, is_market_neutral):
"""
Helper method to make the weight sum constraint. If market neutral,
validate the weights proided in the constructor.
"""
if is_market_neutral:
# Check and fix bounds
portfolio_possible = np.any(self._lower_bounds < 0)
if not portfolio_possible:
warnings.warn(
"Market neutrality requires shorting - bounds have been amended",
RuntimeWarning,
)
self._map_bounds_to_constraints((-1, 1))
# Delete original constraints
del self._constraints[0]
del self._constraints[0]
self.add_constraint(lambda w: cp.sum(w) == 0)
else:
self.add_constraint(lambda w: cp.sum(w) == 1)
self._market_neutral = is_market_neutral
[文档]
def min_volatility(self):
"""
最小化波动率。
:return: asset weights for the volatility-minimising portfolio
:rtype: OrderedDict
"""
self._objective = objective_functions.portfolio_variance(
self._w, self.cov_matrix
)
for obj in self._additional_objectives:
self._objective += obj
self.add_constraint(lambda w: cp.sum(w) == 1)
return self._solve_cvxpy_opt_problem()
def _max_return(self, return_value=True):
"""
Helper method to maximise return. This should not be used to optimize a portfolio.
:return: asset weights for the return-minimising portfolio
:rtype: OrderedDict
"""
if self.expected_returns is None:
raise ValueError("no expected returns provided")
self._objective = objective_functions.portfolio_return(
self._w, self.expected_returns
)
self.add_constraint(lambda w: cp.sum(w) == 1)
res = self._solve_cvxpy_opt_problem()
if return_value:
return -self._opt.value
else:
return res
[文档]
def max_sharpe(self, risk_free_rate=0.02):
"""
最大化夏普比率。这个结果也被称为切线组合，因为它是资本市场线与有效前沿相切的组合。
这是一个进行了一定的变量替换后的凸优化问题。详见 `Cornuejols and Tutuncu (2006) <http://web.math.ku.dk/~rolf/CT_FinOpt.pdf>`_ 。
: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: OrderedDict
"""
if not isinstance(risk_free_rate, (int, float)):
raise ValueError("risk_free_rate should be numeric")
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"
)
self._risk_free_rate = risk_free_rate
# 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
if len(self._additional_objectives) > 0:
warnings.warn(
"max_sharpe transforms the optimization problem so additional objectives may not work as expected."
)
for obj in self._additional_objectives:
self._objective += obj
new_constraints = []
# Must rebuild the constraints
for constr in self._constraints:
if isinstance(constr, cp.constraints.nonpos.Inequality):
# Either the first or second item is the expression
if isinstance(
constr.args[0], cp.expressions.constants.constant.Constant
):
new_constraints.append(constr.args[1] >= constr.args[0] * k)
else:
new_constraints.append(constr.args[0] <= constr.args[1] * k)
elif isinstance(constr, cp.constraints.zero.Equality):
new_constraints.append(constr.args[0] == constr.args[1] * k)
else:
raise TypeError(
"Please check that your constraints are in a suitable format"
)
# Transformed max_sharpe convex problem:
self._constraints = [
(self.expected_returns - risk_free_rate).T @ self._w == 1,
cp.sum(self._w) == k,
k >= 0,
] + new_constraints
self._solve_cvxpy_opt_problem()
# Inverse-transform
self.weights = (self._w.value / k.value).round(16) + 0.0
return self._make_output_weights()
[文档]
def max_quadratic_utility(self, risk_aversion=1, market_neutral=False):
r"""
使给定的二次方效用最大化，即：
.. math::
\max_w w^T \mu - \frac \delta 2 w^T \Sigma w
:param risk_aversion: risk aversion parameter (must be greater than 0),
defaults to 1
:type risk_aversion: positive float
:param market_neutral: whether the portfolio should be market neutral (weights sum to zero),
defaults to False. Requires negative lower weight bound.
:param market_neutral: bool, optional
:return: asset weights for the maximum-utility portfolio
:rtype: OrderedDict
"""
if risk_aversion <= 0:
raise ValueError("risk aversion coefficient must be greater than zero")
update_existing_parameter = self.is_parameter_defined("risk_aversion")
if update_existing_parameter:
self._validate_market_neutral(market_neutral)
self.update_parameter_value("risk_aversion", risk_aversion)
else:
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)
return self._solve_cvxpy_opt_problem()
[文档]
def efficient_risk(self, target_volatility, market_neutral=False):
"""
在目标风险下实现收益最大化。由此产生的投资组合的波动率将小于目标值（但不保证相等）。
:param target_volatility: the desired maximum volatility of the resulting portfolio.
:type target_volatility: float
:param market_neutral: whether the portfolio should be market neutral (weights sum to zero),
defaults to False. Requires negative lower weight bound.
:param market_neutral: bool, optional
:raises ValueError: if ``target_volatility`` is not a positive float
:raises ValueError: if no portfolio can be found with volatility equal to ``target_volatility``
:raises ValueError: if ``risk_free_rate`` is non-numeric
:return: asset weights for the efficient risk portfolio
:rtype: OrderedDict
"""
if not isinstance(target_volatility, (float, int)) or target_volatility < 0:
raise ValueError("target_volatility should be a positive float")
global_min_volatility = np.sqrt(1 / np.sum(np.linalg.pinv(self.cov_matrix)))
if target_volatility < global_min_volatility:
raise ValueError(
"The minimum volatility is {:.3f}. Please use a higher target_volatility".format(
global_min_volatility
)
)
update_existing_parameter = self.is_parameter_defined("target_variance")
if update_existing_parameter:
self._validate_market_neutral(market_neutral)
self.update_parameter_value("target_variance", target_volatility**2)
else:
self._objective = objective_functions.portfolio_return(
self._w, self.expected_returns
)
variance = objective_functions.portfolio_variance(self._w, self.cov_matrix)
for obj in self._additional_objectives:
self._objective += obj
target_variance = cp.Parameter(
name="target_variance", value=target_volatility**2, nonneg=True
)
self.add_constraint(lambda _: variance <= target_variance)
self._make_weight_sum_constraint(market_neutral)
return self._solve_cvxpy_opt_problem()
[文档]
def efficient_return(self, target_return, market_neutral=False):
"""
计算“马科维茨投资组合”，在给定的目标收益下，使波动率最小。
:param target_return: the desired return of the resulting portfolio.
:type target_return: float
:param market_neutral: whether the portfolio should be market neutral (weights sum to zero),
defaults to False. Requires negative lower weight bound.
:type market_neutral: bool, optional
:raises ValueError: if ``target_return`` is not a positive float
:raises ValueError: if no portfolio can be found with return equal to ``target_return``
:return: asset weights for the Markowitz portfolio
:rtype: OrderedDict
"""
if not isinstance(target_return, float):
raise ValueError("target_return should be a float")
if not self._max_return_value:
a = self.deepcopy()
self._max_return_value = a._max_return()
if target_return > self._max_return_value:
raise ValueError(
"target_return must be lower than the maximum possible return"
)
update_existing_parameter = self.is_parameter_defined("target_return")
if update_existing_parameter:
self._validate_market_neutral(market_neutral)
self.update_parameter_value("target_return", target_return)
else:
self._objective = objective_functions.portfolio_variance(
self._w, self.cov_matrix
)
ret = objective_functions.portfolio_return(
self._w, self.expected_returns, negative=False
)
for obj in self._additional_objectives:
self._objective += obj
target_return_par = cp.Parameter(name="target_return", value=target_return)
self.add_constraint(lambda _: ret >= target_return_par)
self._make_weight_sum_constraint(market_neutral)
return self._solve_cvxpy_opt_problem()
def _validate_market_neutral(self, market_neutral: bool) -> None:
if self._market_neutral != market_neutral:
raise exceptions.InstantiationError(
"A new instance must be created when changing market_neutral."
)
