Merge pull request #2 from Carsten134/refactoring

.gitignore

@@ -6,3 +6,9 @@
 *.idea
 *app.py
 *.env
+
+# VSCode settings
+.vscode
+
+# Pycache
+*/__pycache__

main.py

@@ -1,255 +1,18 @@
-import yfinance as yf
-import numpy as np
 import pandas as pd
 
-import streamlit as st
-
 from utilities.py.styling import streamlit_style
-from utilities.py import plots
-from utilities.py import summary_tables
-# from utilities.py import mongodb
-
-from pypfopt import EfficientFrontier
-from pypfopt import risk_models
-from pypfopt import expected_returns
-from pypfopt import HRPOpt, hierarchical_portfolio
+from utilities.py.composer import CapiPortApp
 
-import plotly.express as px
-import plotly.graph_objects as go
 
 streamlit_style()
-
+# data import
 company_list_df = pd.read_csv("utilities/data/Company List.csv")
 
-company_name = company_list_df["Name"].to_list()
-company_symbol = (company_list_df["Ticker"] + ".NS").to_list()
-
-name_to_symbol_dict = dict()
-symbol_to_name_dict = dict()
-
-for CSymbol, CName in zip(company_symbol, company_name):
-    name_to_symbol_dict[CName] = CSymbol
-
-for CSymbol, CName in zip(company_symbol, company_name):
-    symbol_to_name_dict[CSymbol] = CName
-
-streamlit_company_list_input = st.multiselect(
-    "Select Multiple Companies", company_name, default=None
-)
-
-optimisation_method = st.selectbox(
-    "Choose an optimization method accordingly",
-    (
-        "Efficient Frontier",
-        "Hierarchical Risk Parity",
-    ),
-)
-
-parameter_for_optimisation = 0
-if optimisation_method == "Efficient Frontier":
-    parameter_for_optimisation = st.selectbox(
-        "Choose an optimization parameter accordingly",
-        (
-            "Maximum Sharpe Ratio",
-            "Efficient Risk",
-            "Minimum Volatility",
-            "Efficient Return",
-        ),
-    )
-
-company_name_to_symbol = [name_to_symbol_dict[i] for i in streamlit_company_list_input]
-
-number_of_symbols = len(company_name_to_symbol)
-
-start_date = st.date_input(
-    "Start Date",
-    format="YYYY-MM-DD",
-    value=pd.Timestamp("1947-08-15"),
-    max_value=pd.Timestamp.now(),
-)
-
-initial_investment = st.number_input("How much would you want to invest?", value=45000)
-
-if number_of_symbols > 1:
-    company_data = pd.DataFrame()
-
-    for cname in company_name_to_symbol:
-        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,
-        )
-
-    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()
-
-    st.write(
-        f"Note: Due to unavailability of full data, this Analysis uses data from the date: {company_data.index[0]}"
-    )
-
-    number_of_symbols = len(company_data.columns)
-
-    st.dataframe(company_data, use_container_width=True)
-
-    if number_of_symbols > 1:
-        company_stock_returns_data = company_data.pct_change().dropna()
-
-        mu = 0
-        S = 0
-        ef = 0
-        company_asset_weights = 0
-
-        if optimisation_method == "Efficient Frontier":
-            mu = expected_returns.mean_historical_return(company_data)
-            S = risk_models.sample_cov(company_data)
-
-            ef = EfficientFrontier(mu, S)
-
-            if parameter_for_optimisation == "Maximum Sharpe Raio":
-                ef.max_sharpe()
-            elif parameter_for_optimisation == "Minimum Volatility":
-                ef.min_volatility()
-            elif parameter_for_optimisation == "Efficient Risk":
-                ef.efficient_risk(0.5)
-            else:
-                ef.efficient_return(0.05)
-
-            company_asset_weights = pd.DataFrame.from_dict(
-                ef.clean_weights(), orient="index"
-            ).reset_index()
-        elif optimisation_method == "Hierarchical Risk Parity":
-            mu = expected_returns.returns_from_prices(company_data)
-            S = risk_models.sample_cov(company_data)
-
-            ef = HRPOpt(mu, S)
-
-            company_asset_weights = ef.optimize()
-            company_asset_weights = pd.DataFrame.from_dict(
-                company_asset_weights, orient="index", columns=["Weight"]
-            ).reset_index()
-
-        company_asset_weights.columns = ["Ticker", "Allocation"]
-
-        company_asset_weights_copy = company_asset_weights
-
-        company_asset_weights["Name"] = [
-            symbol_to_name_dict[i] for i in company_asset_weights["Ticker"]
-        ]
-
-        company_asset_weights = company_asset_weights[["Name", "Ticker", "Allocation"]]
-
-        st.dataframe(company_asset_weights, use_container_width=True)
-
-        ef.portfolio_performance()
-
-        (
-            expected_annual_return,
-            annual_volatility,
-            sharpe_ratio,
-        ) = 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"]
-
-        if optimisation_method == "Efficient Frontier":
-            st.write(
-                "Optimization Method - ",
-                optimisation_method,
-                "---- Parameter - ",
-                parameter_for_optimisation,
-            )
-        else:
-            st.write("Optimization Method - ", optimisation_method)
-
-        st.dataframe(st_portfolio_performance, use_container_width=True)
-
-        plots.pie_chart_company_asset_weights(company_asset_weights)
-
-        portfolio_returns = (
-            company_stock_returns_data * list(ef.clean_weights().values())
-        ).sum(axis=1)
-
-        annual_portfolio_returns = portfolio_returns.resample("Y").apply(
-            lambda x: (x + 1).prod() - 1
-        )
-
-        cumulative_returns = (portfolio_returns + 1).cumprod() * initial_investment
-
-        tab1, tab2, tab3 = st.tabs(["Plots", "Annual Returns", "Montly Returns"])
-
-        with tab1:
-            plots.plot_annual_returns(annual_portfolio_returns)
-            plots.plot_cummulative_returns(cumulative_returns)
-
-        with tab2:
-            annual_portfolio_returns = summary_tables.annual_returns_dataframe(
-                annual_portfolio_returns
-            )
-            annual_cumulative_returns = (
-                summary_tables.annual_cumulative_returns_dataframe(cumulative_returns)
-            )
-            annual_stock_returns = summary_tables.company_wise_annual_return(
-                company_stock_returns_data, company_asset_weights
-            )
-
-            merged_annual_returns_data = pd.merge(
-                annual_portfolio_returns,
-                annual_cumulative_returns,
-                on="Year",
-                suffixes=("_portfolio", "_cumulative"),
-            )
-
-            merged_annual_returns_data = pd.merge(
-                merged_annual_returns_data, annual_stock_returns, on="Year"
-            )
-
-            st.write("Annual Returns")
-            st.dataframe(merged_annual_returns_data, use_container_width=True)
-
-        with tab3:
-            monthly_portfolio_return = summary_tables.monthly_returns_dataframe(
-                portfolio_returns
-            )
-            monthly_stock_return = summary_tables.company_wise_monthly_return(
-                company_stock_returns_data, company_asset_weights
-            )
-            monthly_cumulative_returns = (
-                summary_tables.monthly_cumulative_returns_dataframe(cumulative_returns)
-            )
-
-            merged_monthly_returns_data = pd.merge(
-                monthly_portfolio_return,
-                monthly_cumulative_returns,
-                on=["Year", "Month"],
-                how="inner",
-            )
+capi_port = CapiPortApp(company_list_df)
 
