#+--------------------------------------------------------------------------------------------+ # Model for Stock Price Prediction via Dense only Neural Network # Using today vs tomorrow analysis # # Written by: Prakash R. Kota # Location: East Greenbush, NY # # Using just a Dense Neural Network Model # Adding lots of direct stock parmeters from Yahoo Finance # Open, High, Low, Close, Volume # # Will not be using other parameters such as # Return, SMA10, EMA10, RollingVol10, # SP500, Nasdaq, VIX # RSI, Day-of-Week # Removed all saving and graphing from PRK_1a_tf_Stock_NN.ipnyb # Keeping it minimal for just the stock prediction and output display table # Goal is to make a MVP - Minimal Viable Product # Infer the NN Model from the Saved Model - app.py required for Hugging Face Space # # PRK_1b_2_Model_Infer_tf_Stock_NN.ipnyb - Using Tensorflow 2.16.2 CPU # based on # PRK_1b_tf_Stock_NN.ipnyb # PRK_1a_tf_Stock_NN.ipnyb # PRK_11b_tf_Stock_DenseOnly.ipynb # Renamed PRK_10e_tf_Stock_DenseOnly.ipynb for convenience # This is the best Notebook Code for the NN Model # # Written on: 28 Mar 2025 # Last update: 02 Apr 2025 #+--------------------------------------------------------------------------------------------+ import os # MUST come before importing yfinance os.environ["YFINANCE_NO_CACHE"] = "1" # disable cache os.environ["XDG_CACHE_HOME"] = "/tmp/xfake_cache" if not os.path.exists("/tmp/xfake_cache"): os.makedirs("/tmp/xfake_cache", exist_ok=True) import yfinance as yf import gradio as gr import numpy as np import pandas as pd from datetime import datetime from pandas.tseries.offsets import BDay from tabulate import tabulate from tensorflow.keras.models import load_model from sklearn.preprocessing import MinMaxScaler import joblib import sklearn import tensorflow as tf import shutil from pytz import timezone from pandas.tseries.offsets import BDay import hashlib import yfinance as yf import time import gc import threading lock = threading.Lock() # This forces all inference requests to run one at a time, which avoids file # lock conflicts in SQLite. It’s simple but effective for debugging. # --- Load saved model and scalers --- # model_dir = "./model" NN_model = load_model(os.path.join(model_dir, "NN_CPU_model.keras")) NN_model.compile(optimizer="adam", loss="mse") # even if not used, ensures full init NN_model.predict(np.zeros((1, 5))) # warm-up dummy prediction scaler_X = joblib.load(os.path.join(model_dir, "scaler_X.pkl")) scaler_y = joblib.load(os.path.join(model_dir, "scaler_y.pkl")) def safe_download(*args, retries=3, delay=1, **kwargs): for i in range(retries): try: df = yf.download(*args, **kwargs) if not df.empty: return df except Exception as e: print(f"[Attempt {i+1}] yfinance error: {e}") time.sleep(delay) raise RuntimeError("yfinance download failed after retries.") # --- Inference Function --- # def predict_stock(): with lock: # lightweight log print near the top of predict_stock() to verify it's hitting the cache config: print("YFINANCE_NO_CACHE =", os.getenv("YFINANCE_NO_CACHE")) # Check for time zone now_est = datetime.now(timezone("US/Eastern")) print("Current Eastern Time:", now_est) print("Trying to fetch data up to:", now_est.strftime('%Y-%m-%d')) # --- Clear yfinance cache to get latest volume and price data --- # cache_path = os.path.expanduser("~/.cache/py-yfinance") if os.path.exists(cache_path): print("Clearing yfinance cache...") shutil.rmtree(cache_path) Stock = "NVDA" start_date = "2020-01-01" train_end_date = "2024-12-31" #today = datetime.today().strftime('%Y-%m-%d') # Use EST for consistently for today today = now_est.strftime('%Y-%m-%d') # Download the full dataset (might contain stale final row) # solves any error with empty dataframes try: full_data = safe_download( tickers=Stock, start=start_date, end=today, interval="1d", auto_adjust=False, actions=False, progress=False, threads=True #<-- for parallel downloads, use True ) if full_data.empty: print("yfinance returned empty data for:", today) return "Error: Stock data not available at this time. Please try again shortly.", pd.DataFrame() except Exception as e: print("yfinance error:", e) return "Error: Could not fetch stock data. Please try again later.", pd.DataFrame() features = ["Open", "High", "Low", "Close", "Volume"] X_scaled = scaler_X.transform(full_data[features]) y = full_data["Close"].values.reshape(-1, 1) y_scaled = scaler_y.transform(y) X_all = X_scaled[:-1] y_all = y_scaled[1:].flatten() dates_all = full_data.index[1:] test_mask = dates_all > pd.to_datetime(train_end_date) X_test, y_test = X_all[test_mask], y_all[test_mask] dates_test = dates_all[test_mask] # Predict next day prices y_pred_scaled = NN_model.predict(X_test).flatten() y_pred = scaler_y.inverse_transform(y_pred_scaled.reshape(-1, 1)).flatten() y_true = scaler_y.inverse_transform(y_test.reshape(-1, 1)).flatten() last_date = full_data.index[-1] last_close_price = float(full_data["Close"].iloc[-1].item()) X_input = X_scaled[-1].reshape(1, -1) next_day_pred_scaled = NN_model.predict(X_input) next_day_pred = scaler_y.inverse_transform(next_day_pred_scaled)[0][0] mape = np.mean(np.abs((y_true - y_pred) / y_true)) * 100 std_ape = np.std(np.abs((y_true - y_pred) / y_true)) * 100 mape_margin = next_day_pred * (mape / 100) sae_margin = next_day_pred * (std_ape / 100) next_date = (last_date + BDay(1)).date() summary_lines = [ f"Prediction for {Stock}:", f"Last available date: {last_date.date()}, Close = ${last_close_price:.2f}", f"Predicted closing price for next trading day ({next_date}): ${next_day_pred:.2f}", f"Expected range (±MAPE): ${next_day_pred - mape_margin:.2f} to ${next_day_pred + mape_margin:.2f}", f"Expected range (±SAE): ${next_day_pred - sae_margin:.2f} to ${next_day_pred + sae_margin:.2f}" ] summary = "\n".join(summary_lines) prediction_df = pd.DataFrame({ 'Date': dates_test, 'Actual Close': y_true, 'Predicted Close': y_pred }) prediction_df['% Error'] = ((prediction_df['Actual Close'] - prediction_df['Predicted Close']) / prediction_df['Actual Close']) * 100 prediction_df['% Error'] = prediction_df['% Error'].map(lambda x: f"{x:+.2f}%") prediction_df['±MAPE Range'] = prediction_df['Predicted Close'].apply( lambda x: f"${x * (1 - mape/100):.2f} to ${x * (1 + mape/100):.2f}" ) prediction_df['Date'] = prediction_df['Date'].dt.strftime("%Y-%m-%d") prediction_df['Actual Close'] = prediction_df['Actual Close'].map(lambda x: f"${x:.2f}") prediction_df['Predicted Close'] = prediction_df['Predicted Close'].map(lambda x: f"${x:.2f}") prediction_df = prediction_df.sort_values("Date", ascending=False) headers = ["Prediction For Date", "Actual Close", "Predicted Close", "% Error", "±MAPE Range"] table = tabulate(prediction_df.values, headers=headers, tablefmt="plain") # Start Sanity Checks assert not np.any(np.isnan(X_scaled[-1].reshape(1, -1))), "NaNs detected in input!" assert X_scaled[-1].reshape(1, -1).shape == (1, X_scaled.shape[1]), f"Unexpected shape: {X_scaled[-1].reshape(1, -1).shape}" #print("X_input shape:", X_scaled[-1].reshape(1, -1).shape) #print("X_input contains NaNs:", np.any(np.isnan(X_scaled[-1].reshape(1, -1)))) #print("Latest data row used for prediction:") #print(full_data[features].iloc[-1]) #print("scikit-learn version:", sklearn.__version__) #print("tensorflow version:", tf.__version__) #print("scaler_X min/max:", scaler_X.data_min_, scaler_X.data_max_) #print("scaler_y min/max:", scaler_y.data_min_, scaler_y.data_max_) # Run a prediction on a known fixed input #x_debug = np.array([[0.5, 0.5, 0.5, 0.5, 0.5]]) #y_debug = NN_model.predict(x_debug) #y_debug_unscaled = scaler_y.inverse_transform(y_debug) #print("Debug prediction (scaled):", y_debug) #print("Debug prediction (unscaled):", y_debug_unscaled) gc.collect() #<-- garbage collection after predict import hashlib def md5(fname): with open(fname, "rb") as f: return hashlib.md5(f.read()).hexdigest() #print("Model MD5 checksum:", md5(os.path.join(model_dir, "NN_CPU_model.keras"))) #print(full_data.tail(3)) # End Sanitiy Checks return summary, prediction_df[["Date", "Actual Close", "Predicted Close", "% Error", "±MAPE Range"]] # --- Gradio Interface --- # """ demo = gr.Interface( fn=predict_stock, inputs=[], outputs=[ gr.Textbox(label="Prediction Summary", lines=6), gr.Dataframe(headers=["Prediction For Date", "Actual Close", "Predicted Close", "% Error", "±MAPE Range"], label="Prediction Table (2025+)") ], title="📈 NVDA Stock Predictor", description="This app uses a Dense Neural Network to predict NVDA's next trading day's closing price.", live=True #<-- changed to True for live queuing ) demo.launch() """ with gr.Blocks() as demo: # Title gr.Markdown("## 📊 NVDA Stock Prediction Using Neural Networks") # Outputs output_summary = gr.Textbox(label="Prediction Summary", lines=6) output_table = gr.Dataframe(headers=["Prediction For Date", "Actual Close", "Predicted Close", "% Error", "±MAPE Range"], label="Prediction Table (2025+)") # Generate Button generate_btn = gr.Button("Generate") generate_btn.click(fn=predict_stock, inputs=[], outputs=[output_summary, output_table]) # External article link gr.HTML("""

👉 Read the full article here - AI/ML in Finance: How a Lightweight Neural Network Forecasts NVDA’s Next Stock Price Move

""") # Disclaimer paragraph gr.HTML("""

Disclaimer: The information provided in this article and through the linked prediction model is for educational and informational purposes only. It does not constitute financial, investment, or trading advice, and should not be relied upon as such.

Any decisions made based on the model's output are solely at the user’s own risk. I make no guarantees regarding the accuracy, completeness, or reliability of the predictions. I am not responsible for any financial losses or gains resulting from the use of this model.

Always consult with a licensed financial advisor before making any investment decisions.

""") demo.launch(share=True)