technical_analysis: merge and plot forecasts

src/technical_analysis.py

@@ -1,6 +1,7 @@
 import datetime as dt
 import logging
 
+import dotenv
 import matplotlib.pyplot as plt
 import matplotlib.dates as mdates
 from matplotlib.axes import Axes
@@ -11,12 +12,15 @@ from ta.momentum import RSIIndicator, StochasticOscillator
 from ta.trend import MACD 
 import yfinance as yf
 
+from fetch_forecast import FetchForecast
 
 class TechnicalAnalysis():
     def __init__(
             self,
             ticker:str,
             fetchperiodinweeks:int=12,
+            df_past=None,
+            df_fcst=None,
             plot_ta:bool=True,
             savefig:bool=False,
             debug=False):
@@ -28,6 +32,10 @@ class TechnicalAnalysis():
             stock ticker to analyze
         fetchperiodinweeks : int, optional, default: 8
             number of weeks to fetch historical data from YahooFinance
+        df_past: pd.DataFrame, optional, default: None
+            Closeing price of the ticker for the past few days
+        df_fcst: pd.DataFrame, optional, default: None
+            Forecasted closing price and relative returns nextf few days
         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
@@ -44,6 +52,8 @@ class TechnicalAnalysis():
         # input arguments
         self.ticker = ticker
         self.fetchperiodinweeks = fetchperiodinweeks
+        self.df_past = df_past
+        self.df_fcst = df_fcst
         self.plot_ta = plot_ta
         self.savefig = savefig
         # done initializing
@@ -60,20 +70,23 @@ class TechnicalAnalysis():
         - plots the price and TA metrics.
         """
         # fetch data from yf
-        self.df = self.fetch_data()
+        df = self.fetch_data()
         # add the features based on technical analysis
-        if self.df.shape[0] > 0:
-            self.df = self.tech_analysis()
+        if df.shape[0] > 0:
+            df = self.tech_analysis(df)
+            # Merge with forecast data
+            df_merged = self.merge_hist_with_forecast(df, self.df_past, self.df_fcst)
             # plot the results
             if self.plot_ta:
                 os.makedirs('plots', exist_ok=True)
                 fig = self.plot_stock_metrics(
-                    self.df,
+                    df_merged,
                     datasets={
-                            'Volume': ['Volume'],
-                            'Prices': ['Close', 'VWAP'],  # 'High','Low', 
+                            'Volume': ['Volume'], 
                             'Indices': ['RSI', 'StochOsc'],
-                            'Trend': ['MACD', 'MACDsig', 'MACDdif']},
+                            'Trend': ['MACD', 'MACDsig', 'MACDdif'],
+                            'Prices': ['Close', 'VWAP']  # 'High','Low',
+                            },
                     savefig=self.savefig
                     )
             else:
@@ -84,7 +97,7 @@ class TechnicalAnalysis():
             else:
                 fig = None
 
-        return self.df, fig
+        return df_merged, fig
     
     def fetch_data(
             self
@@ -130,7 +143,8 @@ class TechnicalAnalysis():
         return df
     
     def tech_analysis(
-            self
+            self,
+            df: pd.DataFrame
             ) -> pd.DataFrame:
         """
         Calculates technical analysis indicators for the fetched stock price data.
@@ -144,11 +158,13 @@ class TechnicalAnalysis():
         - Trend Indicators: 
             - Moving Average Convergence Divergence (MACD)
         The calculated indicators are added to the DataFrame as new columns.
