src/fetch_data.py

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