-            merged_monthly_returns_data = pd.merge(
-                merged_monthly_returns_data,
-                monthly_stock_return,
-                on=["Year", "Month"],
-                how="inner",
-            )
+capi_port.render_user_input()
 
-            st.write("Montly Return")
-            st.dataframe(merged_monthly_returns_data, use_container_width=True)
+ready_to_render_results = len(capi_port.user_input.get_selected_comp_ids()) > 1
 
-        # mongodb.mongodb_push_data(company_name_to_symbol, number_of_symbols)
+if ready_to_render_results:
+    capi_port.render_results()

utilities/py/composer.py

@@ -0,0 +1,140 @@
+import pandas as pd
+import streamlit as st
+
+from utilities.py.data_management import CompData, PortfolioOptimizer
+from utilities.py.ui_elements import UserInput
+from utilities.py import plots, summary_tables
+
+
+class CapiPortApp:
+    def __init__(self, company_df):
+        """
+        Class for composing the apps UI on a high level. Is meant to provide a readable overview, of what happens inside the app, without being concerned with unnecessary details.
+
+        Uppon initializing the composer fetches the data within the apps source code (for now) and let's a dedicated class handle the raw data.
+        """
+        self.comp_data = CompData(company_df)
+
+        # get all the necessary ui elements
+        self.user_input = UserInput(self.comp_data)
+
+    def render_user_input(self):
+        self.user_input.company_selection()
+        self.user_input.opt_method_selection()
+        self.user_input.start_date()
+        self.user_input.initial_investment()
+      
+
+    def render_results(self):
+        """
+        CAUTION: the composer assumes, that the user has given all the necessary data.
+
+        NECESSARY DATA:
+            - at least two companies
+            - an optimization method
+            - a start date
+            - initial investment
+
+        Check the conditions the necessary conditions beforehand :)
+        """
+        # fetch user input
+        user_input_data = self.user_input.get_user_input_data()
+
+        # optimize the chosen portfolio according to the specifications
+        portfolio_opt = PortfolioOptimizer(self.comp_data,
+                                           self.user_input.get_selected_comp_ids(),
+                                           user_input_data.start_date)
+
+        company_asset_weights = portfolio_opt.optimize(user_input_data.opt_method,
+                                                       user_input_data.ef_parameter)
+
+        # show first the stock data...
+        st.dataframe(portfolio_opt.stock_data, use_container_width=True)
+
+        # print disclaimer
+        first_date_available = portfolio_opt.stock_data.index[0]
+        opt_header = f"Used {user_input_data.opt_method}" if user_input_data.ef_parameter is None else f"Used {user_input_data.opt_method} with {user_input_data.ef_parameter}"
+
+
+        st.write(
+            f"Note: Due to unavailability of full data, this Analysis uses data from the date: {first_date_available}")
+
+        st.write(opt_header)
+
+        # show asset weights, portfolio performance and the pie chart
+        st.dataframe(company_asset_weights, use_container_width=True)
+
+        st.dataframe(portfolio_opt.get_portfolio_performance(),
+                     use_container_width=True)
+
+        plots.pie_chart_company_asset_weights(company_asset_weights)
+
+        # summarize the resulting data
+        portfolio_returns = portfolio_opt.get_portfolio_returns()
+        annual_portfolio_returns = portfolio_opt.get_annual_portfolio_returns()
+
+        cumulative_returns = (portfolio_returns +
+                              1).cumprod() * user_input_data.init_invest
+
+        # render the tabs
+        tab1, tab2, tab3 = st.tabs(
+            ["Plots", "Annual Returns", "Montly Returns"])
+
+        with tab1:
+            plots.plot_annual_returns(annual_portfolio_returns)
+            plots.plot_cummulative_returns(cumulative_returns)
+
+        with tab2:
+            annual_portfolio_returns = summary_tables.annual_returns_dataframe(
+                annual_portfolio_returns
+            )
+            annual_cumulative_returns = (
+                summary_tables.annual_cumulative_returns_dataframe(
+                    cumulative_returns)
+            )
+            annual_stock_returns = summary_tables.company_wise_annual_return(
+                portfolio_opt.stock_data_returns, company_asset_weights
+            )
+
+            merged_annual_returns_data = pd.merge(
+                annual_portfolio_returns,
+                annual_cumulative_returns,
+                on="Year",
+                suffixes=("_portfolio", "_cumulative"),
+            )
+
+            merged_annual_returns_data = pd.merge(
+                merged_annual_returns_data, annual_stock_returns, on="Year"
+            )
+
+            st.write("Annual Returns")
+            st.dataframe(merged_annual_returns_data, use_container_width=True)
+
+        with tab3:
+            monthly_portfolio_return = summary_tables.monthly_returns_dataframe(
+                portfolio_returns
+            )
+            monthly_stock_return = summary_tables.company_wise_monthly_return(
+                portfolio_opt.stock_data_returns, company_asset_weights
+            )
+            monthly_cumulative_returns = (
+                summary_tables.monthly_cumulative_returns_dataframe(
+                    cumulative_returns)
+            )
+
+            merged_monthly_returns_data = pd.merge(
+                monthly_portfolio_return,
+                monthly_cumulative_returns,
+                on=["Year", "Month"],
+                how="inner",
+            )
+
+            merged_monthly_returns_data = pd.merge(
+                merged_monthly_returns_data,
+                monthly_stock_return,
+                on=["Year", "Month"],
+                how="inner",
+            )
+
+            st.write("Montly Return")
+            st.dataframe(merged_monthly_returns_data, use_container_width=True)

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)