+        Args:
+        df: pd.DataFrame
+            The DataFrame containing the fetched stock price data.
         Returns:
         pd.DataFrame
             The DataFrame with the calculated technical analysis indicators.
         """
-        df = self.df
 
         # Price Indicators
         # Volume-Weighted Average Price (VWAP)
@@ -188,13 +204,41 @@ class TechnicalAnalysis():
 
         return df
 
+    def merge_hist_with_forecast(self, df_hist: pd.DataFrame, df_past: pd.DataFrame | None, df_fcst: pd.DataFrame | None) -> pd.DataFrame:
+        # make sure we are merging the right thing
+        """
+        Merge historical data with forecast data. If forecast data is available, merge it with historical data based on date.
+        If forecast data is not available, return the historical data as is.
+        Args:
+        df_hist: pd.DataFrame
+            Historical data
+        df_past: pd.DataFrame
+            Recent data used for comparison
+        df_fcst: pd.DataFrame or None
+            Forecast data
+        Returns:
+        df_merged: pd.DataFrame
+            Merged data
+        """
+        if df_fcst is not None:
+            # Make sure that the previous hist close price is matching to that of the past close price
+            assert df_hist.Close.iloc[-2] == df_past.Close.iloc[-2]
+            df_hist.reset_index(inplace=True)
+            # in case there are overlapping dates, make sure to remove them 
+            df_fcst = df_fcst.loc[~df_fcst["Date"].isin(df_hist["Date"]), ["Date", "Close"]]
+            date_diff = df_fcst.Date.iloc[0] - df_hist.Date.iloc[-1]
+            if date_diff > pd.Timedelta('3 days'):
+                self.logger.warning(f'Date diff between the first forecast and the last hist is {date_diff}')
+            df_merged = pd.concat([df_hist, df_fcst], ignore_index=True)
+            df_merged.set_index("Date", inplace=True)
+        else:
+            df_merged = df
+        return df_merged
+
     def plot_stock_metrics(
             self,
             df,
-            datasets={
-                'Volume': ['Volume'],
-                'Price': ['Close']  # 'High','Low'
-                },
+            datasets,
             savefig=False
             ) -> None:
         """
@@ -262,7 +306,7 @@ class TechnicalAnalysis():
             'High' columns is added.
         """
 
-        print(f'plotting {colstoplot} in {dataset}')
+        self.logger.info(f'plotting {colstoplot} in {dataset}')
         colorcycle = ['black', 'blue', 'red', 'green', 'orange']
         for i, col in enumerate(colstoplot):
             ax.plot(
@@ -271,6 +315,29 @@ class TechnicalAnalysis():
                 color=colorcycle[i],
                 label=col,
                 linewidth=2)
+        
+        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, label='30-70 Range')
+        elif dataset in ['Price', 'Prices']:
+            # Add a transparent shaded region daily lows and highs
+            ax.fill_between(df.index, df['Low'], df['High'], color='gray', alpha=0.3, label='Price Range')
+            # extract the Price rows for which 'High's are NaN
+            nanind = df.High.isna()
+            df_fcst_close = df.loc[nanind, 'Close']
+            if df_fcst_close.shape[0] > 0:
+                ax.plot(
+                    df_fcst_close.index,
+                    df_fcst_close,
+                    color='red',
+                    marker='*',
+                    label='Forecast')
+        # else:
+        #     # plot a transparent line across the full df.index just to make sure that x-axis limits are identical for all panels
+        #     ax.plot(df.index, df[col], color='gray', alpha=0.0)
+        #     # ax.fill_between(df.index, 30, 70, color='gray', alpha=0.0)
+        ax.set_xlim([df.index.min()-pd.Timedelta(days=5), df.index.max()+pd.Timedelta(days=5)])
         # Format major ticks with year
         # Set major ticks (every Monday with labels)
         ax.xaxis.set_major_locator(mdates.WeekdayLocator(byweekday=mdates.MO))
@@ -278,22 +345,15 @@ class TechnicalAnalysis():
         ax.xaxis.set_major_formatter(mdates.DateFormatter('%m.%d'))
         # Set minor ticks (every day, but without labels)
         ax.xaxis.set_minor_locator(mdates.DayLocator())
-
         plt.setp(ax.get_xticklabels(), rotation=90, ha='center')
-        
+    
+        ax.set_ylabel(dataset)
+
+        ax.grid(True, linestyle='--', alpha=0.7)
         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)
+            ax.legend(loc='upper left')
 
     def get_fetcherror_fig(
             self,
@@ -326,8 +386,11 @@ class TechnicalAnalysis():
         return fig
 
 if __name__ == '__main__':
-    ticker = 'GOOG'
-    df, fig = TechnicalAnalysis(ticker, plot_ta=True, savefig=True, debug=False).run()
-    print(f'columns: {df.columns}')
+    ticker = 'AAPL'
+    # fetch the forecasts
+    dotenv.load_dotenv(dotenv.find_dotenv())
+    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()
+    # print(f'columns: {df.columns}')