functions.py

from asyncio.constants import LOG_THRESHOLD_FOR_CONNLOST_WRITES
import yfinance as yf
import pandas as pd
import numpy as np
import plotly.graph_objs as go
from stocks import *
from transformers import AutoModelForSequenceClassification, pipeline, AutoTokenizer
import os
from random import random
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '2'
import tensorflow as tf
import math
import datetime
import random 
import time
#import kaleido 
from sklearn.preprocessing import MinMaxScaler 
import matplotlib.pyplot as plt
#import warnings
import tensorflow as tf
from tensorflow import keras
from keras.layers import Dropout, Activation 
from keras import layers
from keras.callbacks import EarlyStopping
from sklearn.metrics import r2_score
import plotly.graph_objs as go  
import plotly.io as pio
pio.templates

model = AutoModelForSequenceClassification.from_pretrained("fine_tuned_FinBERT", from_tf=False, config="config.json") 
tokenizer = AutoTokenizer.from_pretrained("fine_tuned_FinBERT/tokenizer/")

class Models(object):
    def __init__(self):
        self.stock_data = Stock_Data()

    def bollinger_bands_20d_2std(self, ticker):
        '''
        This method calculates the Bollinger Bands with a Rolling average of the last 20 days and 2 standard deviations. In a plot, 
        this would be represented as 3 lines: a rolling average, an upper bound (rolling average + 2 standard deviations) and a lower
        bound (rolling average - 2 standard deviations). When the price of a stock is between the rolling average and lower bound, it is
        considered as oversold, so it makes sense to buy, if it is between the roll. avg. and the upper bound, it is considered as 
        overbought, so it makes sense to sell, if it is equal to the roll.avg. it is neutral and if it is outside the bounds, it is 
        considered an Unusual Event. The function returns the outlook of the stock (either "Buy", or "Sell" or "Hold" or "Unusual Event")
        '''
        if self.stock_data.status_getter(ticker) != "Open":
            return "Market Closed"
        else:
            data = self.stock_data.stock_data_getter(ticker)
            low_high_closing_df = pd.DataFrame(data)
            low_high_closing_df = data.iloc[:, 4:5]  # Getting only the "Adj Close" column
            low_high_closing_df = low_high_closing_df.tail(40) # Getting the last 40 days

            low_high_closing_df["rolling_avg_20d"] = low_high_closing_df['Adj Close'].rolling(20, min_periods = 20).mean()
            low_high_closing_df["sd"] = low_high_closing_df["Adj Close"].rolling(20, min_periods = 20).std()
            low_high_closing_df = low_high_closing_df.tail(20) # Keeping the last 20 days only

            recent_data = low_high_closing_df.iloc[-1, :].to_list() # Creating a Series object with the most recent data (last row only)

            upper_bound = recent_data[1] + 2*recent_data[2] # Upper Bound
            lower_bound = recent_data[1] - 2*recent_data[2] # Lower Bound
            mean_20d = recent_data[1] # Rolling average of last 20 days
            
            if self.stock_data.current_price_getter(ticker) is None:
               return "Market Closed"
            else:
                message = ""

                if self.stock_data.current_price_getter(ticker) < mean_20d and self.stock_data.current_price_getter(ticker) >= lower_bound:
                    message = "Buy"
                elif self.stock_data.current_price_getter(ticker) > mean_20d and self.stock_data.current_price_getter(ticker) <= upper_bound:
                    message = "Sell"
                elif self.stock_data.current_price_getter(ticker) == mean_20d:
                    message = "Hold"
                elif self.stock_data.current_price_getter(ticker) <= lower_bound or self.stock_data.current_price_getter(ticker) >= upper_bound:
                    message = "Unusual Event"
                return message

    def bollinger_bands_10d_1point5std(self, ticker):
        '''
        This method calculates the Bollinger Bands with a Rolling average of the last 10 days and 1.5 standard deviations. In a plot, 
        this would be represented as 3 lines: a rolling average, an upper bound (rolling average + 1.5 standard deviations) and a lower
        bound (rolling average - 1.5 standard deviations). When the price of a stock is between the rolling average and lower bound, it is
        considered as oversold, so it makes sense to buy, if it is between the roll. avg. and the upper bound, it is considered as 
        overbought, so it makes sense to sell, if it is equal to the roll.avg. it is neutral and if it is outside the bounds, it is 
        considered an Unusual Event. The function returns the outlook of the stock (either "Buy", or "Sell" or "Hold" or "Unusual Event")
        '''
        if self.stock_data.status_getter(ticker) != "Open":
                return "Market Closed"
        else:
            data = self.stock_data.stock_data_getter(ticker)

            low_high_closing_df = pd.DataFrame(data)
            low_high_closing_df = data.iloc[:, 4:5]  # Getting only the "Adj Close" column
            low_high_closing_df = low_high_closing_df.tail(20) # Getting the last 20 days

            low_high_closing_df["rolling_avg_10d"] = low_high_closing_df['Adj Close'].rolling(10, min_periods = 10).mean()
            low_high_closing_df["sd"] = low_high_closing_df["Adj Close"].rolling(10, min_periods = 10).std()
            low_high_closing_df = low_high_closing_df.tail(10) # Keeping the last 10 days only

            recent_data = low_high_closing_df.iloc[-1, :].to_list() # Creating a Series object with the most recent data (last row only)

            upper_bound = recent_data[1] + 1.5*recent_data[2] # Upper Bound
            lower_bound = recent_data[1] - 1.5*recent_data[2] # Lower Bound
            mean_10d = recent_data[1] # Rolling average of last 10 days

            if self.stock_data.current_price_getter(ticker) is None:
               return "Market Closed"
            else:
                message = ""

                if self.stock_data.current_price_getter(ticker) < mean_10d and self.stock_data.current_price_getter(ticker) >= lower_bound:
                    message = "Buy"
                elif self.stock_data.current_price_getter(ticker) > mean_10d and self.stock_data.current_price_getter(ticker) <= upper_bound:
                    message = "Sell"
                elif self.stock_data.current_price_getter(ticker) == mean_10d:
                    message = "Hold"
                elif self.stock_data.current_price_getter(ticker) <= lower_bound or self.stock_data.current_price_getter(ticker) >= upper_bound:
                    message = "Unusual Event"
                return message

    def bollinger_bands_50d_3std(self, ticker):
        '''
        This method calculates the Bollinger Bands with a Rolling average of the last 50 days and 3 standard deviations. In a plot, 
        this would be represented as 3 lines: a rolling average, an upper bound (rolling average + 3 standard deviations) and a lower
        bound (rolling average - 3 standard deviations). When the price of a stock is between the rolling average and lower bound, it is
        considered as oversold, so it makes sense to buy, if it is between the roll. avg. and the upper bound, it is considered as 
        overbought, so it makes sense to sell, if it is equal to the roll.avg. it is neutral and if it is outside the bounds, it is 
        considered an Unusual Event. The function returns the outlook of the stock (either "Buy", or "Sell" or "Hold" or "Unusual Event")
        '''
        if self.stock_data.status_getter(ticker) != "Open":
                return "Market Closed"
        else:
            data = self.stock_data.stock_data_getter(ticker)

            low_high_closing_df = pd.DataFrame(data)
            low_high_closing_df = data.iloc[:, 4:5]  # Getting only the "Adj Close" column
            low_high_closing_df = low_high_closing_df.tail(100) # Getting the last 100 days

            low_high_closing_df["rolling_avg_50d"] = low_high_closing_df['Adj Close'].rolling(50, min_periods = 50).mean()
            low_high_closing_df["sd"] = low_high_closing_df["Adj Close"].rolling(50, min_periods = 50).std()
            low_high_closing_df = low_high_closing_df.tail(50) # Keeping the last 50 days only

            recent_data = low_high_closing_df.iloc[-1, :].to_list() # Creating a Series object with the most recent data (last row only)

            upper_bound = recent_data[1] + 3*recent_data[2] # Upper Bound
            lower_bound = recent_data[1] - 3*recent_data[2] # Lower Bound
            mean_50d = recent_data[1] # Rolling average of last 50 days

            # Finding the outlook dependent on the current price
            if self.stock_data.current_price_getter(ticker) is None:
               return "Market Closed"
            else:
                message = ""
                if self.stock_data.current_price_getter(ticker) < mean_50d and self.stock_data.current_price_getter(ticker) >= lower_bound:
                    message = "Buy"
                elif self.stock_data.current_price_getter(ticker) > mean_50d and self.stock_data.current_price_getter(ticker) <= upper_bound:
                    message = "Sell"
                elif self.stock_data.current_price_getter(ticker) == mean_50d:
                    message = "Hold"
                elif self.stock_data.current_price_getter(ticker) <= lower_bound or self.stock_data.current_price_getter(ticker) >= upper_bound:
                    message = "Unusual Event"
                return message

    def MACD(self, ticker):
        '''
        This method calculates the MACD (Mean Average Convergence Divergence) for a stock. The decision of whether to buy or sell
        a stock when using this method, depends on the difference of two "lines". The 1st one is called "MACD" and is equal to the 
        difference between the Exponential Moving Average of the adjusted closing price of the last 12 days, and the Moving Average
        of the adjusted closing price of the last 26 days. The 2nd line is the 9 day moving average of the adj. closing price. 
        When MACD > 9 day M.A. it is considered that there is an uptrend, else, an downtrend. 
        At last, when MACD line crosses the 9 day M.A. from "above", a "Sell" signal is given, 
        while it crosses it from below, a "Buy" signal is given.
        '''
        if self.stock_data.status_getter(ticker) != "Open":
            return "Market Closed"
        else:
            data = self.stock_data.stock_data_getter(ticker)

            low_high_closing_df = pd.DataFrame(data)
            low_high_closing_df = data.iloc[:, 4:5]  # Getting only the "Adj Close" column
            low_high_closing_df = low_high_closing_df.tail(52) # Getting the last 52 days


            # Get the 12-day EMA of the closing price
            low_high_closing_df['EMA_12d'] = low_high_closing_df['Adj Close'].ewm(span=12, adjust=False, min_periods=12).mean()
            # Get the 26-day MA of the closing price
            low_high_closing_df['MA_26d'] = low_high_closing_df['Adj Close'].ewm(span=26, adjust=False, min_periods=26).mean()
            # Subtract the 26-day EMA from the 12-Day EMA to get the MACD
            low_high_closing_df['MACD'] = low_high_closing_df['EMA_12d'] - low_high_closing_df['MA_26d']
            # Making the signal line
            low_high_closing_df['MA_9d'] = low_high_closing_df['MACD'].ewm(span=9, adjust=False, min_periods=9).mean()

            low_high_closing_df['Diff'] = low_high_closing_df['MACD'] - low_high_closing_df['MA_9d'] 

            Diff = low_high_closing_df['Diff'].astype(float)
            
            if self.stock_data.current_price_getter(ticker) is None:
               return "Market Closed"
            else:
                message = ""

                if Diff.iloc[-1] < 0:
                    if Diff.iloc[-2] >= 0:
                        message = "Downtrend and sell signal"
                    else:
                        message = "Downtrend and no signal"
                else:
                    if Diff.iloc[-2] <= 0:
                        message = "Uptrend and buy signal"
                    else:
                        message = "Uptrend and no signal"
                return message

    def finbert_headlines_sentiment(self, ticker):
        '''
        This method uses a the "weights" and the "tokenizer" of a fine-tuned Fin-BERT model, which is a BERT model that 
        was furtherly trained on financial data. The "article_parser()" method scraps www.marketwatch.com and returns the
        last 17 headers of the chosen stock's articles. The, the FinBERT model classifies each one of them as either "Positive"
        or "Negative" or "Neutral", and a score is assigned to each header (+100, -100, and 0) correspondingly. At last, a
        rolling average of window size = 5 is used to "smooth" the sentiment line of the "plotly" plot that is returned.
        '''

        articles_df = self.stock_data.article_parser(ticker)
        articles_list = articles_df["headline"].tolist()
        
        clf = pipeline("text-classification", model=model, tokenizer=tokenizer) 
        outputs_list = clf(articles_list)
        
        sentiments = []

        for item in outputs_list:
            sentiments.append(item["label"])
        
        sentiments_df = pd.DataFrame(sentiments)
        sentiments_df.rename(columns = {0:'sentiment'}, inplace = True)

        sentiments_df["sentiment"] = sentiments_df["sentiment"].apply(lambda x: 100 if x == "positive" else -100 if x=="negative" else 0)            
        sentiments_df["roll_avg"] = round(sentiments_df["sentiment"].rolling(5, min_periods = 1).mean(), 2)
        sentiments_df = sentiments_df.tail(12).reset_index()

        pd.options.plotting.backend = "plotly"

        fig = sentiments_df["roll_avg"].plot(title="Sentiment Analysis of the last 12 www.marketwatch.com articles about " + ticker, 
        
        template="plotly_dark",
        labels=dict(index="12 most recent article headlines", value="sentiment  score (rolling avg. of window size 5)"))
        fig.update_traces(line=dict(color="#3D9140", width=3))
        fig.update_layout(yaxis_range=[-100,100])
        fig.update_layout(xaxis_range=[0,12])
        fig.update_layout(showlegend=False)
        fig.add_hline(y=0, line_width=1.5, line_color="black")
       
        current_sentiment = sentiments_df["roll_avg"].tail(1).values[0]

        return {'fig': fig, 'current_sentiment': current_sentiment}

    def LSTM_7_days_price_predictor(self, ticker):
        '''
        This method predicts the price of a chosen stock for the next 7 days as of today, by using the daily adjusted closing 
        prices for the last 2 years. At first, a 60-day window of historical prices (i-60) is created as our feature data (x_train)
        and the following 60-days window as label data (y_train). For every stock available, we have manually defined different 
        parameters so that they fit as good as it gets to the model. Finally we combute the R2 metric and make the predictions. At 
        last, we proceed with the predictions. The model looks back in our data (60 days back) and predicta for the following 7 days.
        '''

        stock_data = self.stock_data.LSTM_stock_data_getter(ticker)
        stock_data=pd.DataFrame(data=stock_data).drop(['Open','High','Low','Close', 'Volume'],axis=1).reset_index()
        stock_data['Date'] = pd.to_datetime(stock_data['Date'])
        stock_data=stock_data.dropna()

        # Data Preprocessing
        random.seed(1997)
        close_prices = stock_data['Adj Close']
        values = close_prices.values
        training_data_len = math.ceil(len(values)* 0.8)

        scaler = MinMaxScaler(feature_range=(0,1))
        scaled_data = scaler.fit_transform(values.reshape(-1,1))
        train_data = scaled_data[0: training_data_len, :]

        x_train = []
        y_train = []

        for i in range(60, len(train_data)):
            x_train.append(train_data[i-60:i, 0])
            y_train.append(train_data[i, 0])

        x_train, y_train = np.array(x_train), np.array(y_train)
        x_train = np.reshape(x_train, (x_train.shape[0], x_train.shape[1], 1))

        # Preparation of test set
        test_data = scaled_data[training_data_len-60: , : ]
        x_test = []
        y_test = values[training_data_len:]

        for i in range(60, len(test_data)):
            x_test.append(test_data[i-60:i, 0])

        x_test = np.array(x_test)
        x_test = np.reshape(x_test, (x_test.shape[0], x_test.shape[1], 1))

        ##### Setting Up LSTM Network Architecture and the Training of the LSTM Model
        def LSTM_trainer(seed, DROPOUT, LSTM_units,patience,batch_size, epochs):

            tf.random.set_seed(seed)
            DROPOUT = DROPOUT
            global model_lstm
            model_lstm = keras.Sequential()
            model_lstm.add(layers.LSTM(LSTM_units, return_sequences=True, input_shape=(x_train.shape[1], 1)))
            model_lstm.add(Dropout(rate=DROPOUT))
            model_lstm.add(layers.LSTM(LSTM_units, return_sequences=False))
            model_lstm.add(Dropout(rate=DROPOUT))
            model_lstm.add(layers.Dense(25))
            model_lstm.add(Dropout(rate=DROPOUT))
            model_lstm.add(layers.Dense(1))
            model_lstm.add(Activation('linear'))
        
            print('\n')
            print("Compiling the LSTM Model for the "  + str(ticker) + " stock....\n")
            t0 = time.time()
            model_lstm.compile(optimizer='adam', loss='mean_squared_error',metrics=['mae'])
            callback=EarlyStopping(monitor='val_loss',
                              min_delta=0,
                              patience=patience,
                              verbose=1, mode='auto')
            model_lstm.fit(x_train, 
                          y_train,
                          batch_size= batch_size, 
                          epochs=epochs,
                          validation_split=0.1,# ...holding out 10% of the data for validation 
                          shuffle=True,verbose=0,callbacks=[callback])
            t1 = time.time()
            global ex_time
            ex_time = round(t1-t0, 2)
            print("Compiling took :",ex_time,"seconds")

            predictions = model_lstm.predict(x_test)
            predictions = scaler.inverse_transform(predictions)
            #rmse = np.sqrt(np.mean(((predictions - y_test) ** 2)))
            global r_squared_score 
            global rmse
            r_squared_score = round(r2_score(y_test, predictions),2)
            rmse = np.sqrt(np.mean(((predictions - y_test) ** 2)))
            #print('Rmse Score: ', round(rmse),2)
            print('R2 Score: ', r_squared_score)

        if ticker == 'AAPL':
            LSTM_trainer(1, 0.2, 100,2, 20, 30)
        elif ticker == 'NVDA':
            LSTM_trainer(2, 0.2, 100,2, 30, 50)    
        elif ticker == 'PYPL':
            LSTM_trainer(6, 0.2, 100,10,25, 30)
        elif ticker == 'MSFT':
            LSTM_trainer(4, 0.1, 80, 2,20, 40)
        elif ticker == 'TSLA':
            LSTM_trainer(5, 0.1, 120, 4,20, 25)
        elif ticker == 'AMZN':
            LSTM_trainer(6, 0.1, 120,2, 20, 25)    
        elif ticker == 'SPOT':
            LSTM_trainer(9, 0.2, 200,5, 20, 40)
        #elif ticker == 'TWTR' :
         #   LSTM_trainer(15, 0.2, 100,4,20, 40)
        elif ticker == 'UBER':
            LSTM_trainer(15, 0.2, 100,7,20, 40)
        elif ticker == 'adanipower.ns':
            LSTM_trainer(15, 0.2, 120,8,20, 40)
        elif ticker == 'GOOG':
            LSTM_trainer(15, 0.2, 100,3,20, 25)

        # Unseen Predictions for the next 7 days 
        close_data = scaled_data
        look_back = 60

        def predict(num_prediction, model):
            prediction_list = close_data[-look_back:]

            for _ in range(num_prediction):
                x = prediction_list[-look_back:]
                x = x.reshape((1, look_back, 1))
                
                out = model.predict(x)[0][0]
                prediction_list = np.append(prediction_list, out)
            prediction_list = prediction_list[look_back-1:]

            return prediction_list
    
        def predict_dates(num_prediction):
            last_date = stock_data['Date'].values[-1]
            prediction_dates = pd.date_range(last_date, periods=num_prediction+1).tolist()
            return prediction_dates

        num_prediction = 7
        
        forecast = predict(num_prediction, model_lstm)
        forecast_dates = predict_dates(num_prediction)
    
        plt.figure(figsize=(25,10))
        forecast = forecast.reshape(-1, 1)
        forecast_inverse = scaler.inverse_transform(forecast)

        # Ploting the Actual Prices and the Predictions of them for the next 7 days
        base = stock_data['Date'].iloc[[-1]] # Here we create our base date (the last existing date with actual prices)
        testdata = pd.DataFrame(forecast_inverse)# Here we create a data frame that contains the prediction prices and an empty column for their dates
        testdata['Date'] = ""
        testdata.columns = ["Adj Close","Date"]
        testdata = testdata.iloc[1:,:]
        testdata["Label"] = "" # Let's add a column "Label" that would show if the respective price is a prediction or not
        testdata["Label"] = "Prediction"
        testdata = testdata[["Date", "Adj Close", "Label"]]

        date_list = [base + datetime.timedelta(days=x+1) for x in range(testdata.shape[0]+1)] 
        date_list = pd.DataFrame(date_list)
        date_list.columns = ["Date"]
        date_list.reset_index(inplace = True)
        date_list.drop(["index"], axis = 1, inplace = True)
        date_list.index = date_list.index + 1
        testdata.Date = date_list

        stock_data["Label"] = ""
        stock_data["Label"] = "Actual price"
        finaldf = pd.concat([stock_data,testdata], axis=0) # Here we concatenate the "testdata" and the original data frame "df" into a final one
        finaldf.reset_index(inplace = True)
        finaldf.drop(["index"], axis = 1, inplace = True)
        finaldf['Date'] = pd.to_datetime(finaldf['Date'])
    
        plt.rcParams["figure.figsize"] = (25,10)
        #We create two different data frames, one that contains the actual prices and one that has only the predictions
        finaldfPredictions = finaldf.iloc[-8:] 
        finaldfActuals = finaldf.iloc[:-7]

        plot_1 = go.Scatter(
            x = finaldfActuals['Date'],
            y = finaldfActuals['Adj Close'],
            mode = 'lines',
            name = 'Historical Data (2 years)',
            line=dict(width=1,color='#3D9140'))
        plot_2 = go.Scatter(
            x = finaldfPredictions['Date'],
            y = finaldfPredictions['Adj Close'],
            mode = 'lines',
            name = '7-day Prediction',
            line=dict(width=1,color="#EE3B3B"))
        plot_3 = go.Scatter(
            x = finaldfPredictions['Date'][:1],
            y = finaldfPredictions['Adj Close'][:1],
            mode = 'markers',
            name = 'Latest Actual Closing Price',
            line=dict(width=1))

        layout = go.Layout(
            title = 'Next 7 days stock price prediction of ' + str(ticker),
            xaxis = {'title' : "Date"},
            yaxis = {'title' : "Price ($)"}
        )
        fig = go.Figure(data=[plot_1, plot_2,plot_3], layout=layout)
        fig.update_layout(template='plotly_dark',autosize=True)
        fig.update_layout(legend=dict(
            orientation="h",
            yanchor="bottom",
            y=1.02,
            xanchor="right",
            x=1),
            annotations = [dict(x=0.5,
                                y=0, 
                                xref='paper',
                                yref='paper',
                                text="Current In Sample R- Squared : " + str(r_squared_score*100) + " % \n",
                                showarrow = False)],
            xaxis=dict(showgrid=False),
            yaxis=dict(showgrid=False)
            

                        )
        fig.add_annotation(x=0.5, 
                           y=0.05,
                           xref='paper',
                           yref='paper',
                           text="Current In Sample Root Mean Square Error : " + str(round(rmse,2)) + " % ",
                           showarrow=False)
        
        return fig