Restructure project architecture

Author: paty-oliveira

.github/workflows/pipeline.yml

@@ -23,5 +23,5 @@ jobs:
         uses: streamlit/streamlit-app-action@v0.0.3
         with:
           skip-smoke: 'true'
-          app-path: 'src/app.py'
+          app-path: 'src/main.py'
           ruff: 'true'

src/__init__.py

@@ -0,0 +1,9 @@
+from .yahoo import YahooFinance
+
+YAHOO = "yahoo-finance"
+
+
+def financial_data_provider(provider_name: str) -> object:
+    if provider_name is YAHOO:
+        return YahooFinance
+    raise ValueError(f"provider_name must be one of {['yahoo-finance']}")

src/app.py

@@ -1,38 +1,41 @@
 import logging
-from abc import ABC
 
 import pandas as pd
 import yfinance as yf
 
-from .FinancialDataProvider import FinancialDataProvider
-
 logging.basicConfig(filename="YahooFinanceProvider.log", level=logging.ERROR)
 
 
-class YahooFinanceProvider(FinancialDataProvider, ABC):
-    def __init__(self, ticker):
-        super().__init__(ticker)
-        self.stock = yf.Ticker(ticker)
-        self.general_info = self.stock.info
-        self.balance_sheet = self.stock.balance_sheet
-        self.income_stmt = self.stock.income_stmt
-        self.cash_flow = self.stock.cash_flow
+class YahooFinance:
+    @staticmethod
+    def is_valid_ticker(ticker: str) -> bool:
+        stock = yf.Ticker(ticker)
+        return stock.isin is not None and len(stock.info) > 1
 
-    def get_historical_data(self, period: str, interval: str) -> pd.DataFrame:
+    @staticmethod
+    def get_historical_data(
+        ticker: str, columns: [], period: str, interval: str
+    ) -> pd.DataFrame:
+        stock = yf.Ticker(ticker)
         try:
-            return self.stock.history(period=period, interval=interval)
+            return stock.history(period=period, interval=interval)[columns]
         except Exception as e:
             logging.error(
-                f"Error fetching historical data from Yahoo Finance for {self.stock}: {e}"
+                f"Error fetching historical data from Yahoo Finance for {stock}: {e}"
             )
 
-    def get_stock_info(self) -> dict:
+    @staticmethod
+    def get_stock_info(ticker: str) -> dict:
+        stock = yf.Ticker(ticker)
         try:
-            return self.general_info
+            return stock.info
         except Exception as e:
-            logging.error(f"Error fetching stock info for {self.stock}: {e}")
+            logging.error(f"Error fetching stock info for {stock}: {e}")
+
+    @staticmethod
+    def get_balance_sheet(ticker: str) -> pd.DataFrame:
+        stock = yf.Ticker(ticker)
 
-    def get_balance_sheet(self) -> pd.DataFrame:
         try:
             fields = [
                 "Current Assets",
@@ -47,23 +50,26 @@ def get_balance_sheet(self) -> pd.DataFrame:
                 "Stockholders Equity",
             ]
 
-            self.balance_sheet.loc["Debt to Equity"] = (
-                self.balance_sheet.loc["Total Liabilities Net Minority Interest"]
-                / self.balance_sheet.loc["Stockholders Equity"]
+            stock.balance_sheet.loc["Debt to Equity"] = (
+                stock.balance_sheet.loc["Total Liabilities Net Minority Interest"]
+                / stock.balance_sheet.loc["Stockholders Equity"]
             )
-            self.balance_sheet.loc["Current Ratio"] = (
-                self.balance_sheet.loc["Current Assets"]
-                / self.balance_sheet.loc["Current Liabilities"]
+            stock.balance_sheet.loc["Current Ratio"] = (
+                stock.balance_sheet.loc["Current Assets"]
+                / stock.balance_sheet.loc["Current Liabilities"]
             )
 
-            df_balance_sheet = self.balance_sheet.transpose()
+            df_balance_sheet = stock.balance_sheet.transpose()
             df_balance_sheet = df_balance_sheet.reindex(columns=fields, fill_value=0)
 
             return df_balance_sheet.sort_index().tail(4)
         except Exception as e:
-            logging.error(f"Error fetching balance sheet for {self.stock}: {e}")
+            logging.error(f"Error fetching balance sheet for {stock}: {e}")
+
+    @staticmethod
+    def get_income_statement(ticker: str) -> pd.DataFrame:
+        stock = yf.Ticker(ticker)
 
-    def get_income_statement(self) -> pd.DataFrame:
         try:
             fields = [
                 "Total Revenue",
@@ -81,15 +87,17 @@ def get_income_statement(self) -> pd.DataFrame:
                 "Basic EPS",
             ]
 
-            df_income_stmt = self.income_stmt.transpose()
+            df_income_stmt = stock.income_stmt.transpose()
             df_income_stmt = df_income_stmt.reindex(columns=fields, fill_value=0)
 
             return df_income_stmt.sort_index().tail(4)
 
         except Exception as e:
-            logging.error(f"Error fetching income statement for {self.stock}: {e}")
+            logging.error(f"Error fetching income statement for {stock}: {e}")
 
-    def get_cash_flow(self):
+    @staticmethod
+    def get_cashflow(ticker: str):
+        stock = yf.Ticker(ticker)
         try:
             fields = [
                 "Operating Cash Flow",
@@ -98,10 +106,10 @@ def get_cash_flow(self):
                 "Capital Expenditure",
                 "Free Cash Flow",
             ]
-            df_cashflow = self.cash_flow.transpose()
+            df_cashflow = stock.cash_flow.transpose()
             df_cashflow = df_cashflow.reindex(columns=fields, fill_value=0)
 
             return df_cashflow.sort_index().tail(4)
 
         except Exception as e:
-            logging.error(f"Error fetching cash flow for {self.stock}: {e}")
+            logging.error(f"Error fetching cash flow for {stock}: {e}")

src/infra/__init__.py

@@ -0,0 +1,19 @@
+import infra
+import ui
+
+_financial_provider = None
+
+
+def initialize_financial_provider(provider_name):
+    global _financial_provider
+    if _financial_provider is None:
+        _financial_provider = infra.financial_data_provider(provider_name)
+
+
+def main():
+    initialize_financial_provider(infra.YAHOO)
+    ui.run(_financial_provider)
+
+
+if __name__ == "__main__":
+    main()

src/ui/data_providers/YahooFinanceProvider.py renamed to src/infra/yahoo.py

@@ -0,0 +1,23 @@
+import streamlit as st
+
+from .layout import render as render_layout
+from .sidebar import render as render_sidebar
+
+initial_state = {"ticker": ""}
+
+
+def run(financial_data):
+    st.set_page_config(
+        page_title="Stock Analysis Dashboard",
+        page_icon=":chart_with_upwards_trend:",
+        layout="wide",
+        initial_sidebar_state="expanded",
+    )
+
+    for k, v in initial_state.items():
+        if k not in st.session_state:
+            st.session_state[k] = v
+
+    # performs the rendering
+    render_layout(financial_data)
+    render_sidebar()