"""
``objective_functions`` 模块提供优化目标，包括 ``EfficientFrontier`` 对象的优化方法所调用的实际目标函数。
这些方法主要是为优化过程中的内部使用而设计的，每个方法都需要不同的参数（这就是为什么它们没有被归入一个类的原因）。
由于显而易见的原因，任何目标函数都必须接受 ``weights`` 作为参数，并且必须至少有 ``expected_returns`` 或 ``cov_matrix`` 中的一个。
目标函数要么计算给定权重的 numpy 数组的目标，要么当权重是 ``cp.Variable`` 时，它返回一个 cvxpy 表达式。
这样一来，同一个目标函数既可以在内部用于优化，也可以在外部用于计算给定权重的目标。 ``_objective_value()`` 会自动在这两种行为之间选择。
``objective_functions`` 默认为用于最小化的优化。
在明显应该最大化的目标的情况下（例如夏普比率，投资组合收益），目标函数实际上返回负数，因为最小化负数等同于最大化正数。
这种行为可由 ``negative=True`` 的可选参数控制。
目前已经实现：
- 投资组合方差（即波动率的平方）。
- 投资组合收益率
- 夏普比率
- L2 正则化（最小化它可以减少非零权重）。
- 二次方效用
- 交易成本模型（简单的模型）
- 事前（平方）跟踪误差
- 事后（平方）跟踪误差
"""
import cvxpy as cp
import numpy as np
def _objective_value(w, obj):
"""
Helper method to return either the value of the objective function
or the objective function as a cvxpy object depending on whether
w is a cvxpy variable or np array.
:param w: weights
:type w: np.ndarray OR cp.Variable
:param obj: objective function expression
:type obj: cp.Expression
:return: value of the objective function OR objective function expression
:rtype: float OR cp.Expression
"""
if isinstance(w, np.ndarray):
if np.isscalar(obj):
return obj
elif np.isscalar(obj.value):
return obj.value
else:
return obj.value.item()
else:
return obj
[文档]
def portfolio_variance(w, cov_matrix):
"""
计算投资组合的总方差（即平方波动率）。
: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 portfolio_return(w, expected_returns, negative=True):
"""
计算一个投资组合的（负）平均收益率
: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: np.ndarray
:param negative: whether quantity should be made negative (so we can minimise)
:type negative: boolean
:return: negative mean return
:rtype: float
"""
sign = -1 if negative else 1
mu = w @ expected_returns
return _objective_value(w, sign * mu)
[文档]
def sharpe_ratio(w, expected_returns, cov_matrix, risk_free_rate=0.02, negative=True):
"""
计算一个投资组合的（负）夏普比率
: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: np.ndarray
:param cov_matrix: covariance matrix
:type cov_matrix: np.ndarray
: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 = (mu - risk_free_rate) / sigma
return _objective_value(w, sign * sharpe)
[文档]
def L2_reg(w, gamma=1):
r"""
L2 正则化，即 :math:`\gamma ||w||^2` ，来增加非零权重的数量。
例如::
ef = EfficientFrontier(mu, S)
ef.add_objective(objective_functions.L2_reg, gamma=2)
ef.min_volatility()
:param w: asset weights in the portfolio
: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 quadratic_utility(w, expected_returns, cov_matrix, risk_aversion, negative=True):
r"""
二次方效用函数，即 :math:`\mu - \frac 1 2 \delta w^T \Sigma w` 。
: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: np.ndarray
:param cov_matrix: covariance matrix
:type cov_matrix: np.ndarray
:param risk_aversion: risk aversion coefficient. Increase to reduce risk.
:type risk_aversion: float
:param negative: whether quantity should be made negative (so we can minimise).
:type negative: boolean
:return: value of the objective function OR objective function expression
:rtype: float OR cp.Expression
"""
sign = -1 if negative else 1
mu = w @ expected_returns
variance = cp.quad_form(w, cov_matrix)
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)
[文档]
def transaction_cost(w, w_prev, k=0.001):
"""
一个非常简单的交易成本模型：将所有的权重变化相加，再乘以一个给定的分数（默认为 10bps）。这模拟了经纪人的固定百分比佣金。
:param w: asset weights in the portfolio
:type w: np.ndarray OR cp.Variable
:param w_prev: previous weights
:type w_prev: np.ndarray
:param k: fractional cost per unit weight exchanged
:type k: float
:return: value of the objective function OR objective function expression
:rtype: float OR cp.Expression
"""
return _objective_value(w, k * cp.norm(w - w_prev, 1))
[文档]
def ex_ante_tracking_error(w, cov_matrix, benchmark_weights):
"""
计算事前跟踪误差的（平方），即 :math:`(w - w_b)^T \\Sigma (w-w_b)` 。
:param w: asset weights in the portfolio
:type w: np.ndarray OR cp.Variable
:param cov_matrix: covariance matrix
:type cov_matrix: np.ndarray
:param benchmark_weights: asset weights in the benchmark
:type benchmark_weights: np.ndarray
:return: value of the objective function OR objective function expression
:rtype: float OR cp.Expression
"""
relative_weights = w - benchmark_weights
tracking_error = cp.quad_form(relative_weights, cov_matrix)
return _objective_value(w, tracking_error)
[文档]
def ex_post_tracking_error(w, historic_returns, benchmark_returns):
"""
计算事后跟踪误差的（平方），即 :math:`Var(r - r_b)` 。
:param w: asset weights in the portfolio
:type w: np.ndarray OR cp.Variable
:param historic_returns: historic asset returns
:type historic_returns: np.ndarray
:param benchmark_returns: historic benchmark returns
:type benchmark_returns: pd.Series or np.ndarray
:return: value of the objective function OR objective function expression
:rtype: float OR cp.Expression
"""
if not isinstance(historic_returns, np.ndarray):
historic_returns = np.array(historic_returns)
if not isinstance(benchmark_returns, np.ndarray):
benchmark_returns = np.array(benchmark_returns)
x_i = w @ historic_returns.T - benchmark_returns
mean = cp.sum(x_i) / len(benchmark_returns)
tracking_error = cp.sum_squares(x_i - mean)
return _objective_value(w, tracking_error)
