introduced src.technical_analysis.py

src/technical_analysis.py

@@ -0,0 +1,277 @@
+import datetime as dt
+import logging
+
+import matplotlib.pyplot as plt
+import matplotlib.dates as mdates
+from matplotlib.axes import Axes
+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
+
+
+class TechnicalAnalysis():
+    def __init__(
+            self,
+            ticker:str,
+            fetchperiodinweeks:int=8,
+            plot_ta:bool=True,
+            debug=False):
+        # input arguments
+        """
+        Initialize TechnicalAnalysis object.
+        Args:
+        ticker : str
+            stock ticker to analyze
+        fetchperiodinweeks : int, optional, default: 8
+            number of weeks to fetch historical data from YahooFinance
+        plot_ta : bool, optional, default: True
+            whether to generate plots of technical analysis metrics. Plot will be created under plots/{ticker}.png
+        debug : bool, optional, default: False
+            whether run in debug mode, so that logging should be produced at debug level
+        """
+        self.ticker = ticker
+        self.fetchperiodinweeks = fetchperiodinweeks
+        self.plot_ta = plot_ta
+        
+        # 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', 
+
+    def run(
+            self
+            ) -> None:
+        # fetch data from yf
+        """
+        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.
+        """
+        self.df = self.fetch_data()
+        # add the features based on technical analysis
+        self.df = self.tech_analysis()
+        # plot the results
+        if self.plot_ta:
+            os.makedirs('plots', exist_ok=True)
+            self.plot_stock_metrics(
+                self.df,
+                datasets={
+                        'Volume': ['Volume'],
+                        'Prices': ['Close', 'VWAP'],  # 'High','Low', 
+                        'Indices': ['RSI', 'StochOsc'],
+                        'Trend': ['MACD', 'MACDsig', 'MACDdif']}
+                        )
+        
+
+    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 {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 {ticker}')
+        else:
+            raise Exception(f'Failed to fetch data for {ticker}')
+        
+        return df
+    
+    def tech_analysis(
+            self
+            ) -> pd.DataFrame:
+        """
+        Calculates technical analysis indicators for the fetched stock price data.
+        This method takes the fetched stock price data and calculates several
+        technical analysis indicators. The following indicators are calculated:
+        - Additional Price Indicators:
+          - Volume-Weighted Average Price (VWAP)
+        - Momentum Indicators:
+          - Relative Strength Index (RSI)
+            - Stochastic Oscillator
+        - Trend Indicators: 
+            - Moving Average Convergence Divergence (MACD)
+        The calculated indicators are added to the DataFrame as new columns.
+        Returns:
+        pd.DataFrame
+            The DataFrame with the calculated technical analysis indicators.
+        """
+        df = self.df
+
+        # Price Indicators
+        # Volume-Weighted Average Price (VWAP)
+        # https://chartschool.stockcharts.com/table-of-contents/technical-indicators-and-overlays/technical-overlays/volume-weighted-average-price-vwap
+        df['VWAP'] = volume_weighted_average_price(
+            high=df['High'],
+            low=df['Low'],
+            close=df['Close'],
+            volume=df['Volume'],
+        )
+
+        # Indices
+        # RSI:
+        # https://www.investopedia.com/terms/r/rsi.asp
+        df['RSI'] = RSIIndicator(
+            df['Close'],
+            window=14).rsi()
+        # Stochastic Oscillator: 
+        # https://chartschool.stockcharts.com/table-of-contents/technical-indicators-and-overlays/technical-indicators/stochastic-oscillator-fast-slow-and-full
+        df['StochOsc'] = StochasticOscillator(
+            df['High'],
+            df['Low'],
+            df['Close'],
+            window=14).stoch()
+
+        # Trend signals
+        # Moving Average Convergence Divergence (MACD):
+        # https://chartschool.stockcharts.com/table-of-contents/technical-indicators-and-overlays/technical-indicators/macd-moving-average-convergence-divergence-oscillator
+        macd = MACD(
+            df['Close'],
+            window_slow=26,
+            window_fast=12,
+            window_sign=9)
+        df['MACD'] = macd.macd()
+        df['MACDsig'] = macd.macd_signal()
+        df['MACDdif'] = macd.macd_diff()
+
+        return df
+
+    def plot_stock_metrics(
+            self,
+            df,
+            datasets={
+                'Volume': ['Volume'],
+                'Price': ['Close']  # 'High','Low'
+                }
+                ) -> None:
+        """
+        Plots the given stock metrics datasets as subplots.
+        This method takes in a DataFrame and a dictionary of datasets, where
+        each key is a dataset name and the value is a list of column names.
+        The method creates a figure with subplots for each dataset and plots
+        the corresponding columns of the DataFrame.
+        The figure is then saved to a file in the 'plots' directory in png format
+        with the ticker symbol as the filename.
+        Args:
+        df (pd.DataFrame)
+            The DataFrame to plot
+        datasets (dict)
+            A dictionary of datasets, where each key is a dataset name and
+            the value is a list of column names to be plotted
+        """
+        numax = len(datasets)
+        fig, axes = plt.subplots(
+            nrows=numax,
+            ncols=1,
+            figsize=(10, 5*numax))
+        for i, ax in enumerate(axes.flat):
+            dataset = list(datasets.keys())[i]
+            colstoplot = datasets[dataset]
+            self.plot_stock_metrics_ax(
+                ax,
+                dataset,
+                df,
+                colstoplot)
+        plt.tight_layout()
+        plt.savefig(os.path.join('plots',  f'{self.ticker}.png'))
+
+    def plot_stock_metrics_ax(
+            self,
+            ax:Axes,
+            dataset:str,
+            df:pd.DataFrame,
+            colstoplot:list) -> None:
+        """
+        Plots specified stock metrics on the provided Axes object.
+        This function takes in an Axes object and plots the specified columns from
+        the DataFrame `df` on it. It formats the x-axis with major ticks set to 
+        every Monday and minor ticks set to every day. The plot includes a title, 
+        x and y labels, and optionally a legend if more than one column is plotted.
+        Additional shaded regions are added for certain datasets.
+        Args:
+            ax (matplotlib.axes.Axes): The flattened axes to plot on.
+            dataset (str): The name of the dataset, used for the title and y-label.
+            df (pd.DataFrame): The DataFrame containing the data to plot.
+            colstoplot (list): A list of column names to plot from the DataFrame.
+        Note:
+            - If `dataset` is 'Index' or 'Indices', the y-axis is limited to [0, 100]
+            and a shaded region between y=30 and y=70 is added.
+            - If `dataset` is 'Price' or 'Prices', a shaded region between 'Low' and 
+            'High' columns is added.
+        """
+
+        print(f'plotting {colstoplot} in {dataset}')
+        colorcycle = ['black', 'blue', 'red', 'green', 'orange']
+        for i, col in enumerate(colstoplot):
+            ax.plot(
+                df.index,
+                df[col],
+                color=colorcycle[i],
+                label=col,
+                linewidth=2)
+        # Format major ticks with year
+        # Set major ticks (every Monday with labels)
+        ax.xaxis.set_major_locator(mdates.WeekdayLocator(byweekday=mdates.MO))
+        ax.xaxis.set_major_formatter(mdates.DateFormatter('%b %d'))
+        # Set minor ticks (every day, but without labels)
+        ax.xaxis.set_minor_locator(mdates.DayLocator())
+        
+        ax.set_title(dataset)
+        ax.set_xlabel('Date')
+        ax.set_ylabel(dataset)
+        if len(colstoplot) > 1:
+            ax.legend()
+        if dataset in ['Index', 'Indices']:
+            ax.set_ylim([0, 100])
+            # Add a transparent shaded region between y=30 and y=70
+            ax.fill_between(df.index, 30, 70, color='gray', alpha=0.3)
+        if dataset in ['Price', 'Prices']:
+            # Add a transparent shaded region between y=30 and y=70
+            ax.fill_between(df.index, df['Low'], df['High'], color='gray', alpha=0.3)
+        ax.grid(True, linestyle='--', alpha=0.7)
+
+
+if __name__ == '__main__':
+    ticker = 'GOOG'
+    ta = TechnicalAnalysis(ticker, debug=False).run()
+
+