took out fetch_data from technical analysis to its own module

app.py

@@ -1,6 +1,7 @@
 import dotenv
 import gradio as gr
 
+from src.fetch_data import FetchData
 from src.fetch_forecast import FetchForecast
 from src.stock_analysis_agent import StockAnalyst
 from src.technical_analysis import TechnicalAnalysis
@@ -48,11 +49,12 @@ def plot_stock(agent_instance, ticker):
     if ticker_exists:
         # fetch the forecasts
         dotenv.load_dotenv(dotenv.find_dotenv())
+        df_hist = FetchData(ticker, fetchperiodinweeks=12).run()
         df_past, df_fcst = FetchForecast(ticker).run()
         # run the technical analysis
         _, fig = TechnicalAnalysis(
             ticker=ticker,
-            fetchperiodinweeks=12,
+            df_hist=df_hist,
             df_past=df_past,
             df_fcst=df_fcst,
             plot_ta=True,

src/fetch_data.py

@@ -0,0 +1,64 @@
+
+import datetime as dt
+import logging
+import os
+
+import pandas as pd
+import yfinance as yf
+
+
+class FetchData:
+    def __init__(self, ticker: str, fetchperiodinweeks: int = 12, debug=False):
+        # set up logging
+        if debug:
+            self.logger_level = logging.DEBUG
+        else:
+            self.logger_level = logging.INFO
+        self.logger = logging.getLogger(__name__)
+        logging.basicConfig(level=self.logger_level)  # filename='TechnicalAnalysis.log',
+
+        self.ticker = ticker
+        self.fetchperiodinweeks = fetchperiodinweeks
+    
+    def run(
+        self
+        ) -> pd.DataFrame:
+        """
+        Fetches historical stock price data from Yahoo Finance.
+        This method downloads historical stock price data for the specified
+        ticker over a given period of weeks. The data is fetched on a daily
+        interval and stored in a pandas DataFrame. If the download is successful,
+        redundant ticker columns are removed, and logging information is
+        recorded. In case of failure, an empty DataFrame is returned and an
+        exception is raised.
+        Returns:
+        pd.DataFrame
+            A DataFrame containing the historical stock price data with columns
+            for open, high, low, close, volume, and adjusted close prices.
+        Raises:
+        Exception
+            If the data fetching fails, an exception is raised with an error
+            message.
+        """
+
+        period_start = dt.datetime.now() - dt.timedelta(weeks=self.fetchperiodinweeks)
+        self.logger.info(f'Fetching price data for {self.ticker}')
+        try:
+            df = yf.download(
+                self.ticker,
+                start=period_start,
+                end=dt.datetime.now(),
+                interval='1d'
+            )
+        except Exception as e:
+            self.logger.error(f'{e}')
+            # create empty df
+            df = pd.DataFrame()
+
+        if df.shape[0] > 0:
+            # get rid of the redundant ticker column
+            df.columns = df.columns.droplevel('Ticker')
+            self.logger.debug(df.head(10))
+            self.logger.info(f'Fetched {df.shape[0]} rows for {self.ticker}')
+        
+        return df

src/stock_analysis_agent.py

@@ -12,6 +12,7 @@ from langgraph.graph.message import add_messages
 import logging
 from pydantic import BaseModel, Field
 
+from src.fetch_data import FetchData
 from src.fetch_forecast import FetchForecast
 from src.technical_analysis import TechnicalAnalysis
 from src.fundamental_analysis import FundamentalAnalysis
@@ -27,10 +28,11 @@ def get_stock_prices(
     ticker: str
         The stock ticker symbol to fetch data for.
     """
+    df_hist = FetchData(ticker, fetchperiodinweeks=12).run()
     df_past, df_fcst = FetchForecast(ticker).run()
     df, _ = TechnicalAnalysis(
         ticker=ticker,
-        fetchperiodinweeks=12,
+        df_hist=df_hist,
         df_past=df_past,
         df_fcst=df_fcst,
         plot_ta=False,

src/technical_analysis.py

@@ -1,25 +1,22 @@
-import datetime as dt
 import logging
+import os
 
 import dotenv
 import matplotlib.pyplot as plt
 import matplotlib.dates as mdates
 from matplotlib.axes import Axes
 import numpy as np
-import os
 import pandas as pd
 from ta.volume import volume_weighted_average_price
 from ta.momentum import RSIIndicator, StochasticOscillator
 from ta.trend import MACD 
-import yfinance as yf
 
-from src.fetch_forecast import FetchForecast
 
 class TechnicalAnalysis():
     def __init__(
             self,
-            ticker:str,
-            fetchperiodinweeks:int=12,
+            ticker: str,
+            df_hist: pd.DataFrame,
             df_past=None,
             df_fcst=None,
             plot_ta:bool=True,
@@ -31,8 +28,8 @@ class TechnicalAnalysis():
         Args:
         ticker : str
             stock ticker to analyze
-        fetchperiodinweeks : int, optional, default: 8
-            number of weeks to fetch historical data from YahooFinance
+        df_hist: pd.DataFrame
+            historical price data for ticker
         df_past: pd.DataFrame, optional, default: None
             Closeing price of the ticker for the past few days
         df_fcst: pd.DataFrame, optional, default: None
@@ -52,7 +49,7 @@ class TechnicalAnalysis():
 
         # input arguments
         self.ticker = ticker
-        self.fetchperiodinweeks = fetchperiodinweeks
+        self.df_hist = df_hist
         self.df_past = df_past
         self.df_fcst = df_fcst
         self.plot_ta = plot_ta
@@ -66,12 +63,10 @@ class TechnicalAnalysis():
         """
         Main entry point for the TechnicalAnalysis object.
         This method:
-        - fetches historical data from YahooFinance,
         - computes the technical analysis metrics
         - plots the price and TA metrics.
         """
-        # fetch data from yf
-        df = self.fetch_data()
+        df = self.df_hist
         # add the features based on technical analysis
         if df.shape[0] > 0:
             df = self.tech_analysis(df)
@@ -100,48 +95,6 @@ class TechnicalAnalysis():
 
         return df, fig
     
-    def fetch_data(
-            self
-            ) -> pd.DataFrame:
-        """
-        Fetches historical stock price data from Yahoo Finance.
-        This method downloads historical stock price data for the specified
-        ticker over a given period of weeks. The data is fetched on a daily
-        interval and stored in a pandas DataFrame. If the download is successful,
-        redundant ticker columns are removed, and logging information is
-        recorded. In case of failure, an empty DataFrame is returned and an
-        exception is raised.
-        Returns:
-        pd.DataFrame
-            A DataFrame containing the historical stock price data with columns
-            for open, high, low, close, volume, and adjusted close prices.
-        Raises:
-        Exception
-            If the data fetching fails, an exception is raised with an error
-            message.
-        """
-
-        period_start = dt.datetime.now() - dt.timedelta(weeks=self.fetchperiodinweeks)
-        self.logger.info(f'Fetching price data for {self.ticker}')
-        try:
-            df = yf.download(
-                self.ticker,
-                start=period_start,
-                end=dt.datetime.now(),
-                interval='1d'
-            )
-        except Exception as e:
-            self.logger.error(f'{e}')
-            # create empty df
-            df = pd.DataFrame()
-
-        if df.shape[0] > 0:
-            # get rid of the redundant ticker column
-            df.columns = df.columns.droplevel('Ticker')
-            self.logger.debug(df.head(10))
-            self.logger.info(f'Fetched {df.shape[0]} rows for {self.ticker}')
-        
-        return df
     
     def tech_analysis(
             self,
@@ -390,10 +343,13 @@ class TechnicalAnalysis():
 
 if __name__ == '__main__':
     ticker = 'AAPL'
-    # fetch the forecasts
+    # testing
+    from src.fetch_forecast import FetchForecast
+    from src.fetch_data import FetchData
     dotenv.load_dotenv(dotenv.find_dotenv())
+    df_hist = FetchData(ticker, fetchperiodinweeks=12).run()
     df_past, df_fcst = FetchForecast(ticker).run()
-    df, fig = TechnicalAnalysis(ticker, df_past=df_past, df_fcst=df_fcst, plot_ta=True, savefig=True, debug=False).run()
+    df, fig = TechnicalAnalysis(ticker, df_hist=df_hist, df_past=df_past, df_fcst=df_fcst, plot_ta=True, savefig=True, debug=False).run()
     # print(f'columns: {df.columns}')