"""
``expected_returns`` 模块提供了估算资产期望收益的函数，这是均值-方差优化中的一个必要输入。
按照惯例，这些方法的输出是期望年化收益。在这里假设提供的是每日价格，尽管实际上这些函数与时间频率无关（只要改变 ``frequency`` 参数）。
资产价格必须以 pandas dataframe 的形式提供，按照 :ref:`user-guide` 中描述的那样。
所有的函数都将价格数据处理成百分比收益率数据，然后再计算它们各自对期望收益的估计。
目前已经实现：
- 一般收益模型函数，允许调用一个函数来运行收益模型。
- 历史平均收益率
- 指数加权平均历史收益率
- CAPM 估计收益率
此外，还提供了从收益率转换为价格以及反向转换的函数。
"""
import warnings
import numpy as np
import pandas as pd
def _check_returns(returns):
# Check NaNs excluding leading NaNs
if np.any(np.isnan(returns.mask(returns.ffill().isnull(), 0))):
warnings.warn(
"Some returns are NaN. Please check your price data.", UserWarning
)
if np.any(np.isinf(returns)):
warnings.warn(
"Some returns are infinite. Please check your price data.", UserWarning
)
[文档]
def returns_from_prices(prices, log_returns=False):
"""
计算给定价格的收益率。
:param prices: 调整后的（日频）资产收盘价，每一行是一个日期，每一列是一个股票/ID。
:type prices: pd.DataFrame
:param log_returns: 是否使用对数收益进行计算
:type log_returns: bool, 默认为 False
:return: （日频）收益率
:rtype: pd.DataFrame
"""
if log_returns:
returns = np.log(1 + prices.pct_change()).dropna(how="all")
else:
returns = prices.pct_change().dropna(how="all")
return returns
[文档]
def prices_from_returns(returns, log_returns=False):
"""
计算给定收益率的伪价格。这些不是真正的价格，因为初始价格都被设置为 1，结果传递给任何 PyPortfolioOpt 方法时，行为能表现良好。
:param returns: （日频）资产的百分比收益
:type returns: pd.DataFrame
:param log_returns: 是否使用对数收益进行计算
:type log_returns: bool, 默认为 False
:return: （日频）伪价格
:rtype: pd.DataFrame
"""
if log_returns:
ret = np.exp(returns)
else:
ret = 1 + returns
ret.iloc[0] = 1 # set first day pseudo-price
return ret.cumprod()
def return_model(prices, method="mean_historical_return", **kwargs):
"""
Compute an estimate of future returns, using the return model specified in ``method``.
:param prices: adjusted closing prices of the asset, each row is a date
and each column is a ticker/id.
:type prices: pd.DataFrame
:param returns_data: if true, the first argument is returns instead of prices.
:type returns_data: bool, defaults to False.
:param method: the return model to use. Should be one of:
- ``mean_historical_return``
- ``ema_historical_return``
- ``capm_return``
:type method: str, optional
:raises NotImplementedError: if the supplied method is not recognised
:return: annualised sample covariance matrix
:rtype: pd.DataFrame
"""
if method == "mean_historical_return":
return mean_historical_return(prices, **kwargs)
elif method == "ema_historical_return":
return ema_historical_return(prices, **kwargs)
elif method == "capm_return":
return capm_return(prices, **kwargs)
else:
raise NotImplementedError("Return model {} not implemented".format(method))
[文档]
def mean_historical_return(
prices, returns_data=False, compounding=True, frequency=252, log_returns=False
):
"""
根据输入的（日频）资产价格计算年化平均（日频）历史收益率。使用 ``compounding`` 参数切换使用默认的几何平均数（CAGR）或算术平均数。
:param prices: 调整后的资产收盘价，每一行是一个日期，每一列是一个股票/ID。
:type prices: pd.DataFrame
:param returns_data: 如果是 True, 第一个参数是收益率而不是价格。不应是对数收益率。
:type returns_data: bool, 默认为 False.
:param compounding: 如果为 True 则计算几何平均数，否则为算数平均数，可选。
:type compounding: bool, 默认为 True
:param frequency: 一年中的时间数，默认为252（一年中的交易日数）
:type frequency: int, optional
:param log_returns: 是否使用对数收益率进行计算
:type log_returns: bool, 默认为 False
:return: 每项资产的年化平均（日频）收益率
:rtype: pd.Series
"""
if not isinstance(prices, pd.DataFrame):
warnings.warn("prices are not in a dataframe", RuntimeWarning)
prices = pd.DataFrame(prices)
if returns_data:
returns = prices
else:
returns = returns_from_prices(prices, log_returns)
_check_returns(returns)
if compounding:
return (1 + returns).prod() ** (frequency / returns.count()) - 1
else:
return returns.mean() * frequency
[文档]
def ema_historical_return(
prices,
returns_data=False,
compounding=True,
span=500,
frequency=252,
log_returns=False,
):
"""
计算（日频）历史收益的指数加权平均值，给较近的数据以较高的权重。
:param prices: 调整后的资产收盘价，每一行是一个日期，每一列是一个股票/ID。
:type prices: pd.DataFrame
:param returns_data: 如果是 True, 第一个参数是收益率而不是价格。不应是对数收益率。
:type returns_data: bool, 默认为 False.
:param compounding: 如果为真则计算几何平均数，否则计算算数平均数，可选。
:type compounding: bool, 默认为 True
:param frequency: 一年中的时间段数，默认为252（一年中的交易日数）。
:type frequency: int, optional
:param span: EMA 的时间跨度，默认为 500 天 EMA。
:type span: int, optional
:param log_returns: 是否使用对数收益进行计算
:type log_returns: bool, 默认为 False
:return: 每项资产的年化指数加权平均（日频）收益率
:rtype: pd.Series
"""
if not isinstance(prices, pd.DataFrame):
warnings.warn("prices are not in a dataframe", RuntimeWarning)
prices = pd.DataFrame(prices)
if returns_data:
returns = prices
else:
returns = returns_from_prices(prices, log_returns)
_check_returns(returns)
if compounding:
return (1 + returns.ewm(span=span).mean().iloc[-1]) ** frequency - 1
else:
return returns.ewm(span=span).mean().iloc[-1] * frequency
[文档]
def capm_return(
prices,
market_prices=None,
returns_data=False,
risk_free_rate=0.02,
compounding=True,
frequency=252,
log_returns=False,
):
"""
使用资本资产定价模型计算收益估计。在 CAPM 下，资产收益等于市场收益加乘以 β。
.. math::
R_i = R_f + \\beta_i (E(R_m) - R_f)
:param prices: 调整后的资产收盘价，每一行是一个日期，每一列是一个股票/ID。
:type prices: pd.DataFrame
:param market_prices: 基准的调整后收盘价，默认为 None
:type market_prices: pd.DataFrame, optional
:param returns_data: 如果是 True, 第一个参数是收益率而不是价格
:type returns_data: bool, 默认为 False.
:param risk_free_rate: 借贷的无风险利率，默认为 0.02。应该使用适当的时间段，与频率参数相对应。
:type risk_free_rate: float, optional
:param compounding: 如果为真则计算几何平均数，否则计算算数平均数，可选。
:type compounding: bool, 默认为 True
:param frequency: 一年中的时间段数，默认为252（一年中的交易日数）。
:type frequency: int, optional
:param log_returns: 是否使用对数收益进行计算
:type log_returns: bool, 默认为 False
:return: 年化收益率估计
:rtype: pd.Series
"""
if not isinstance(prices, pd.DataFrame):
warnings.warn("prices are not in a dataframe", RuntimeWarning)
prices = pd.DataFrame(prices)
market_returns = None
if returns_data:
returns = prices.copy()
if market_prices is not None:
market_returns = market_prices
else:
returns = returns_from_prices(prices, log_returns)
if market_prices is not None:
if not isinstance(market_prices, pd.DataFrame):
warnings.warn("market prices are not in a dataframe", RuntimeWarning)
market_prices = pd.DataFrame(market_prices)
market_returns = returns_from_prices(market_prices, log_returns)
# Use the equally-weighted dataset as a proxy for the market
if market_returns is None:
# Append market return to right and compute sample covariance matrix
returns["mkt"] = returns.mean(axis=1)
else:
market_returns.columns = ["mkt"]
returns = returns.join(market_returns, how="left")
_check_returns(returns)
# Compute covariance matrix for the new dataframe (including markets)
cov = returns.cov()
# The far-right column of the cov matrix is covariances to market
betas = cov["mkt"] / cov.loc["mkt", "mkt"]
betas = betas.drop("mkt")
# Find mean market return on a given time period
if compounding:
mkt_mean_ret = (1 + returns["mkt"]).prod() ** (
frequency / returns["mkt"].count()
) - 1
else:
mkt_mean_ret = returns["mkt"].mean() * frequency
# CAPM formula
return risk_free_rate + betas * (mkt_mean_ret - risk_free_rate)
