predictions.py

from sklearn.preprocessing import MinMaxScaler
import numpy as np
import pandas as pd
from datetime import datetime, timedelta
import tensorflow as tf
from tensorflow.keras.models import load_model as tf_load_model

# Define the scaler globally
scaler = MinMaxScaler(feature_range=(0, 1))

def feature_engineering(stock_data):
    
    # Define the end date for fetching data (yesterday's date)
    end_date = datetime.now() - timedelta(days=1)

    # Fetch data until yesterday's date
    dataset = stock_data["Close"].loc[:end_date]
    dataset = pd.DataFrame(dataset)
    data = dataset.values

    # 75% to Train, 25% to Test
    train_size = int(len(data) * 0.85)
    test_size = len(data) - train_size

    train_data = data[:train_size]
    test_data = data[train_size-5:]

    scaled_train = scaler.fit_transform(np.array(train_data).reshape(-1, 1))
    scaled_test = scaler.transform(np.array(test_data).reshape(-1, 1))

    # Create dataset with 60 time steps and (59 input and only 1 output in each) as this is a regression problem
    X_train = []
    y_train = []

    for i in range(60, len(scaled_train)):
        X_train.append(scaled_train[i-5:i, 0])
        y_train.append(scaled_train[i, 0])

    # Convert Xs, y to arrays
    X_train, y_train = np.array(X_train), np.array(y_train)

    # Reshape data
    X_train = np.reshape(X_train, (X_train.shape[0], X_train.shape[1], 1))

    # Creating a testing set with 60 time-steps and 1 output
    X_test = []
    y_test = []

    for i in range(5, len(scaled_test)):
        X_test.append(scaled_test[i-5:i, 0])
        y_test.append(scaled_test[i, 0])

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

    return X_test

# Function to load a specific model
def load_custom_model(selected_model):
    if selected_model == 'BCA':
        model = tf_load_model('model_bca.keras')
    elif selected_model == 'BNI':
        model = tf_load_model('model_bni.keras')
    elif selected_model == 'BRI':
        model = tf_load_model('model_bri.keras')
    elif selected_model == 'Mandiri':
        model = tf_load_model('model_mri.keras')
    else:
        model = None
    return model

def run_inference(model, X_test):
    # Perform inference using the loaded model
    predicted_prices = model.predict(X_test)

    # Inverse scaling to get the actual prices
    predicted_prices = scaler.inverse_transform(predicted_prices)

    return predicted_prices