utilities/py/ui_elements.py

@@ -0,0 +1,75 @@
+import pandas as pd
+
+import streamlit as st
+from utilities.py.data_management import CompData
+
+class UserInputData:
+    def __init__(self, company_list, opt_method, start_date, init_invest, ef_parameter = None):
+        self.company_list = company_list
+        self.opt_method = opt_method
+        self.ef_parameter = ef_parameter
+        self.start_date = start_date
+        self.init_invest = init_invest
+
+
+class UserInput:
+
+    def __init__(self, company_data: CompData):
+        """
+        Class that renders the user selection (company, optimization technique, etc.)
+        """
+        self.comp_data = company_data
+        self.ef_parameter_input = None
+        self.company_list_input = None
+        self.opt_method_input = None
+        self.start_date_input = None
+        self.initial_investment_input = None
+
+    def company_selection(self):
+        self.company_list_input = st.multiselect(
+            "Select Multiple Companies", self.comp_data.company_names, default=None
+        )
+
+    def opt_method_selection(self):
+        self.opt_method_input = st.selectbox(
+            "Choose an optimization method accordingly",
+            (
+                "Efficient Frontier",
+                "Hierarchical Risk Parity",
+            ),
+        )
+        
+        if self.opt_method_input == "Efficient Frontier":
+            self.ef_parameter_input = st.selectbox(
+                "Choose an optimization parameter accordingly",
+                (
+                    "Maximum Sharpe Ratio",
+                    "Efficient Risk",
+                    "Minimum Volatility",
+                    "Efficient Return",
+                ),
+            )
+
+    def start_date(self):
+        self.start_date_input = st.date_input(
+            "Start Date",
+            format="YYYY-MM-DD",
+            value=pd.Timestamp("1947-08-15"),
+            max_value=pd.Timestamp.now(),
+        )
+
+    def initial_investment(self):
+        self.innit_invest_input = st.number_input("How much would you want to invest?", value=45000)
+
+    def get_selected_comp_ids(self):
+        if self.company_list_input is not None:
+            return self.comp_data.comp_name_to_id(self.company_list_input)
+        print("WARINING: Selected company ids accessed, eventhough company not yet rendered in UI")
+        return None
+
+    def get_user_input_data(self) -> UserInputData:
+        return UserInputData(self.company_list_input,
+                             self.opt_method_input,
+                             self.start_date_input,
+                             self.innit_invest_input,
+                             self.ef_parameter_input)