Introduced data management classes to seperate backend from frontend

utilities/py/data_management.py

@@ -0,0 +1,151 @@
+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)