Spaces:
Sleeping
Sleeping
File size: 5,155 Bytes
d31af6a |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 |
import pandas as pd
import yfinance as yf
from pypfopt import EfficientFrontier
from pypfopt import risk_models
from pypfopt import expected_returns
from pypfopt import HRPOpt, hierarchical_portfolio
class CompData:
def __init__(self, company_data):
"""
Class that manages company and stock data
"""
self.df = company_data
self.company_names = self.df["Name"].to_list()
self.company_symbols = (self.df["Ticker"] + ".NS").to_list()
# utilities for tranlation
name_to_id_dict = dict()
id_to_name_dict = dict()
for CSymbol, CName in zip(self.company_symbols, self.company_names):
name_to_id_dict[CName] = CSymbol
for CSymbol, CName in zip(self.company_symbols, self.company_names):
id_to_name_dict[CSymbol] = CName
self.name_to_id = name_to_id_dict
self.id_to_name = id_to_name_dict
def fetch_stock_data(self, company_ids: list, start_date: str) -> pd.DataFrame:
"""
Use yfinance client sdk to fetch stock data from the yahoo finance api
"""
company_data = pd.DataFrame()
# get the stock data for the companies
for cname in company_ids:
stock_data_temp = yf.download(
cname, start=start_date, end=pd.Timestamp.now().strftime("%Y-%m-%d")
)["Adj Close"]
stock_data_temp.name = cname
company_data = pd.merge(
company_data,
stock_data_temp,
how="outer",
right_index=True,
left_index=True,
)
# cleaning the data
company_data.dropna(axis=1, how="all", inplace=True)
company_data.dropna(inplace=True)
for i in company_data.columns:
company_data[i] = company_data[i].abs()
return company_data
def comp_id_to_name(self, list_of_ids: list):
return [self.id_to_name[i] for i in list_of_ids]
def comp_name_to_id(self, list_of_names: list):
return [self.name_to_id[i] for i in list_of_names]
class PortfolioOptimizer:
def __init__(self, comp_data: CompData, company_ids: list, start_date: str):
self.comp_data = comp_data
self.stock_data = self.comp_data.fetch_stock_data(
company_ids, start_date)
self.stock_data_returns = self.stock_data.pct_change().dropna()
def optimize(self, method: str, ef_parameter=None):
company_asset_weights = 0
# Do the portfolio optimization
if method == "Efficient Frontier":
mu = expected_returns.mean_historical_return(self.stock_data)
S = risk_models.sample_cov(self.stock_data)
self.ef = EfficientFrontier(mu, S)
if ef_parameter == "Maximum Sharpe Raio":
self.ef.max_sharpe()
elif ef_parameter == "Minimum Volatility":
self.ef.min_volatility()
elif ef_parameter == "Efficient Risk":
self.ef.efficient_risk(0.5)
else:
self.ef.efficient_return(0.05)
company_asset_weights = pd.DataFrame.from_dict(
self.ef.clean_weights(), orient="index"
).reset_index()
elif method == "Hierarchical Risk Parity":
mu = expected_returns.returns_from_prices(self.stock_data)
S = risk_models.sample_cov(self.stock_data)
self.ef = HRPOpt(mu, S)
company_asset_weights = self.ef.optimize()
company_asset_weights = pd.DataFrame.from_dict(
company_asset_weights, orient="index", columns=["Weight"]
).reset_index()
# cleaning the returned data from the optimization
company_asset_weights.columns = ["Ticker", "Allocation"]
company_asset_weights["Name"] = self.comp_data.comp_id_to_name(
company_asset_weights["Ticker"])
company_asset_weights = company_asset_weights[[
"Name", "Ticker", "Allocation"]]
return company_asset_weights
def get_portfolio_performance(self):
if self.ef is not None:
(
expected_annual_return,
annual_volatility,
sharpe_ratio,
) = self.ef.portfolio_performance()
st_portfolio_performance = pd.DataFrame.from_dict(
{
"Expected annual return": (expected_annual_return * 100).round(2),
"Annual volatility": (annual_volatility * 100).round(2),
"Sharpe ratio": sharpe_ratio.round(2),
},
orient="index",
).reset_index()
st_portfolio_performance.columns = ["Metrics", "Summary"]
return st_portfolio_performance
else:
return None
def get_portfolio_returns(self):
return (
self.stock_data_returns * list(self.ef.clean_weights().values())
).sum(axis=1)
def get_annual_portfolio_returns(self):
return self.get_portfolio_returns().resample("Y").apply(lambda x: (x + 1).prod() - 1)
|