jerry/OpenBB
1
0
Fork 0
mirror of https://github.com/OpenBB-finance/OpenBB.git synced 2026-11-15 13:51:19 +00:00
Code Issues Packages Projects Releases Wiki Activity

Portfolio optimization bug fixes (#3675)

Browse Source 
* remove hcp

* add prams dict to statics

* little controller bug

* fix view bug

* avoid crash if one stock

* check params on setup

* create parameter class

* check and convert parameters using class

* change params name to avoid confusion

* create a parameter statics dict

* change some funcs names

* remove unused imports

* remove default dict

* optional type

* add multi choices

* cast only int and float

* fix completer

* fix bugs with parameter validation

* fix bugs with parameter validation

* add excel formatting

* sdk needs mapping as well, controller takes care of this in terminal

* small formating

* add some safe guard try except

* controller bugs

* oops

* change export path of parameters to portfolio folder

* add more commands to scripts

* catch optimization exception

* log errors

* black and exceptions

* add flag to test

* black

* flake8

* forgot this

* pylint

* change defaults

* fix ef default

* fix bl

* sync sdk defaults

* sync last defaults

* fix plot heat and add more choices to controller autocomplete

* patch weights

* fix wrong bool parsing

Co-authored-by: James Maslek <jmaslek11@gmail.com>
This commit is contained in:
montezdesousa
2022-12-02 23:56:41 +00:00
committed by GitHub
parent f47d15088f
commit faca7ab67d
16 changed files with 1351 additions and 677 deletions
Download Patch File Download Diff File Expand all files Collapse all files

BIN
openbb_terminal/miscellaneous/portfolio_examples/optimization/OpenBB_Parameters_Template_v1.0.0.xlsx vendored
View File

Binary file not shown.

1
openbb_terminal/miscellaneous/scripts/portfolio/test_po_show_plot_rpf.openbb vendored
View File

@@ -52,3 +52,4 @@ plot -pf name_0 -ct sector -he
maxsharpe
rpf -pf a,b,c
rpf -pf name_0,maxsharpe_1
exit

5
openbb_terminal/miscellaneous/scripts/portfolio/test_portfolio_boolean_args.openbb vendored
View File

@@ -2,8 +2,9 @@ portfolio
load --file Public_Equity_Orderbook.xlsx
po
params
file --file OpenBB_Parameters_Template_v1.0.0.xlsx
load OpenBB_Parameters_Template_v1.0.0.xlsx
set ef
arg tangency False
arg log_returns True
exit
clear
exit

22
openbb_terminal/miscellaneous/scripts/portfolio/test_portfolio_po.openbb vendored
View File

@@ -1,13 +1,14 @@
portfolio
po
load 60_40_Portfolio.xlsx
file --file example.ini
maxsharpe -s 2022-10-20
minrisk -r 0.05
maxutil -a 0.1
maxutil -a 0.1 -ra 2 -tk 0.4 -tr 8
maxret -e 2022-11-20
maxdiv -mn 0.1
maxdecorr -lr
ef -t
ef -t -se 69 -n 10
equal -th 0.1
riskparity -p 1y
relriskparity -de 0.9
@@ -21,4 +22,21 @@ plot -pf MAXSHARPE_0 -dd
plot -pf MAXSHARPE_0 -rc
plot -pf MAXSHARPE_0 -he
rpf NAME_6
..
load --file market.csv
po
equal
maxsharpe -s 1995-05-07
minrisk -r 0.08
maxutil -a 0.1 -ra 2 -tk 0.4 -tr 8
maxret -e 2022-11-20
maxdiv -mn 0.1
maxdecorr -lr -p max
ef -t -se 420 -n 0
equal -th 0.1
riskparity -p 1y
relriskparity -de 0.9
hrp -cd pearson
herc -at 0.1
nco -rm EDaR -lr
exit

48
openbb_terminal/portfolio/portfolio_optimization/optimizer_helper.py
View File

@@ -2,9 +2,14 @@
__docformat__ = "numpy"
import argparse
from typing import Any
import pandas as pd
from openbb_terminal.portfolio.portfolio_optimization import statics
from openbb_terminal.portfolio.portfolio_optimization.statics import (
RISK_CHOICES,
OPTIMIZATION_PARAMETERS,
TERMINAL_TEMPLATE_MAP,
)
from openbb_terminal.rich_config import console
# These are all the possible yfinance properties
@@ -146,8 +151,45 @@ def validate_risk_measure(risk_measure: str, warning: bool = True) -> str:
str
Validated risk measure
"""
if risk_measure.lower() in statics.RISK_CHOICES:
return statics.RISK_CHOICES[risk_measure.lower()]
if risk_measure.lower() in RISK_CHOICES:
return RISK_CHOICES[risk_measure.lower()]
if warning:
console.print("[yellow]Risk measure not found. Using 'MV'.[/yellow]")
return "MV"
def get_kwarg(key: str, kwargs: dict, default: Any = None) -> Any:
"""Get a key from kwargs
If key is in kwargs, returns it.
Otherwise, if default provided, returns it.
Otherwise, if key is in OPTIMIZATION_PARAMETERS, returns it.
Parameters
----------
key : str
The key to be searched
kwargs : dict
The kwargs to be searched
default : Any
The default value to be returned if the key is not found
Returns
-------
Any
The value of the key if it exists, else None
"""
if key in kwargs:
return kwargs[key]
if default:
return default
# TODO: Remove this line when mapping between template and terminal is not needed
template_key = TERMINAL_TEMPLATE_MAP.get(key, key)
PARAMETER = OPTIMIZATION_PARAMETERS.get(template_key)
if PARAMETER is None:
return default
return PARAMETER.default

721
openbb_terminal/portfolio/portfolio_optimization/optimizer_model.py
View File

@@ -21,7 +21,11 @@ from scipy.interpolate import interp1d
from openbb_terminal.decorators import log_start_end
from openbb_terminal.portfolio.portfolio_optimization import (
yahoo_finance_model,
optimizer_helper,
)
from openbb_terminal.portfolio.portfolio_optimization.optimizer_helper import (
get_kwarg,
validate_risk_measure,
valid_property_infos,
)
from openbb_terminal.rich_config import console
@@ -199,15 +203,15 @@ def get_equal_weights(
Dictionary of weights where keys are the tickers, dataframe of stock returns
"""
interval = kwargs.get("interval", "3y")
start_date = kwargs.get("start_date", "")
end_date = kwargs.get("end_date", "")
log_returns = kwargs.get("log_returns", False)
freq = kwargs.get("freq", "D")
maxnan = kwargs.get("maxnan", 0.05)
threshold = kwargs.get("threshold", 0.0)
method = kwargs.get("method", "time")
value = kwargs.get("value", 1.0)
interval = get_kwarg("interval", kwargs)
start_date = get_kwarg("start_date", kwargs)
end_date = get_kwarg("end_date", kwargs)
log_returns = get_kwarg("log_returns", kwargs)
freq = get_kwarg("freq", kwargs)
maxnan = get_kwarg("maxnan", kwargs)
threshold = get_kwarg("threshold", kwargs)
method = get_kwarg("method", kwargs)
value = get_kwarg("value", kwargs)
stock_prices = yahoo_finance_model.process_stocks(
symbols, interval, start_date, end_date
@@ -271,17 +275,17 @@ def get_property_weights(
Dictionary of portfolio weights or allocations
"""
interval = kwargs.get("interval", "3y")
start_date = kwargs.get("start_date", "")
end_date = kwargs.get("end_date", "")
log_returns = kwargs.get("log_returns", False)
freq = kwargs.get("freq", "D")
maxnan = kwargs.get("maxnan", 0.05)
threshold = kwargs.get("threshold", 0.0)
method = kwargs.get("method", "time")
value = kwargs.get("value", 1.0)
interval = get_kwarg("interval", kwargs)
start_date = get_kwarg("start_date", kwargs)
end_date = get_kwarg("end_date", kwargs)
log_returns = get_kwarg("log_returns", kwargs)
freq = get_kwarg("freq", kwargs)
maxnan = get_kwarg("maxnan", kwargs)
threshold = get_kwarg("threshold", kwargs)
method = get_kwarg("method", kwargs)
value = get_kwarg("value", kwargs)
s_property = kwargs.get("s_property", "marketCap")
s_property = get_kwarg("s_property", kwargs, default="marketCap")
stock_prices = yahoo_finance_model.process_stocks(
symbols, interval, start_date, end_date
@@ -427,29 +431,29 @@ def get_mean_risk_portfolio(
DataFrame of stock returns.
"""
interval = kwargs.get("interval", "3y")
start_date = kwargs.get("start_date", "")
end_date = kwargs.get("end_date", "")
log_returns = kwargs.get("log_returns", False)
freq = kwargs.get("freq", "D")
maxnan = kwargs.get("maxnan", 0.05)
threshold = kwargs.get("threshold", 0.0)
method = kwargs.get("method", "time")
value = kwargs.get("value", 1.0)
value_short = kwargs.get("value_short", 0.0)
interval = get_kwarg("interval", kwargs)
start_date = get_kwarg("start_date", kwargs)
end_date = get_kwarg("end_date", kwargs)
log_returns = get_kwarg("log_returns", kwargs)
freq = get_kwarg("freq", kwargs)
maxnan = get_kwarg("maxnan", kwargs)
threshold = get_kwarg("threshold", kwargs)
method = get_kwarg("method", kwargs)
value = get_kwarg("value", kwargs)
value_short = get_kwarg("value_short", kwargs)
risk_measure = kwargs.get("risk_measure", "MV")
objective = kwargs.get("objective", "Sharpe")
risk_free_rate = kwargs.get("risk_free_rate", 0.0)
risk_aversion = kwargs.get("risk_aversion", 1.0)
alpha = kwargs.get("alpha", 0.05)
target_return = kwargs.get("target_return", -1.0)
target_risk = kwargs.get("target_risk", -1.0)
mean = kwargs.get("mean", "hist")
covariance = kwargs.get("covariance", "hist")
d_ewma = kwargs.get("d_ewma", 0.94)
risk_measure = get_kwarg("risk_measure", kwargs)
objective = get_kwarg("objective", kwargs)
risk_free_rate = get_kwarg("risk_free_rate", kwargs)
risk_aversion = get_kwarg("risk_aversion", kwargs)
alpha = get_kwarg("alpha", kwargs)
target_return = get_kwarg("target_return", kwargs)
target_risk = get_kwarg("target_risk", kwargs)
mean = get_kwarg("mean", kwargs)
covariance = get_kwarg("covariance", kwargs)
d_ewma = get_kwarg("d_ewma", kwargs)
risk_measure = optimizer_helper.validate_risk_measure(risk_measure)
risk_measure = validate_risk_measure(risk_measure)
stock_prices = yahoo_finance_model.process_stocks(
symbols, interval, start_date, end_date
@@ -468,48 +472,64 @@ def get_mean_risk_portfolio(
)
return {}, pd.DataFrame()
if stock_returns.shape[1] < 2:
console.print(
f"[red]Given the parameters could only get data for '{stock_returns.columns[0]}'.[/red]\n"
"[red]Optimization needs at least two assets.[/red]\n",
)
return {}, pd.DataFrame()
first_day = stock_returns.index[0].strftime("%Y-%m-%d")
console.print(
f"[yellow]First day of data respecting parameters: {first_day}[/yellow]\n"
)
risk_free_rate = risk_free_rate / time_factor[freq.upper()]
# Building the portfolio object
port = rp.Portfolio(returns=stock_returns, alpha=alpha)
try:
# Building the portfolio object
port = rp.Portfolio(returns=stock_returns, alpha=alpha)
# Estimate input parameters:
port.assets_stats(method_mu=mean, method_cov=covariance, d=d_ewma)
# Estimate input parameters:
port.assets_stats(method_mu=mean, method_cov=covariance, d=d_ewma)
# Budget constraints
port.upperlng = value
if value_short > 0:
port.sht = True
port.uppersht = value_short
port.budget = value - value_short
else:
port.budget = value
# Estimate optimal portfolio:
model = "Classic"
hist = True
if target_return > -1:
port.lowerret = float(target_return) / time_factor[freq.upper()]
if target_risk > -1:
if risk_measure not in ["ADD", "MDD", "CDaR", "EDaR", "UCI"]:
setattr(
port,
upper_risk[risk_measure],
float(target_risk) / time_factor[freq.upper()] ** 0.5,
)
# Budget constraints
port.upperlng = value
if value_short > 0:
port.sht = True
port.uppersht = value_short
port.budget = value - value_short
else:
setattr(port, upper_risk[risk_measure], float(target_risk))
port.budget = value
weights = port.optimization(
model=model,
rm=risk_measure,
obj=objective,
rf=risk_free_rate,
l=risk_aversion,
hist=hist,
)
# Estimate optimal portfolio:
model = "Classic"
hist = True
if target_return > -1:
port.lowerret = float(target_return) / time_factor[freq.upper()]
if target_risk > -1:
if risk_measure not in ["ADD", "MDD", "CDaR", "EDaR", "UCI"]:
setattr(
port,
upper_risk[risk_measure],
float(target_risk) / time_factor[freq.upper()] ** 0.5,
)
else:
setattr(port, upper_risk[risk_measure], float(target_risk))
weights = port.optimization(
model=model,
rm=risk_measure,
obj=objective,
rf=risk_free_rate,
l=risk_aversion,
hist=hist,
)
except Exception as _:
weights = None
if weights is not None:
weights = weights.round(5)
@@ -1032,19 +1052,19 @@ def get_max_diversification_portfolio(
DataFrame of stock returns.
"""
interval = kwargs.get("interval", "3y")
start_date = kwargs.get("start_date", "")
end_date = kwargs.get("end_date", "")
log_returns = kwargs.get("log_returns", False)
freq = kwargs.get("freq", "D")
maxnan = kwargs.get("maxnan", 0.05)
threshold = kwargs.get("threshold", 0.0)
method = kwargs.get("method", "time")
value = kwargs.get("value", 1.0)
value_short = kwargs.get("value_short", 0.0)
interval = get_kwarg("interval", kwargs)
start_date = get_kwarg("start_date", kwargs)
end_date = get_kwarg("end_date", kwargs)
log_returns = get_kwarg("log_returns", kwargs)
freq = get_kwarg("freq", kwargs)
maxnan = get_kwarg("maxnan", kwargs)
threshold = get_kwarg("threshold", kwargs)
method = get_kwarg("method", kwargs)
value = get_kwarg("value", kwargs)
value_short = get_kwarg("value_short", kwargs)
covariance = kwargs.get("covariance", "hist")
d_ewma = kwargs.get("d_ewma", 0.94)
covariance = get_kwarg("covariance", kwargs)
d_ewma = get_kwarg("d_ewma", kwargs)
stock_prices = yahoo_finance_model.process_stocks(
symbols, interval, start_date, end_date
@@ -1058,24 +1078,29 @@ def get_max_diversification_portfolio(
method=method,
)
# Building the portfolio object
port = rp.Portfolio(returns=stock_returns)
try:
# Building the portfolio object
port = rp.Portfolio(returns=stock_returns)
# Estimate input parameters:
port.assets_stats(method_mu="hist", method_cov=covariance, d=d_ewma)
port.mu = stock_returns.std().to_frame().T
# Estimate input parameters:
port.assets_stats(method_mu="hist", method_cov=covariance, d=d_ewma)
port.mu = stock_returns.std().to_frame().T
# Budget constraints
port.upperlng = value
if value_short > 0:
port.sht = True
port.uppersht = value_short
port.budget = value - value_short
else:
port.budget = value
# Budget constraints
port.upperlng = value
if value_short > 0:
port.sht = True
port.uppersht = value_short
port.budget = value - value_short
else:
port.budget = value
# Estimate optimal portfolio:
weights = port.optimization(model="Classic", rm="MV", obj="Sharpe", rf=0, hist=True)
# Estimate optimal portfolio:
weights = port.optimization(
model="Classic", rm="MV", obj="Sharpe", rf=0, hist=True
)
except Exception as _:
weights = None
if weights is not None:
weights = weights.round(5)
@@ -1151,19 +1176,19 @@ def get_max_decorrelation_portfolio(
DataFrame of stock returns.
"""
interval = kwargs.get("interval", "3y")
start_date = kwargs.get("start_date", "")
end_date = kwargs.get("end_date", "")
log_returns = kwargs.get("log_returns", False)
freq = kwargs.get("freq", "D")
maxnan = kwargs.get("maxnan", 0.05)
threshold = kwargs.get("threshold", 0.0)
method = kwargs.get("method", "time")
value = kwargs.get("value", 1.0)
value_short = kwargs.get("value_short", 0.0)
interval = get_kwarg("interval", kwargs)
start_date = get_kwarg("start_date", kwargs)
end_date = get_kwarg("end_date", kwargs)
log_returns = get_kwarg("log_returns", kwargs)
freq = get_kwarg("freq", kwargs)
maxnan = get_kwarg("maxnan", kwargs)
threshold = get_kwarg("threshold", kwargs)
method = get_kwarg("method", kwargs)
value = get_kwarg("value", kwargs)
value_short = get_kwarg("value_short", kwargs)
covariance = kwargs.get("covariance", "hist")
d_ewma = kwargs.get("d_ewma", 0.94)
covariance = get_kwarg("covariance", kwargs)
d_ewma = get_kwarg("d_ewma", kwargs)
stock_prices = yahoo_finance_model.process_stocks(
symbols, interval, start_date, end_date
@@ -1177,26 +1202,30 @@ def get_max_decorrelation_portfolio(
method=method,
)
# Building the portfolio object
port = rp.Portfolio(returns=stock_returns)
try:
# Estimate input parameters:
port.assets_stats(method_mu="hist", method_cov=covariance, d=d_ewma)
port.cov = rp.cov2corr(port.cov)
# Building the portfolio object
port = rp.Portfolio(returns=stock_returns)
# Budget constraints
port.upperlng = value
if value_short > 0:
port.sht = True
port.uppersht = value_short
port.budget = value - value_short
else:
port.budget = value
# Estimate input parameters:
port.assets_stats(method_mu="hist", method_cov=covariance, d=d_ewma)
port.cov = rp.cov2corr(port.cov)
# Estimate optimal portfolio:
weights = port.optimization(
model="Classic", rm="MV", obj="MinRisk", rf=0, hist=True
)
# Budget constraints
port.upperlng = value
if value_short > 0:
port.sht = True
port.uppersht = value_short
port.budget = value - value_short
else:
port.budget = value
# Estimate optimal portfolio:
weights = port.optimization(
model="Classic", rm="MV", obj="MinRisk", rf=0, hist=True
)
except Exception as _:
weights = None
if weights is not None:
weights = weights.round(5)
@@ -1284,26 +1313,26 @@ def get_black_litterman_portfolio(
DataFrame of stock returns.
"""
interval = kwargs.get("interval", "3y")
start_date = kwargs.get("start_date", "")
end_date = kwargs.get("end_date", "")
log_returns = kwargs.get("log_returns", False)
freq = kwargs.get("freq", "D")
maxnan = kwargs.get("maxnan", 0.05)
threshold = kwargs.get("threshold", 0.0)
method = kwargs.get("method", "time")
value = kwargs.get("value", 1.0)
value_short = kwargs.get("value_short", 0.0)
interval = get_kwarg("interval", kwargs)
start_date = get_kwarg("start_date", kwargs)
end_date = get_kwarg("end_date", kwargs)
log_returns = get_kwarg("log_returns", kwargs)
freq = get_kwarg("freq", kwargs)
maxnan = get_kwarg("maxnan", kwargs)
threshold = get_kwarg("threshold", kwargs)
method = get_kwarg("method", kwargs)
value = get_kwarg("value", kwargs)
value_short = get_kwarg("value_short", kwargs)
benchmark = kwargs.get("benchmark", None)
p_views = kwargs.get("p_views", None)
q_views = kwargs.get("q_views", None)
objective = kwargs.get("objective", "Sharpe")
risk_free_rate = kwargs.get("risk_free_rate", 0)
risk_aversion = kwargs.get("risk_aversion", 1)
delta = kwargs.get("delta", None)
equilibrium = kwargs.get("equilibrium", True)
optimize = kwargs.get("optimize", True)
benchmark = get_kwarg("benchmark", kwargs)
p_views = get_kwarg("p_views", kwargs)
q_views = get_kwarg("q_views", kwargs)
objective = get_kwarg("objective", kwargs)
risk_free_rate = get_kwarg("risk_free_rate", kwargs)
risk_aversion = get_kwarg("risk_aversion", kwargs)
delta = get_kwarg("delta", kwargs)
equilibrium = get_kwarg("equilibrium", kwargs)
optimize = get_kwarg("optimize", kwargs)
stock_prices = yahoo_finance_model.process_stocks(
symbols, interval, start_date, end_date
@@ -1349,32 +1378,35 @@ def get_black_litterman_portfolio(
weights = pd.DataFrame(weights)
if optimize:
# Building the portfolio object
port = rp.Portfolio(returns=stock_returns)
try:
# Building the portfolio object
port = rp.Portfolio(returns=stock_returns)
# Estimate input parameters:
port.assets_stats(method_mu="hist", method_cov="hist")
port.mu_bl = pd.DataFrame(mu).T
port.cov_bl = pd.DataFrame(cov)
# Estimate input parameters:
port.assets_stats(method_mu="hist", method_cov="hist")
port.mu_bl = pd.DataFrame(mu).T
port.cov_bl = pd.DataFrame(cov)
# Budget constraints
port.upperlng = value
if value_short > 0:
port.sht = True
port.uppersht = value_short
port.budget = value - value_short
else:
port.budget = value
# Budget constraints
port.upperlng = value
if value_short > 0:
port.sht = True
port.uppersht = value_short
port.budget = value - value_short
else:
port.budget = value
# Estimate optimal portfolio:
weights = port.optimization(
model="BL",
rm="MV",
obj=objective,
rf=risk_free_rate,
l=risk_aversion,
hist=True,
)
# Estimate optimal portfolio:
weights = port.optimization(
model="BL",
rm="MV",
obj=objective,
rf=risk_free_rate,
l=risk_aversion,
hist=True,
)
except Exception as _:
weights = None
if weights is not None:
weights = weights.round(5)
@@ -1478,25 +1510,25 @@ def get_ef(
frontier, mu, cov, stock_returns, weights, X1, Y1, port
"""
interval = kwargs.get("interval", "3y")
start_date = kwargs.get("start_date", "")
end_date = kwargs.get("end_date", "")
log_returns = kwargs.get("log_returns", False)
freq = kwargs.get("freq", "D")
maxnan = kwargs.get("maxnan", 0.05)
threshold = kwargs.get("threshold", 0.05)
method = kwargs.get("method", "time")
value = kwargs.get("value", 1.0)
value_short = kwargs.get("value_short", 0.0)
interval = get_kwarg("interval", kwargs)
start_date = get_kwarg("start_date", kwargs)
end_date = get_kwarg("end_date", kwargs)
log_returns = get_kwarg("log_returns", kwargs)
freq = get_kwarg("freq", kwargs)
maxnan = get_kwarg("maxnan", kwargs)
threshold = get_kwarg("threshold", kwargs)
method = get_kwarg("method", kwargs)
value = get_kwarg("value", kwargs)
value_short = get_kwarg("value_short", kwargs)
risk_measure = kwargs.get("risk_measure", "MV")
risk_free_rate = kwargs.get("risk_free_rate", 0.0)
alpha = kwargs.get("alpha", 0.05)
n_portfolios = kwargs.get("n_portfolios", 100)
seed = kwargs.get("seed", 123)
risk_measure = get_kwarg("risk_measure", kwargs)
risk_free_rate = get_kwarg("risk_free_rate", kwargs)
alpha = get_kwarg("alpha", kwargs)
n_portfolios = get_kwarg("n_portfolios", kwargs)
seed = get_kwarg("seed", kwargs)
risk_free_rate = risk_free_rate / time_factor[freq.upper()]
risk_measure = optimizer_helper.validate_risk_measure(risk_measure)
risk_measure = validate_risk_measure(risk_measure)
stock_prices = yahoo_finance_model.process_stocks(
symbols, interval, start_date, end_date
@@ -1510,29 +1542,32 @@ def get_ef(
method=method,
)
# Building the portfolio object
port = rp.Portfolio(returns=stock_returns, alpha=alpha)
try:
# Building the portfolio object
port = rp.Portfolio(returns=stock_returns, alpha=alpha)
# Estimate input parameters:
port.assets_stats(method_mu="hist", method_cov="hist")
# Estimate input parameters:
port.assets_stats(method_mu="hist", method_cov="hist")
# Budget constraints
port.upperlng = value
if value_short > 0:
port.sht = True
port.uppersht = value_short
port.budget = value - value_short
else:
port.budget = value
# Budget constraints
port.upperlng = value
if value_short > 0:
port.sht = True
port.uppersht = value_short
port.budget = value - value_short
else:
port.budget = value
# Estimate tangency portfolio:
weights: Optional[pd.DataFrame] = port.optimization(
model="Classic",
rm=risk_choices[risk_measure.lower()],
obj="Sharpe",
rf=risk_free_rate,
hist=True,
)
# Estimate tangency portfolio:
weights: Optional[pd.DataFrame] = port.optimization(
model="Classic",
rm=risk_choices[risk_measure.lower()],
obj="Sharpe",
rf=risk_free_rate,
hist=True,
)
except Exception as _:
weights = None
points = 20 # Number of points of the frontier
frontier = port.efficient_frontier(
@@ -1684,24 +1719,24 @@ def get_risk_parity_portfolio(
DataFrame of stock returns.
"""
interval = kwargs.get("interval", "3y")
start_date = kwargs.get("start_date", "")
end_date = kwargs.get("end_date", "")
log_returns = kwargs.get("log_returns", False)
freq = kwargs.get("freq", "D")
maxnan = kwargs.get("maxnan", 0.05)
threshold = kwargs.get("threshold", 0.05)
method = kwargs.get("method", "time")
value = kwargs.get("value", 1)
interval = get_kwarg("interval", kwargs)
start_date = get_kwarg("start_date", kwargs)
end_date = get_kwarg("end_date", kwargs)
log_returns = get_kwarg("log_returns", kwargs)
freq = get_kwarg("freq", kwargs)
maxnan = get_kwarg("maxnan", kwargs)
threshold = get_kwarg("threshold", kwargs)
method = get_kwarg("method", kwargs)
value = get_kwarg("value", kwargs)
risk_measure = kwargs.get("risk_measure", "MV")
risk_free_rate = kwargs.get("risk_free_rate", 0)
alpha = kwargs.get("alpha", 0.05)
target_return = kwargs.get("target_return", -1.0)
mean = kwargs.get("mean", "hist")
covariance = kwargs.get("covariance", "hist")
d_ewma = kwargs.get("d_ewma", 0.94)
risk_cont = kwargs.get("risk_cont", None)
risk_measure = get_kwarg("risk_measure", kwargs)
risk_free_rate = get_kwarg("risk_free_rate", kwargs)
alpha = get_kwarg("alpha", kwargs)
target_return = get_kwarg("target_return", kwargs)
mean = get_kwarg("mean", kwargs)
covariance = get_kwarg("covariance", kwargs)
d_ewma = get_kwarg("d_ewma", kwargs)
risk_cont = get_kwarg("risk_cont", kwargs)
stock_prices = yahoo_finance_model.process_stocks(
symbols, interval, start_date, end_date
@@ -1716,30 +1751,33 @@ def get_risk_parity_portfolio(
)
risk_free_rate = risk_free_rate / time_factor[freq.upper()]
risk_measure = optimizer_helper.validate_risk_measure(risk_measure)
risk_measure = validate_risk_measure(risk_measure)
# Building the portfolio object
port = rp.Portfolio(returns=stock_returns, alpha=alpha)
try:
# Building the portfolio object
port = rp.Portfolio(returns=stock_returns, alpha=alpha)
# Calculating optimal portfolio
port.assets_stats(method_mu=mean, method_cov=covariance, d=d_ewma)
# Calculating optimal portfolio
port.assets_stats(method_mu=mean, method_cov=covariance, d=d_ewma)
# Estimate optimal portfolio:
model = "Classic"
hist = True
# Estimate optimal portfolio:
model = "Classic"
hist = True
if risk_cont is None:
risk_cont_ = None # Risk contribution constraints vector
else:
risk_cont_ = np.array(risk_cont).reshape(1, -1)
risk_cont_ = risk_cont_ / np.sum(risk_cont_)
if risk_cont is None:
risk_cont_ = None # Risk contribution constraints vector
else:
risk_cont_ = np.array(risk_cont).reshape(1, -1)
risk_cont_ = risk_cont_ / np.sum(risk_cont_)
if target_return > -1:
port.lowerret = float(target_return) / time_factor[freq.upper()]
if target_return > -1:
port.lowerret = float(target_return) / time_factor[freq.upper()]
weights = port.rp_optimization(
model=model, rm=risk_measure, rf=risk_free_rate, b=risk_cont_, hist=hist
)
weights = port.rp_optimization(
model=model, rm=risk_measure, rf=risk_free_rate, b=risk_cont_, hist=hist
)
except Exception as _:
weights = None
if weights is not None:
if value > 0.0:
@@ -1839,23 +1877,23 @@ def get_rel_risk_parity_portfolio(
DataFrame of stock returns.
"""
interval = kwargs.get("interval", "3y")
start_date = kwargs.get("start_date", "")
end_date = kwargs.get("end_date", "")
log_returns = kwargs.get("log_returns", False)
freq = kwargs.get("freq", "D")
maxnan = kwargs.get("maxnan", 0.05)
threshold = kwargs.get("threshold", 0.05)
method = kwargs.get("method", "time")
value = kwargs.get("value", 1)
interval = get_kwarg("interval", kwargs)
start_date = get_kwarg("start_date", kwargs)
end_date = get_kwarg("end_date", kwargs)
log_returns = get_kwarg("log_returns", kwargs)
freq = get_kwarg("freq", kwargs)
maxnan = get_kwarg("maxnan", kwargs)
threshold = get_kwarg("threshold", kwargs)
method = get_kwarg("method", kwargs)
value = get_kwarg("value", kwargs)
target_return = kwargs.get("target_return", -1)
mean = kwargs.get("mean", "hist")
covariance = kwargs.get("covariance", "hist")
d_ewma = kwargs.get("d_ewma", 0.94)
risk_cont = kwargs.get("risk_cont", None)
version = kwargs.get("version", "A")
penal_factor = kwargs.get("penal_factor", 1)
target_return = get_kwarg("target_return", kwargs)
mean = get_kwarg("mean", kwargs)
covariance = get_kwarg("covariance", kwargs)
d_ewma = get_kwarg("d_ewma", kwargs)
risk_cont = get_kwarg("risk_cont", kwargs)
version = get_kwarg("version", kwargs)
penal_factor = get_kwarg("penal_factor", kwargs)
stock_prices = yahoo_finance_model.process_stocks(
symbols, interval, start_date, end_date
@@ -1869,28 +1907,31 @@ def get_rel_risk_parity_portfolio(
method=method,
)
# Building the portfolio object
port = rp.Portfolio(returns=stock_returns)
try:
# Building the portfolio object
port = rp.Portfolio(returns=stock_returns)
# Calculating optimal portfolio
port.assets_stats(method_mu=mean, method_cov=covariance, d=d_ewma)
# Calculating optimal portfolio
port.assets_stats(method_mu=mean, method_cov=covariance, d=d_ewma)
# Estimate optimal portfolio:
model = "Classic"
hist = True
# Estimate optimal portfolio:
model = "Classic"
hist = True
if risk_cont is None:
risk_cont_ = None # Risk contribution constraints vector
else:
risk_cont_ = np.array(risk_cont).reshape(1, -1)
risk_cont_ = risk_cont_ / np.sum(risk_cont_)
if risk_cont is None:
risk_cont_ = None # Risk contribution constraints vector
else:
risk_cont_ = np.array(risk_cont).reshape(1, -1)
risk_cont_ = risk_cont_ / np.sum(risk_cont_)
if target_return > -1:
port.lowerret = float(target_return) / time_factor[freq.upper()]
if target_return > -1:
port.lowerret = float(target_return) / time_factor[freq.upper()]
weights = port.rrp_optimization(
model=model, version=version, l=penal_factor, b=risk_cont_, hist=hist
)
weights = port.rrp_optimization(
model=model, version=version, l=penal_factor, b=risk_cont_, hist=hist
)
except Exception as _:
weights = None
if weights is not None:
if value > 0.0:
@@ -2084,38 +2125,38 @@ def get_hcp_portfolio(
DataFrame of stock returns.
"""
interval = kwargs.get("interval", "3y")
start_date = kwargs.get("start_date", "")
end_date = kwargs.get("end_date", "")
log_returns = kwargs.get("log_returns", False)
freq = kwargs.get("freq", "D")
maxnan = kwargs.get("maxnan", 0.05)
threshold = kwargs.get("threshold", 0.0)
method = kwargs.get("method", "time")
value = kwargs.get("value", 1.0)
interval = get_kwarg("interval", kwargs)
start_date = get_kwarg("start_date", kwargs)
end_date = get_kwarg("end_date", kwargs)
log_returns = get_kwarg("log_returns", kwargs)
freq = get_kwarg("freq", kwargs)
maxnan = get_kwarg("maxnan", kwargs)
threshold = get_kwarg("threshold", kwargs)
method = get_kwarg("method", kwargs)
value = get_kwarg("value", kwargs)
objective = kwargs.get("objective", "MinRisk")
risk_measure = kwargs.get("risk_measure", "MV")
risk_free_rate = kwargs.get("risk_free_rate", 0.0)
risk_aversion = kwargs.get("risk_aversion", 1.0)
alpha = kwargs.get("alpha", 0.05)
a_sim = kwargs.get("a_sim", 100)
beta = kwargs.get("beta", None)
b_sim = kwargs.get("b_sim", None)
covariance = kwargs.get("covariance", "hist")
d_ewma = kwargs.get("d_ewma", 0.94)
objective = get_kwarg("objective", kwargs)
risk_measure = get_kwarg("risk_measure", kwargs)
risk_free_rate = get_kwarg("risk_free_rate", kwargs)
risk_aversion = get_kwarg("risk_aversion", kwargs)
alpha = get_kwarg("alpha", kwargs)
a_sim = get_kwarg("a_sim", kwargs)
beta = get_kwarg("beta", kwargs)
b_sim = get_kwarg("b_sim", kwargs)
covariance = get_kwarg("covariance", kwargs)
d_ewma = get_kwarg("d_ewma", kwargs)
model = kwargs.get("model", "HRP")
model = get_kwarg("model", kwargs, default="HRP")
codependence = kwargs.get("codependence", "pearson")
linkage = kwargs.get("linkage", "single")
k = kwargs.get("k", None)
max_k = kwargs.get("max_k", 10)
bins_info = kwargs.get("bins_info", "KN")
alpha_tail = kwargs.get("alpha_tail", 0.05)
leaf_order = kwargs.get("leaf_order", True)
codependence = get_kwarg("codependence", kwargs)
linkage = get_kwarg("linkage", kwargs)
k = get_kwarg("k", kwargs)
max_k = get_kwarg("max_k", kwargs)
bins_info = get_kwarg("bins_info", kwargs)
alpha_tail = get_kwarg("alpha_tail", kwargs)
leaf_order = get_kwarg("leaf_order", kwargs)
risk_measure = optimizer_helper.validate_risk_measure(risk_measure)
risk_measure = validate_risk_measure(risk_measure)
stock_prices = yahoo_finance_model.process_stocks(
symbols, interval, start_date, end_date
@@ -2133,31 +2174,35 @@ def get_hcp_portfolio(
linkage = linkage.upper()
risk_free_rate = risk_free_rate / time_factor[freq.upper()]
# Building the portfolio object
port = rp.HCPortfolio(
returns=stock_returns,
alpha=alpha,
a_sim=a_sim,
beta=beta,
b_sim=b_sim,
)
weights = port.optimization(
model=model,
codependence=codependence,
covariance=covariance,
obj=objective,
rm=risk_measure,
rf=risk_free_rate,
l=risk_aversion,
linkage=linkage,
k=k,
max_k=max_k,
bins_info=bins_info,
alpha_tail=alpha_tail,
leaf_order=leaf_order,
d=d_ewma,
)
try:
# Building the portfolio object
port = rp.HCPortfolio(
returns=stock_returns,
alpha=alpha,
a_sim=a_sim,
beta=beta,
b_sim=b_sim,
)
weights = port.optimization(
model=model,
codependence=codependence,
covariance=covariance,
obj=objective,
rm=risk_measure,
rf=risk_free_rate,
l=risk_aversion,
linkage=linkage,
k=k,
max_k=max_k,
bins_info=bins_info,
alpha_tail=alpha_tail,
leaf_order=leaf_order,
d=d_ewma,
)
except Exception as _:
weights = None
if weights is not None:
if value > 0.0:
@@ -2766,12 +2811,12 @@ def black_litterman(
Portfolio weights.
"""
p_views = kwargs.get("p_views", None)
q_views = kwargs.get("q_views", None)
delta = kwargs.get("delta", None)
risk_free_rate = kwargs.get("risk_free_rate", 0)
equilibrium = kwargs.get("equilibrium", True)
factor = kwargs.get("factor", 252)
p_views = get_kwarg("p_views", kwargs)
q_views = get_kwarg("q_views", kwargs)
delta = get_kwarg("delta", kwargs)
risk_free_rate = get_kwarg("risk_free_rate", kwargs)
equilibrium = get_kwarg("equilibrium", kwargs)
factor = get_kwarg("factor", kwargs, default=252)
symbols = stock_returns.columns.tolist()
benchmark = pd.Series(benchmark).to_numpy().reshape(-1, 1)
@@ -2879,7 +2924,7 @@ def get_properties() -> List[str]:
List[str]:
List of available properties to use on property optimization.
"""
return optimizer_helper.valid_property_infos
return valid_property_infos
@log_start_end(log=logger)

382
openbb_terminal/portfolio/portfolio_optimization/optimizer_view.py
View File

@@ -225,68 +225,76 @@ def portfolio_performance(
Number of CVaRs used to approximate Tail Gini of gains. If None
it duplicates a_sim value. The default is None.
"""
freq = freq.upper()
weights = pd.Series(weights).to_frame()
returns = data @ weights
mu = returns.mean().item() * time_factor[freq]
sigma = returns.std().item() * time_factor[freq] ** 0.5
sharpe = (mu - risk_free_rate) / sigma
factor_1 = str(int(time_factor[freq])) + ") "
factor_2 = "√" + factor_1
try:
freq = freq.upper()
weights = pd.Series(weights).to_frame()
returns = data @ weights
mu = returns.mean().item() * time_factor[freq]
sigma = returns.std().item() * time_factor[freq] ** 0.5
sharpe = (mu - risk_free_rate) / sigma
print("\nAnnual (by " + factor_1 + f"expected return: {100 * mu:.2f}%")
print("Annual (by " + factor_2 + f"volatility: {100 * sigma:.2f}%")
print(f"Sharpe ratio: {sharpe:.4f}")
factor_1 = str(int(time_factor[freq])) + ") "
factor_2 = "√" + factor_1
if risk_measure != "MV":
risk = rp.Sharpe_Risk(
weights,
cov=data.cov(),
returns=data,
rm=risk_measure,
rf=risk_free_rate,
alpha=alpha,
a_sim=a_sim,
beta=beta,
b_sim=b_sim,
)
print("\nAnnual (by " + factor_1 + f"expected return: {100 * mu:.2f}%")
print("Annual (by " + factor_2 + f"volatility: {100 * sigma:.2f}%")
print(f"Sharpe ratio: {sharpe:.4f}")
drawdowns = [
"MDD",
"ADD",
"DaR",
"CDaR",
"EDaR",
"UCI",
"MDD_Rel",
"ADD_Rel",
"DaR_Rel",
"CDaR_Rel",
"EDaR_Rel",
"UCI_Rel",
]
if risk_measure in drawdowns:
sharpe_2 = (mu - risk_free_rate) / risk
print(
risk_names[risk_measure.lower()].capitalize()
+ " : "
+ f"{100 * risk:.2f}%"
if risk_measure != "MV":
risk = rp.Sharpe_Risk(
weights,
cov=data.cov(),
returns=data,
rm=risk_measure,
rf=risk_free_rate,
alpha=alpha,
a_sim=a_sim,
beta=beta,
b_sim=b_sim,
)
else:
risk = risk * time_factor[freq] ** 0.5
sharpe_2 = (mu - risk_free_rate) / risk
drawdowns = [
"MDD",
"ADD",
"DaR",
"CDaR",
"EDaR",
"UCI",
"MDD_Rel",
"ADD_Rel",
"DaR_Rel",
"CDaR_Rel",
"EDaR_Rel",
"UCI_Rel",
]
if risk_measure in drawdowns:
sharpe_2 = (mu - risk_free_rate) / risk
print(
risk_names[risk_measure.lower()].capitalize()
+ " : "
+ f"{100 * risk:.2f}%"
)
else:
risk = risk * time_factor[freq] ** 0.5
sharpe_2 = (mu - risk_free_rate) / risk
print(
"Annual (by "
+ factor_2
+ risk_names[risk_measure.lower()]
+ " : "
+ f"{100 * risk:.2f}%"
)
print(
"Annual (by "
+ factor_2
"Return / "
+ risk_names[risk_measure.lower()]
+ " : "
+ f"{100 * risk:.2f}%"
+ f" ratio: {sharpe_2:.4f}"
)
print(
"Return / " + risk_names[risk_measure.lower()] + f" ratio: {sharpe_2:.4f}"
except Exception as _:
console.print(
"[red]\nFailed to calculate portfolio performance indicators.[/red]"
)
@@ -308,7 +316,7 @@ def display_weights(weights: dict, market_neutral: bool = False):
if df.empty:
return
if math.isclose(df.sum()["value"], 1, rel_tol=0.1):
if math.isclose(df.sum()["value"], 1, rel_tol=0.5):
df["value"] = (df["value"] * 100).apply(lambda s: f"{s:.2f}") + " %"
df["value"] = (
df["value"]
@@ -685,6 +693,10 @@ def display_equal_weight(
value=value,
)
if not weights:
console.print("There is no solution with these parameters.")
return {}
if table:
console.print(s_title)
display_weights(weights)
@@ -799,6 +811,10 @@ def display_property_weighting(
value=value,
)
if not weights:
console.print("There is no solution with these parameters.")
return {}
if table:
console.print(s_title)
display_weights(weights)
@@ -981,8 +997,8 @@ def display_mean_risk(
value_short=value_short,
)
if weights is None:
console.print("\n", "There is no solution with these parameters")
if not weights:
console.print("There is no solution with these parameters.")
return {}
if table:
@@ -1150,11 +1166,8 @@ def display_max_sharpe(
value_short=value_short,
)
if stock_returns is None or stock_returns.empty:
return {}
if weights is None:
console.print("\n", "There is no solution with these parameters")
if not weights:
console.print("There is no solution with these parameters.")
return {}
if table:
@@ -1321,8 +1334,8 @@ def display_min_risk(
value_short=value_short,
)
if weights is None:
console.print("\n", "There is no solution with these parameters")
if not weights:
console.print("There is no solution with these parameters.")
return {}
if table:
@@ -1489,8 +1502,8 @@ def display_max_util(
value_short=value_short,
)
if weights is None:
console.print("\n", "There is no solution with these parameters")
if not weights:
console.print("There is no solution with these parameters.")
return {}
if table:
@@ -1657,8 +1670,8 @@ def display_max_ret(
value_short=value_short,
)
if weights is None:
console.print("\n", "There is no solution with these parameters")
if not weights:
console.print("There is no solution with these parameters.")
return {}
if table:
@@ -1771,8 +1784,9 @@ def display_max_div(
value=value,
value_short=value_short,
)
if weights is None:
console.print("\n", "There is no solution with this parameters")
if not weights:
console.print("There is no solution with these parameters.")
return {}
if table:
@@ -1886,8 +1900,8 @@ def display_max_decorr(
value_short=value_short,
)
if weights is None:
console.print("\n", "There is no solution with this parameters")
if not weights:
console.print("There is no solution with this parameters")
return {}
if table:
@@ -2026,8 +2040,8 @@ def display_black_litterman(
value_short=value_short,
)
if weights is None:
console.print("\n", "There is no solution with this parameters")
if not weights:
console.print("There is no solution with this parameters")
return {}
if table:
@@ -2160,110 +2174,113 @@ def display_ef(
seed=seed,
)
risk_free_rate = risk_free_rate / time_factor[freq.upper()]
try:
risk_free_rate = risk_free_rate / time_factor[freq.upper()]
if external_axes is None:
_, ax = plt.subplots(figsize=plot_autoscale(), dpi=PLOT_DPI)
else:
ax = external_axes[0]
if external_axes is None:
_, ax = plt.subplots(figsize=plot_autoscale(), dpi=PLOT_DPI)
else:
ax = external_axes[0]
ax = rp.plot_frontier(
w_frontier=frontier,
mu=mu,
cov=cov,
returns=stock_returns,
rm=risk_choices[risk_measure.lower()],
rf=risk_free_rate,
alpha=alpha,
cmap="RdYlBu",
w=weights,
label="",
marker="*",
s=16,
c="r",
t_factor=time_factor[freq.upper()],
ax=ax,
)
# Add risk free line
if tangency:
ret_sharpe = (mu @ weights).to_numpy().item() * time_factor[freq.upper()]
risk_sharpe = rp.Sharpe_Risk(
weights,
ax = rp.plot_frontier(
w_frontier=frontier,
mu=mu,
cov=cov,
returns=stock_returns,
rm=risk_choices[risk_measure.lower()],
rf=risk_free_rate,
alpha=alpha,
# a_sim=a_sim,
# beta=beta,
# b_sim=b_sim,
cmap="RdYlBu",
w=weights,
label="",
marker="*",
s=16,
c="r",
t_factor=time_factor[freq.upper()],
ax=ax,
)
if risk_choices[risk_measure.lower()] not in [
"ADD",
"MDD",
"CDaR",
"EDaR",
"UCI",
]:
risk_sharpe = risk_sharpe * time_factor[freq.upper()] ** 0.5
y = ret_sharpe * 1.5
b = risk_free_rate * time_factor[freq.upper()]
m = (ret_sharpe - b) / risk_sharpe
x2 = (y - b) / m
x = [0, x2]
y = [b, y]
line = Line2D(x, y, label="Capital Allocation Line")
ax.set_xlim(xmin=min(X1) * 0.8)
ax.add_line(line)
ax.plot(X1, Y1, color="b")
plot_tickers = True
if plot_tickers:
ticker_plot = pd.DataFrame(columns=["ticker", "var"])
for ticker in port.cov.columns:
weight_df = pd.DataFrame({"weights": 1}, index=[ticker])
risk = rp.Sharpe_Risk(
weight_df,
cov=port.cov[ticker][ticker],
returns=stock_returns.loc[:, [ticker]],
# Add risk free line
if tangency:
ret_sharpe = (mu @ weights).to_numpy().item() * time_factor[freq.upper()]
risk_sharpe = rp.Sharpe_Risk(
weights,
cov=cov,
returns=stock_returns,
rm=risk_choices[risk_measure.lower()],
rf=risk_free_rate,
alpha=alpha,
# a_sim=a_sim,
# beta=beta,
# b_sim=b_sim,
)
if risk_choices[risk_measure.lower()] not in [
"MDD",
"ADD",
"MDD",
"CDaR",
"EDaR",
"UCI",
]:
risk = risk * time_factor[freq.upper()] ** 0.5
risk_sharpe = risk_sharpe * time_factor[freq.upper()] ** 0.5
ticker_plot = ticker_plot.append(
{"ticker": ticker, "var": risk}, ignore_index=True
y = ret_sharpe * 1.5
b = risk_free_rate * time_factor[freq.upper()]
m = (ret_sharpe - b) / risk_sharpe
x2 = (y - b) / m
x = [0, x2]
y = [b, y]
line = Line2D(x, y, label="Capital Allocation Line")
ax.set_xlim(xmin=min(X1) * 0.8)
ax.add_line(line)
ax.plot(X1, Y1, color="b")
plot_tickers = True
if plot_tickers:
ticker_plot = pd.DataFrame(columns=["ticker", "var"])
for ticker in port.cov.columns:
weight_df = pd.DataFrame({"weights": 1}, index=[ticker])
risk = rp.Sharpe_Risk(
weight_df,
cov=port.cov[ticker][ticker],
returns=stock_returns.loc[:, [ticker]],
rm=risk_choices[risk_measure.lower()],
rf=risk_free_rate,
alpha=alpha,
)
if risk_choices[risk_measure.lower()] not in [
"MDD",
"ADD",
"CDaR",
"EDaR",
"UCI",
]:
risk = risk * time_factor[freq.upper()] ** 0.5
ticker_plot = ticker_plot.append(
{"ticker": ticker, "var": risk}, ignore_index=True
)
ticker_plot = ticker_plot.set_index("ticker")
ticker_plot = ticker_plot.merge(
port.mu.T * time_factor[freq.upper()], right_index=True, left_index=True
)
ticker_plot = ticker_plot.set_index("ticker")
ticker_plot = ticker_plot.merge(
port.mu.T * time_factor[freq.upper()], right_index=True, left_index=True
)
ticker_plot = ticker_plot.rename(columns={0: "ret"})
ax.scatter(ticker_plot["var"], ticker_plot["ret"])
for row in ticker_plot.iterrows():
ax.annotate(row[0], (row[1]["var"], row[1]["ret"]))
ax.set_title(f"Efficient Frontier simulating {n_portfolios} portfolios")
ax.legend(loc="best", scatterpoints=1)
theme.style_primary_axis(ax)
l, b, w, h = ax.get_position().bounds
ax.set_position([l, b, w * 0.9, h])
ax1 = ax.get_figure().axes
ll, bb, ww, hh = ax1[-1].get_position().bounds
ax1[-1].set_position([ll * 1.02, bb, ww, hh])
if external_axes is None:
theme.visualize_output(force_tight_layout=False)
ticker_plot = ticker_plot.rename(columns={0: "ret"})
ax.scatter(ticker_plot["var"], ticker_plot["ret"])
for row in ticker_plot.iterrows():
ax.annotate(row[0], (row[1]["var"], row[1]["ret"]))
ax.set_title(f"Efficient Frontier simulating {n_portfolios} portfolios")
ax.legend(loc="best", scatterpoints=1)
theme.style_primary_axis(ax)
l, b, w, h = ax.get_position().bounds
ax.set_position([l, b, w * 0.9, h])
ax1 = ax.get_figure().axes
ll, bb, ww, hh = ax1[-1].get_position().bounds
ax1[-1].set_position([ll * 1.02, bb, ww, hh])
if external_axes is None:
theme.visualize_output(force_tight_layout=False)
except Exception as _:
console.print("[red]Error plotting efficient frontier.[/red]")
@log_start_end(log=logger)
@@ -2401,8 +2418,8 @@ def display_risk_parity(
value=value,
)
if weights is None:
console.print("\n", "There is no solution with this parameters")
if not weights:
console.print("There is no solution with this parameters")
return {}
if table:
@@ -2542,8 +2559,8 @@ def display_rel_risk_parity(
value=value,
)
if weights is None:
console.print("\n", "There is no solution with this parameters")
if not weights:
console.print("There is no solution with this parameters")
return {}
if table:
@@ -2807,8 +2824,8 @@ def display_hcp(
value=value,
)
if weights is None:
console.print("\n", "There is no solution with this parameters")
if not weights:
console.print("There is no solution with this parameters")
return {}
if table:
@@ -3051,8 +3068,8 @@ def display_hrp(
value=value,
)
if weights is None:
console.print("\n", "There is no solution with this parameters")
if not weights:
console.print("There is no solution with this parameters")
return {}
if table:
@@ -3303,8 +3320,8 @@ def display_herc(
value=value,
)
if weights is None:
console.print("\n", "There is no solution with this parameters")
if not weights:
console.print("There is no solution with this parameters")
return {}
if table:
@@ -3566,8 +3583,8 @@ def display_nco(
value=value,
)
if weights is None:
console.print("\n", "There is no solution with this parameters")
if not weights:
console.print("There is no solution with this parameters")
return {}
if table:
@@ -3703,8 +3720,9 @@ def pie_chart_weights(
def additional_plots(
weights: Dict,
data: pd.DataFrame,
category: Dict = None,
title_opt: str = "",
category_dict: Dict = None,
category: str = "",
portfolio_name: str = "",
freq: str = "D",
risk_measure: str = "MV",
risk_free_rate: float = 0,
@@ -3728,8 +3746,12 @@ def additional_plots(
Dict of portfolio weights
data: pd.DataFrame
DataFrame of stock returns
title_opt: str
Title to be used on the pie chart
category_dict: Dict
Dict of categories
category: str
Category to plot
portfolio_name: str
Portfolio name
freq: str, optional
The frequency used to calculate returns. Default value is 'D'. Possible
values are:
@@ -3795,12 +3817,15 @@ def additional_plots(
external_axes: Optional[List[plt.Axes]]
Optional axes to plot data on
"""
if category is not None:
title_opt = category if not portfolio_name else category + " - " + portfolio_name
if category_dict is not None:
weights_df = pd.DataFrame.from_dict(
data=weights, orient="index", columns=["value"], dtype=float
)
category_df = pd.DataFrame.from_dict(
data=category, orient="index", columns=["category"]
data=category_dict, orient="index", columns=["category"]
)
weights_df = weights_df.join(category_df, how="inner")
weights_df.sort_index(inplace=True)
@@ -3941,7 +3966,10 @@ def additional_plots(
if heat:
if len(weights) == 1:
console.print(f"Heatmap needs at least two values for '{category}'.")
single_key = list(weights.keys())[0].upper()
console.print(
f"[yellow]Heatmap needs at least two values for '{category}', only found '{single_key}'.[/yellow]"
)
return
if external_axes is None:
@@ -3971,7 +3999,7 @@ def additional_plots(
ax[0].grid(False)
ax[0].axis("off")
if category is None:
if category_dict is None:
# Vertical dendrogram
l, b, w, h = ax[4].get_position().bounds
l1 = l * 0.5

85
openbb_terminal/portfolio/portfolio_optimization/parameters/Parameter.py Normal file
View File

@@ -0,0 +1,85 @@
from dataclasses import dataclass
from typing import Any, List, Optional, Type
# pylint: disable=unidiomatic-typecheck
@dataclass(frozen=True)
class Parameter:
"""Class for storing parameters. This dataclass is frozen so no attributes can be changed after initialization.
Parameters
----------
name : str
Name of the parameter.
type_ : type
Type of the parameter.
default : Any
Default value of the parameter.
choices : Optional[List[Any]]
List of possible values for the parameter.
"""
name: str
type_: Type
default: Any
choices: Optional[List[Any]] = None
def __post_init__(self):
"""Post initialization."""
self.validate()
def validate(self):
"""Validate the value of the parameter.
Parameters
----------
value : Any
Value of the parameter.
Raises
------
TypeError
If the value is not of the correct type.
ValueError
If the value is not in the list of possible values.
"""
if not self.validate_type(self.default):
raise TypeError(
f"Default for '{self.name}' must be of type '{self.type_.__name__}'."
)
if self.choices is not None:
if not all(self.validate_type(choice) for choice in self.choices):
raise TypeError(
f"Choices must be of the same type as the parameter.\n"
f"'{self.name}': '{self.choices}' -> '{self.type_.__name__}'"
)
if self.choices is not None and self.default not in self.choices:
raise ValueError(
f"Parameter '{self.name}' must be one of type '{self.choices}'."
)
def validate_type(self, value: Any) -> bool:
"""Check if the value is of the correct type.
Parameters
----------
value : Any
Value of the parameter.
Returns
-------
bool
True if the value is of the correct type, False otherwise.
"""
# Temporary workaround until we drop Python 3.9 for:
# Error: Subscripted generics cannot be used with class and instance checks
#
# isinstance for subscripted generics (e.g. List[str]) is only available
# in Python 3.10, so this method does not check for the type of subscripted generics.
# TODO: Though we should implement it.
# See https://stackoverflow.com/questions/70825351/python-3-typeerror-subscripted-generics-cannot-be-used-with-class-and-instance # noqa: E501
try:
return isinstance(value, self.type_)
except Exception as _: # noqa: F841
return True

17
openbb_terminal/portfolio/portfolio_optimization/parameters/params_controller.py
View File

@@ -9,7 +9,7 @@ from typing import List, Optional
from openbb_terminal.custom_prompt_toolkit import NestedCompleter
from openbb_terminal import feature_flags as gtff
from openbb_terminal import feature_flags as obbff
from openbb_terminal.decorators import log_start_end
from openbb_terminal.menu import session
from openbb_terminal.parent_classes import BaseController
@@ -76,7 +76,7 @@ class ParametersController(BaseController):
self.description: Optional[str] = None
self.DATA_FILES = params_helpers.load_data_files()
if session and gtff.USE_PROMPT_TOOLKIT:
if session and obbff.USE_PROMPT_TOOLKIT:
choices: dict = {c: {} for c in self.controller_choices}
choices["set"] = {c: None for c in self.models}
choices["set"]["--model"] = {c: None for c in self.models}
@@ -93,8 +93,15 @@ class ParametersController(BaseController):
"--show_arguments": {},
"-s": "--show_arguments",
}
self.choices = choices
self.completer = NestedCompleter.from_nested_dict(choices)
def update_runtime_choices(self):
if session and obbff.USE_PROMPT_TOOLKIT:
self.DATA_FILES = params_helpers.load_data_files()
self.choices["load"]["--file"] = {c: {} for c in self.DATA_FILES}
self.completer = NestedCompleter.from_nested_dict(self.choices)
def print_help(self):
"""Print help"""
mt = MenuText("portfolio/po/params/")
@@ -112,8 +119,10 @@ class ParametersController(BaseController):
mt.add_raw("\n")
mt.add_info("_parameters_")
if self.current_model:
max_len = max(len(k) for k in self.params.keys())
for k, v in self.params.items():
v = params_helpers.booltostr(v)
all_params = DEFAULT_PARAMETERS + MODEL_PARAMS[self.current_model]
if k in all_params:
mt.add_raw(
@@ -122,6 +131,7 @@ class ParametersController(BaseController):
else:
max_len = max(len(k) for k in self.params.keys())
for k, v in self.params.items():
v = params_helpers.booltostr(v)
mt.add_raw(
f" [param]{k}{' ' * (max_len - len(k))} :[/param] {v}\n"
)
@@ -200,7 +210,8 @@ class ParametersController(BaseController):
ns_parser = self.parse_known_args_and_warn(parser, other_args)
if ns_parser:
self.current_file = str(params_view.save_file(ns_parser.file, self.params))
console.print()
self.update_runtime_choices()
@log_start_end(log=logger)
def call_clear(self, other_args: List[str]):

124
openbb_terminal/portfolio/portfolio_optimization/parameters/params_helpers.py
View File

@@ -1,10 +1,14 @@
import argparse
from datetime import datetime
from typing import Dict, List
from datetime import datetime, date
from typing import Any, Dict, List
from pathlib import Path
from openbb_terminal.helper_funcs import log_and_raise
from openbb_terminal.core.config import paths
from openbb_terminal.rich_config import console
from openbb_terminal.portfolio.portfolio_optimization.statics import (
OPTIMIZATION_PARAMETERS,
TERMINAL_TEMPLATE_MAP,
)
def check_save_file(file: str) -> str:
@@ -33,7 +37,7 @@ def load_data_files() -> Dict[str, Path]:
The dictionary of filenames and their paths
"""
default_path = paths.MISCELLANEOUS_DIRECTORY / "portfolio_examples" / "optimization"
custom_exports = paths.USER_EXPORTS_DIRECTORY / "portfolio"
custom_exports = paths.USER_PORTFOLIO_DATA_DIRECTORY / "optimization"
data_files = {}
for directory in [default_path, custom_exports]:
for file_type in ["xlsx", "ini"]:
@@ -44,6 +48,77 @@ def load_data_files() -> Dict[str, Path]:
return data_files
def check_convert_parameters(received_parameters: dict) -> dict:
"""Check if the argument is in the parameters list.
If it is, try to cast it to the correct type, else use default value.
Parameters
----------
received_parameters: dict
The parameters to be checked
Returns
-------
dict
The parameters with the correct types
"""
converted_parameters = check_convert_dates(
received_parameters, ["start_period", "start_date", "end_period", "end_date"]
)
for received_name, received_value in received_parameters.items():
# TODO: Remove this line when mapping between template and terminal is not needed
template_name = TERMINAL_TEMPLATE_MAP.get(received_name, received_name)
if template_name in OPTIMIZATION_PARAMETERS:
PARAMETER = OPTIMIZATION_PARAMETERS[template_name]
if not PARAMETER.validate_type(received_value):
converted_parameters[received_name] = check_convert_parameter(
name=received_name, value=received_value, parameter=PARAMETER
)
return converted_parameters
def check_convert_parameter(name, value, parameter):
"""Converts a parameter to the correct type
Parameters
----------
name: str
The name of the received parameter
value: str
The value of the received parameter
parameter: Parameter
The parameter object
Returns
-------
The converted parameter
"""
try:
# Try to cast the value to the correct type if int or float
if parameter.type_ is int:
new_value = parameter.type_(float(value))
elif parameter.type_ is float:
new_value = parameter.type_(value)
elif parameter.type_ is bool:
new_value = strtobool(value)
else:
new_value = value
except ValueError:
new_value = parameter.default
console.print(
f"[red]'{name}' format should be '{parameter.type_.__name__}' type[/red]",
f"[red]and could not be converted. Setting default '{new_value}'.\n[/red]",
)
return new_value
def check_convert_dates(params: dict, param_name_list: List[str]) -> dict:
"""Check if the argument is in the list and convert it to a str object
Parameters
@@ -62,7 +137,7 @@ def check_convert_dates(params: dict, param_name_list: List[str]) -> dict:
for param_name in param_name_list:
if param_name in params:
param_value = params[param_name]
if isinstance(param_value, datetime):
if isinstance(param_value, date):
params[param_name] = param_value.strftime("%Y-%m-%d")
elif isinstance(param_value, str):
try:
@@ -79,3 +154,44 @@ def check_convert_dates(params: dict, param_name_list: List[str]) -> dict:
params.pop(param_name)
return params
def booltostr(value: bool) -> Any:
"""Converts a bool to a string or returns the value itself if not bool
Parameters
----------
value: bool
The bool to be converted or the value itself if not bool
Returns
-------
Any
The converted value
"""
if isinstance(value, bool):
return "True" if value else "False"
return value
def strtobool(value: str) -> Any:
"""Converts a string to a bool or returns the value itself if not string
Parameters
----------
value: str
The string to be converted or the value itself if not string
Returns
-------
Any
The converted value
"""
if isinstance(value, str):
if value.lower() == "true":
return True
elif value.lower() == "false":
return False
return value

20
openbb_terminal/portfolio/portfolio_optimization/parameters/params_view.py
View File

@@ -9,7 +9,8 @@ from openbb_terminal.portfolio.portfolio_optimization import excel_model
from openbb_terminal.rich_config import console
from openbb_terminal.portfolio.portfolio_optimization.parameters import params_statics
from openbb_terminal.portfolio.portfolio_optimization.parameters.params_helpers import (
check_convert_dates,
booltostr,
check_convert_parameters,
)
from openbb_terminal.core.config import paths
@@ -47,35 +48,36 @@ def load_file(path: str = "") -> Tuple[dict, str]:
console.print("Cannot load in the file due to not being an .ini or .xlsx file.")
return {}, ""
params = check_convert_dates(
params=params, param_name_list=["start_period", "end_period"]
)
converted_parameters = check_convert_parameters(received_parameters=params)
max_len = max(len(k) for k in params.keys())
max_len = max(len(k) for k in converted_parameters.keys())
help_text = "[info]Parameters:[/info]\n"
if current_model:
for k, v in params.items():
for k, v in converted_parameters.items():
all_params = (
params_statics.DEFAULT_PARAMETERS
+ params_statics.MODEL_PARAMS[current_model]
)
if k in all_params:
v = booltostr(v)
help_text += f" [param]{k}{' ' * (max_len - len(k))} :[/param] {v}\n"
else:
for k, v in params.items():
for k, v in converted_parameters.items():
v = booltostr(v)
help_text += f" [param]{k}{' ' * (max_len - len(k))} :[/param] {v}\n"
console.print(help_text)
return params, current_model
return converted_parameters, current_model
def save_file(path: str, params: dict) -> Path:
if not path.endswith(".ini"):
console.print("[red]File to be saved needs to be a .ini file.[/red]\n")
# Create file if it does not exist
base_path = paths.USER_EXPORTS_DIRECTORY / "portfolio"
base_path = paths.USER_PORTFOLIO_DATA_DIRECTORY / "optimization"
if not base_path.is_dir():
base_path.mkdir()
filepath = base_path / path

44
openbb_terminal/portfolio/portfolio_optimization/po_controller.py
View File

@@ -277,9 +277,15 @@ class PortfolioOptimizationController(BaseController):
def update_runtime_choices(self):
if session and obbff.USE_PROMPT_TOOLKIT:
if self.portfolios:
self.choices["show"] = {c: {} for c in list(self.portfolios.keys())}
self.choices = {**self.choices, **self.SUPPORT_CHOICES}
self.choices["show"]["--portfolios"] = {
c: {} for c in list(self.portfolios.keys())
}
self.choices["rpf"]["--portfolios"] = {
c: {} for c in list(self.portfolios.keys())
}
self.choices["plot"]["--portfolios"] = {
c: {} for c in list(self.portfolios.keys())
}
self.completer = NestedCompleter.from_nested_dict(self.choices)
def print_help(self):
@@ -697,7 +703,6 @@ class PortfolioOptimizationController(BaseController):
dest="file",
help="Allocation file to be used",
choices=self.allocation_file_map.keys(),
metavar="FILE",
)
if other_args and "-" not in other_args[0][0]:
other_args.insert(0, "--file")
@@ -850,7 +855,7 @@ class PortfolioOptimizationController(BaseController):
optimizer_view.additional_plots(
weights=weights,
data=stock_returns[stocks],
title_opt=i,
portfolio_name=i,
freq=ns_parser.return_frequency,
risk_measure=ns_parser.risk_measure.lower(),
risk_free_rate=ns_parser.risk_free,
@@ -888,8 +893,9 @@ class PortfolioOptimizationController(BaseController):
optimizer_view.additional_plots(
weights=weights,
data=stock_returns[stocks],
category=filtered_categories,
title_opt=category + " - " + i,
category_dict=filtered_categories,
category=category,
portfolio_name=i,
freq=ns_parser.return_frequency,
risk_measure=ns_parser.risk_measure.lower(),
risk_free_rate=ns_parser.risk_free,
@@ -1159,7 +1165,7 @@ class PortfolioOptimizationController(BaseController):
default=self.params["target_return"]
if "target_return" in self.params
else -1,
type=int,
type=float,
help="Constraint on minimum level of portfolio's return",
)
parser.add_argument(
@@ -1167,6 +1173,7 @@ class PortfolioOptimizationController(BaseController):
"--target-risk",
dest="target_risk",
default=self.params["target_risk"] if "target_risk" in self.params else -1,
type=float,
help="Constraint on maximum level of portfolio's risk",
)
parser.add_argument(
@@ -1182,8 +1189,8 @@ class PortfolioOptimizationController(BaseController):
parser.add_argument(
"-cv",
"--covariance",
default=self.params["expected_return"]
if "expected_return" in self.params
default=self.params["covariance"]
if "covariance" in self.params
else "hist",
dest="covariance",
help="""Method used to estimate covariance matrix. Possible values are
@@ -1341,6 +1348,7 @@ class PortfolioOptimizationController(BaseController):
default=self.params["target_return"]
if "target_return" in self.params
else -1,
type=float,
help="Constraint on minimum level of portfolio's return",
)
parser.add_argument(
@@ -1348,6 +1356,7 @@ class PortfolioOptimizationController(BaseController):
"--target-risk",
dest="target_risk",
default=self.params["target_risk"] if "target_risk" in self.params else -1,
type=float,
help="Constraint on maximum level of portfolio's risk",
)
parser.add_argument(
@@ -1518,8 +1527,8 @@ class PortfolioOptimizationController(BaseController):
"--risk-aversion",
type=float,
dest="risk_aversion",
default=self.params["long_allocation"]
if "long_allocation" in self.params
default=self.params["risk_aversion"]
if "risk_aversion" in self.params
else 1,
help="Risk aversion parameter",
)
@@ -1530,6 +1539,7 @@ class PortfolioOptimizationController(BaseController):
default=self.params["target_return"]
if "target_return" in self.params
else -1,
type=float,
help="Constraint on minimum level of portfolio's return",
)
parser.add_argument(
@@ -1537,6 +1547,7 @@ class PortfolioOptimizationController(BaseController):
"--target-risk",
dest="target_risk",
default=self.params["target_risk"] if "target_risk" in self.params else -1,
type=float,
help="Constraint on maximum level of portfolio's risk",
)
parser.add_argument(
@@ -1713,6 +1724,7 @@ class PortfolioOptimizationController(BaseController):
default=self.params["target_return"]
if "target_return" in self.params
else -1,
type=float,
help="Constraint on minimum level of portfolio's return",
)
parser.add_argument(
@@ -1720,6 +1732,7 @@ class PortfolioOptimizationController(BaseController):
"--target-risk",
dest="target_risk",
default=self.params["target_risk"] if "target_risk" in self.params else -1,
type=float,
help="Constraint on maximum level of portfolio's risk",
)
parser.add_argument(
@@ -2208,8 +2221,8 @@ class PortfolioOptimizationController(BaseController):
"--risk-aversion",
type=float,
dest="risk_aversion",
default=self.params["long_allocation"]
if "long_allocation" in self.params
default=self.params["risk_aversion"]
if "risk_aversion" in self.params
else 1,
help="Risk aversion parameter",
)
@@ -2218,6 +2231,7 @@ class PortfolioOptimizationController(BaseController):
"--delta",
default=self.params["delta"] if "delta" in self.params else None,
dest="delta",
type=float,
help="Risk aversion factor of Black Litterman model",
)
parser.add_argument(
@@ -2550,6 +2564,7 @@ class PortfolioOptimizationController(BaseController):
default=self.params["target_return"]
if "target_return" in self.params
else -1,
type=float,
help="Constraint on minimum level of portfolio's return",
)
parser.add_argument(
@@ -2707,6 +2722,7 @@ class PortfolioOptimizationController(BaseController):
default=self.params["target_return"]
if "target_return" in self.params
else -1,
type=float,
help="Constraint on minimum level of portfolio's return",
)
parser.add_argument(

29
openbb_terminal/portfolio/portfolio_optimization/po_engine.py
View File

@@ -194,41 +194,58 @@ class PoEngine:
"""
self._weights = weights
def get_weights(self) -> Dict[str, float]:
def get_weights(self, warning=True) -> Dict[str, float]:
"""Get the weights
Parameters
----------
warning : bool, optional
Display warning, by default True
Returns
-------
Dict[str, float]
Weights
"""
if not self._weights:
console.print("No weights found. Please perform some optimization.")
if warning:
console.print("No weights found. Please perform some optimization.")
return {}
return self._weights
def get_weights_df(self) -> pd.DataFrame:
def get_weights_df(self, warning=True) -> pd.DataFrame:
"""Get the weights
Parameters
----------
warning : bool, optional
Display warning, by default True
Returns
-------
pd.DataFrame
Weights
"""
if not self._weights:
console.print("No weights found. Please perform some optimization.")
if warning:
console.print("No weights found. Please perform some optimization.")
return pd.DataFrame()
return optimizer_helper.dict_to_df(self._weights)
def set_params(self, params: Dict[str, float]):
def set_params(self, params: Dict[str, float], update=False):
"""Set the parameters
Parameters
----------
params : Dict[str, float]
Parameters
update : bool, optional
Update the current model, by default False
"""
self._params.update(params)
if update:
self._params.update(params)
else:
self._params = params
def get_params(self) -> Dict:
"""Get the parameters

159
openbb_terminal/portfolio/portfolio_optimization/po_model.py
View File

@@ -18,11 +18,14 @@ from openbb_terminal.portfolio.portfolio_optimization import (
optimizer_helper,
optimizer_model,
)
from openbb_terminal.portfolio.portfolio_optimization.parameters.params_helpers import (
check_convert_parameters,
)
from openbb_terminal.portfolio.portfolio_optimization.statics import (
RISK_NAMES,
TERMINAL_TEMPLATE_MAP,
TIME_FACTOR,
DRAWDOWNS,
PARAM_TYPES,
)
from openbb_terminal.portfolio.portfolio_optimization.po_engine import PoEngine
from openbb_terminal.rich_config import console
@@ -32,32 +35,6 @@ warnings.filterwarnings("ignore")
logger = logging.getLogger(__name__)
@log_start_end(log=logger)
def validate_parameters_type(parameters):
"""Validate parameters type
Parameters
----------
parameters : dict
Keyword arguments
"""
for key, value in parameters.items():
if key in PARAM_TYPES:
expected_type = PARAM_TYPES[key]
if not isinstance(value, expected_type):
if expected_type is str:
parameters.update({key: str(value)})
elif expected_type is float:
parameters.update({key: float(value)})
elif expected_type is bool:
parameters.update({key: bool(value)})
else:
console.print(
f"[info]Parameter {key} should be of type {expected_type.__name__}. Casting failed, reverting to default.[/info]"
)
parameters.pop(key)
@log_start_end(log=logger)
def generate_portfolio(
symbols_categories: Dict[str, Dict[str, str]] = None,
@@ -165,7 +142,7 @@ def load_parameters_file(
'maxnan': '0.05',
'threshold': '0.3',
'alpha': '0.05'}
>>> p.set_params({"risk_free_rate": 0.05})
>>> p.set_params({"risk_free_rate": 0.05}, update=True)
>>> p.get_params()
{'interval': '3y',
'log_returns': '0',
@@ -208,9 +185,15 @@ def validate_inputs(
else:
console.print("No 'portfolio_engine' provided.")
validate_parameters_type(parameters)
converted_parameters = check_convert_parameters(received_parameters=parameters)
return symbols, portfolio_engine, parameters
terminal_parameters = {}
# TODO: Remove this conversion when mapping between template and terminal is not needed
TEMPLATE_TERMINAL_MAP = {v: k for k, v in TERMINAL_TEMPLATE_MAP.items()}
for key, value in converted_parameters.items():
terminal_parameters[TEMPLATE_TERMINAL_MAP.get(key, key)] = value
return symbols, portfolio_engine, terminal_parameters
@log_start_end(log=logger)
@@ -230,51 +213,61 @@ def get_portfolio_performance(weights: Dict, data: pd.DataFrame, **kwargs) -> Di
Portfolio performance
"""
freq = kwargs.get("freq", "D")
risk_measure = kwargs.get("risk_measure", "MV")
risk_free_rate = kwargs.get("risk_free_rate", 0.0)
alpha = kwargs.get("alpha", 0.05)
a_sim = kwargs.get("a_sim", 100)
beta = kwargs.get("beta", None)
b_sim = kwargs.get("b_sim", None)
try:
if not weights:
return {}
freq = freq.upper()
weights = pd.Series(weights).to_frame()
returns = data @ weights
mu = returns.mean().item() * TIME_FACTOR[freq]
sigma = returns.std().item() * TIME_FACTOR[freq] ** 0.5
sharpe = (mu - risk_free_rate) / sigma
freq = optimizer_helper.get_kwarg("freq", kwargs)
risk_measure = optimizer_helper.get_kwarg("risk_measure", kwargs)
risk_free_rate = optimizer_helper.get_kwarg("risk_free_rate", kwargs)
alpha = optimizer_helper.get_kwarg("alpha", kwargs)
a_sim = optimizer_helper.get_kwarg("a_sim", kwargs)
beta = optimizer_helper.get_kwarg("beta", kwargs)
b_sim = optimizer_helper.get_kwarg("b_sim", kwargs)
performance_dict = {
"Return": mu,
"Volatility": sigma,
"Sharpe ratio": sharpe,
}
freq = freq.upper()
weights = pd.Series(weights).to_frame()
returns = data @ weights
mu = returns.mean().item() * TIME_FACTOR[freq]
sigma = returns.std().item() * TIME_FACTOR[freq] ** 0.5
sharpe = (mu - risk_free_rate) / sigma
risk_measure = optimizer_helper.validate_risk_measure(risk_measure, warning=False)
performance_dict = {
"Return": mu,
"Volatility": sigma,
"Sharpe ratio": sharpe,
}
if risk_measure != "MV":
risk = rp.Sharpe_Risk(
weights,
cov=data.cov(),
returns=data,
rm=risk_measure,
rf=risk_free_rate,
alpha=alpha,
a_sim=a_sim,
beta=beta,
b_sim=b_sim,
risk_measure = optimizer_helper.validate_risk_measure(
risk_measure, warning=False
)
if risk_measure in DRAWDOWNS:
sharpe_2 = (mu - risk_free_rate) / risk
else:
risk = risk * TIME_FACTOR[freq] ** 0.5
sharpe_2 = (mu - risk_free_rate) / risk
if risk_measure != "MV":
risk = rp.Sharpe_Risk(
weights,
cov=data.cov(),
returns=data,
rm=risk_measure,
rf=risk_free_rate,
alpha=alpha,
a_sim=a_sim,
beta=beta,
b_sim=b_sim,
)
performance_dict[RISK_NAMES[risk_measure.lower()]] = risk
performance_dict.update({"Sharpe ratio (risk adjusted)": sharpe_2})
if risk_measure in DRAWDOWNS:
sharpe_2 = (mu - risk_free_rate) / risk
else:
risk = risk * TIME_FACTOR[freq] ** 0.5
sharpe_2 = (mu - risk_free_rate) / risk
performance_dict[RISK_NAMES[risk_measure.lower()]] = risk
performance_dict.update({"Sharpe ratio (risk adjusted)": sharpe_2})
except Exception as _:
console.print(
"[red]\nFailed to calculate portfolio performance indicators.[/red]"
)
performance_dict = {}
return performance_dict
@@ -421,7 +414,7 @@ def get_maxsharpe(
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -567,7 +560,7 @@ def get_minrisk(
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -713,7 +706,7 @@ def get_maxutil(
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -859,7 +852,7 @@ def get_maxret(
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -982,7 +975,7 @@ def get_maxdiv(
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -1083,7 +1076,7 @@ def get_maxdecorr(
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -1193,7 +1186,7 @@ def get_blacklitterman(
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -1450,7 +1443,7 @@ def get_riskparity(
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -1594,7 +1587,7 @@ def get_relriskparity(
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -1793,7 +1786,7 @@ def get_hrp(portfolio_engine: PoEngine = None, **kwargs) -> Tuple[pd.DataFrame,
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -1992,7 +1985,7 @@ def get_herc(portfolio_engine: PoEngine = None, **kwargs) -> Tuple[pd.DataFrame,
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -2191,7 +2184,7 @@ def get_nco(portfolio_engine: PoEngine = None, **kwargs) -> Tuple[pd.DataFrame,
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -2270,7 +2263,7 @@ def get_equal(portfolio_engine: PoEngine = None, **kwargs) -> Tuple[pd.DataFrame
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -2351,7 +2344,7 @@ def get_mktcap(
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -2432,7 +2425,7 @@ def get_dividend(
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)
@@ -2522,7 +2515,7 @@ def get_property(
valid_portfolio_engine.set_weights(weights=weights)
valid_portfolio_engine.set_returns(returns=returns)
return valid_portfolio_engine.get_weights_df(), performance_dict
return valid_portfolio_engine.get_weights_df(warning=False), performance_dict
@log_start_end(log=logger)

37
openbb_terminal/portfolio/portfolio_optimization/po_view.py
View File

@@ -28,6 +28,7 @@ from openbb_terminal.portfolio.portfolio_optimization.statics import (
RISK_CHOICES,
TIME_FACTOR,
)
from openbb_terminal.portfolio.portfolio_optimization.optimizer_helper import get_kwarg
from openbb_terminal.portfolio.portfolio_optimization.po_engine import PoEngine
from openbb_terminal.rich_config import console
@@ -121,11 +122,11 @@ def display_ef(portfolio_engine: PoEngine = None, **kwargs):
>>> openbb.portfolio.po.ef_chart(portfolio_engine=p)
"""
n_portfolios = kwargs.get("n_portfolios", 100)
freq = kwargs.get("freq", "D")
risk_measure = kwargs.get("risk_measure", "MV")
risk_free_rate = kwargs.get("risk_free_rate", 0.0)
alpha = kwargs.get("alpha", 0.05)
n_portfolios = get_kwarg("n_portfolios", kwargs)
freq = get_kwarg("freq", kwargs)
risk_measure = get_kwarg("risk_measure", kwargs)
risk_free_rate = get_kwarg("risk_free_rate", kwargs)
alpha = get_kwarg("alpha", kwargs)
# Pop chart args
tangency = kwargs.pop("tangency", False)
@@ -403,7 +404,10 @@ def display_heat(**kwargs):
external_axes = kwargs.get("external_axes", None)
if len(weights) == 1:
console.print(f"Heatmap needs at least two values for '{category}'.")
single_key = list(weights.keys())[0].upper()
console.print(
f"[yellow]Heatmap needs at least two values for '{category}', only found '{single_key}'.[/yellow]"
)
return
if external_axes is None:
@@ -482,16 +486,17 @@ def display_rc(**kwargs):
weights = kwargs.get("weights", None)
data = kwargs.get("data", None)
colors = kwargs.get("colors", None)
risk_measure = kwargs.get("risk_measure", "MV")
risk_free_rate = kwargs.get("risk_free_rate", 0.0)
title = kwargs.get("title", "")
alpha = kwargs.get("alpha", 0.05)
a_sim = kwargs.get("a_sim", 100.0)
beta = kwargs.get("beta", 0.0)
b_sim = kwargs.get("b_sim", 0.0)
freq = kwargs.get("freq", "D")
external_axes = kwargs.get("external_axes", None)
risk_measure = get_kwarg("risk_measure", kwargs)
risk_free_rate = get_kwarg("risk_free_rate", kwargs)
alpha = get_kwarg("alpha", kwargs)
a_sim = get_kwarg("a_sim", kwargs)
beta = get_kwarg("beta", kwargs)
b_sim = get_kwarg("b_sim", kwargs)
freq = get_kwarg("freq", kwargs)
if external_axes is None:
_, ax = plt.subplots(figsize=plot_autoscale(), dpi=PLOT_DPI)
else:
@@ -533,9 +538,10 @@ def display_hist(**kwargs):
data = kwargs.get("data", None)
colors = kwargs.get("colors", None)
title = kwargs.get("title", "")
alpha = kwargs.get("alpha", 0.05)
external_axes = kwargs.get("external_axes", None)
alpha = kwargs.get("alpha", 0.05)
if external_axes is None:
_, ax = plt.subplots(figsize=plot_autoscale(), dpi=PLOT_DPI)
else:
@@ -571,9 +577,10 @@ def display_dd(**kwargs):
data = kwargs.get("data", None)
colors = kwargs.get("colors", None)
title = kwargs.get("title", "")
alpha = kwargs.get("alpha", 0.05)
external_axes = kwargs.get("external_axes", None)
alpha = get_kwarg("alpha", kwargs)
if external_axes is None:
_, ax = plt.subplots(figsize=plot_autoscale(), dpi=PLOT_DPI)
else:

334
openbb_terminal/portfolio/portfolio_optimization/statics.py
View File

@@ -1,3 +1,10 @@
from typing import List, Optional
from openbb_terminal.helper_funcs import get_rf
from openbb_terminal.portfolio.portfolio_optimization.parameters.Parameter import (
Parameter,
)
PERIOD_CHOICES = [
"1d",
"5d",
@@ -6,6 +13,7 @@ PERIOD_CHOICES = [
"6mo",
"1y",
"2y",
"3y",
"5y",
"10y",
"ytd",
@@ -234,25 +242,309 @@ DRAWDOWNS = [
"UCI_Rel",
]
PARAM_TYPES = {
"interval": str,
"start_date": str,
"end_date": str,
"log_returns": bool,
"freq": str,
"maxnan": float,
"threshold": float,
"method": str,
"risk_measure": str,
"objective": str,
"risk_free_rate": float,
"risk_aversion": float,
"alpha": float,
"target_return": float,
"target_risk": float,
"mean": str,
"covariance": str,
"d_ewma": float,
"value": float,
"value_short": float,
NAN_FILL_METHOD_CHOICES = [
"linear",
"time",
"nearest",
"zero",
"slinear",
"quadratic",
"cubic",
"barycentric",
]
TECHNIQUE_CHOICES = [
"maxsharpe",
"minrisk",
"maxutil",
"maxret",
"maxdiv",
"maxdecorr",
"ef",
"equal",
"mktcap",
"dividend",
"blacklitterman",
"riskparity",
"relriskparity",
"hrp",
"herc",
"nco",
]
# Parameter template files
OPTIMIZATION_PARAMETERS = {
"historic_period": Parameter(
name="historic_period",
type_=str,
default="3y",
choices=PERIOD_CHOICES,
),
"start_period": Parameter(
name="start_period",
type_=str,
default="",
),
"end_period": Parameter(
name="end_period",
type_=str,
default="",
),
"log_returns": Parameter(
name="log_returns",
type_=bool,
default=False,
),
"return_frequency": Parameter(
name="return_frequency",
type_=str,
default="d",
choices=FREQ_CHOICES,
),
"max_nan": Parameter(
name="max_nan",
type_=float,
default=0.05,
),
"threshold_value": Parameter(
name="threshold_value",
type_=float,
default=0.3,
),
"nan_fill_method": Parameter(
name="nan_fill_method",
type_=str,
default="time",
choices=NAN_FILL_METHOD_CHOICES,
),
"risk_free": Parameter(
name="risk_free",
type_=float,
default=get_rf(),
),
"significance_level": Parameter(
name="significance_level",
type_=float,
default=0.05,
),
"technique": Parameter(
name="technique",
type_=str,
default="maxsharpe",
choices=TECHNIQUE_CHOICES,
),
"risk_measure": Parameter(
name="risk_measure",
type_=str,
default="MV",
choices=MEAN_RISK_CHOICES + RISK_PARITY_CHOICES + HCP_CHOICES,
),
"target_return": Parameter(
name="target_return",
type_=float,
default=-1.0,
),
"target_risk": Parameter(
name="target_risk",
type_=float,
default=-1.0,
),
"expected_return": Parameter(
name="expected_return",
type_=str,
default="hist",
choices=MEAN_CHOICES,
),
"covariance": Parameter(
name="covariance",
type_=str,
default="hist",
choices=COVARIANCE_CHOICES,
),
"smoothing_factor_ewma": Parameter(
name="smoothing_factor_ewma",
type_=float,
default=0.94,
),
"long_allocation": Parameter(
name="long_allocation",
type_=float,
default=1.0,
),
"short_allocation": Parameter(
name="short_allocation",
type_=float,
default=0.0,
),
"risk_aversion": Parameter(
name="risk_aversion",
type_=float,
default=1.0,
),
"amount_portfolios": Parameter(
name="amount_portfolios",
type_=int,
default=100,
),
"random_seed": Parameter(
name="random_seed",
type_=int,
default=123,
),
"tangency": Parameter(
name="tangency",
type_=bool,
default=False,
),
"risk_contribution": Parameter(
name="risk_contribution",
type_=Optional[List[str]],
default=None,
),
"risk_parity_model": Parameter(
name="risk_parity_model",
type_=str,
default="A",
choices=REL_RISK_PARITY_CHOICES,
),
"penal_factor": Parameter(
name="penal_factor",
type_=float,
default=1.0,
),
"p_views": Parameter(
name="p_views",
type_=Optional[List[List[float]]],
default=None,
),
"q_views": Parameter(
name="q_views",
type_=Optional[List[List[float]]],
default=None,
),
"delta": Parameter(
name="delta",
type_=Optional[float],
default=None,
),
"equilibrium": Parameter(
name="equilibrium",
type_=bool,
default=True,
),
"optimize": Parameter(
name="optimize",
type_=bool,
default=True,
),
"co_dependence": Parameter(
name="co_dependence",
type_=str,
default="pearson",
choices=CODEPENDENCE_CHOICES,
),
"cvar_simulations_losses": Parameter(
name="cvar_simulations_losses",
type_=int,
default=100,
),
"cvar_simulations_gains": Parameter(
name="cvar_simulations_gains",
type_=Optional[int],
default=None,
),
"cvar_significance": Parameter(
name="cvar_significance",
type_=Optional[float],
default=None,
),
"linkage": Parameter(
name="linkage",
type_=str,
default="single",
choices=LINKAGE_CHOICES,
),
"amount_clusters": Parameter(
name="amount_clusters",
type_=Optional[int],
default=None,
),
"max_clusters": Parameter(
name="max_clusters",
type_=int,
default=10,
),
"amount_bins": Parameter(
name="amount_bins",
type_=str,
default="KN",
choices=BINS_CHOICES,
),
"alpha_tail": Parameter(
name="alpha_tail",
type_=float,
default=0.05,
),
"leaf_order": Parameter(
name="leaf_order",
type_=bool,
default=True,
),
"objective": Parameter(
name="objective",
type_=str,
default="MinRisk",
choices=OBJECTIVE_CHOICES + NCO_OBJECTIVE_CHOICES,
),
}
# Model functions
# TODO: The dictionary below should not be needed, we can input defaults in the Parameter directly
# But since template variable names are different from functions argument variable names
# we need to map them before removing this dictionary. When we standardize the names, we can remove this.
TERMINAL_TEMPLATE_MAP = {
"a_sim": "cvar_simulations_losses",
"alpha": "significance_level",
"alpha_tail": "alpha_tail",
"n_portfolios": "amount_portfolios",
"b_sim": "cvar_simulations_gains",
"beta": "cvar_significance",
"bins_info": "amount_bins",
"codependence": "co_dependence",
"covariance": "covariance",
"d_ewma": "smoothing_factor_ewma",
"delta": "delta",
"end_date": "end_period",
"equilibrium": "equilibrium",
"freq": "return_frequency",
"interval": "historic_period",
"k": "amount_clusters",
"leaf_order": "leaf_order",
"linkage": "linkage",
"log_returns": "log_returns",
"max_k": "max_clusters",
"maxnan": "max_nan",
"mean": "expected_return",
"method": "nan_fill_method",
"objective": "objective",
"optimize": "optimize",
"p_views": "p_views",
"penal_factor": "penal_factor",
"q_views": "q_views",
"seed": "random_seed",
"risk_aversion": "risk_aversion",
"risk_cont": "risk_contribution",
"risk_free_rate": "risk_free",
"risk_measure": "risk_measure",
"version": "risk_parity_model",
"start_date": "start_period",
"tangency": "tangency",
"target_return": "target_return",
"target_risk": "target_risk",
"technique": "technique",
"threshold": "threshold_value",
"value": "long_allocation",
"value_short": "short_allocation",
}