# train_nvidia_ml_model.py import warnings from itertools import product from pathlib import Path import joblib import matplotlib.pyplot as plt import numpy as np import pandas as pd from prophet import Prophet # type: ignore from prophet.diagnostics import cross_validation, performance_metrics # type: ignore from sklearn.ensemble import GradientBoostingRegressor warnings.filterwarnings("ignore") CSV_PATH = Path("nvda_2014_to_2026.csv") MODEL_PATH = Path("nvidia_price_model.pkl") FORECAST_PLOT_PATH = Path("nvidia_forecast.png") FORECAST_CSV_PATH = Path("nvidia_7day_forecast.csv") BACKTEST_CSV_PATH = Path("prophet_90day_backtest.csv") BASE_REGRESSORS = ["MA7", "MA30", "Vol7", "Volume_MA7"] ADVANCED_REGRESSORS = ["RSI", "MACD", "BB_upper", "BB_middle", "BB_lower"] REGRESSOR_COLUMNS = BASE_REGRESSORS + ADVANCED_REGRESSORS RESIDUAL_FEATURE_COLUMNS = REGRESSOR_COLUMNS + ["prophet_yhat"] PARAM_GRID = { "changepoint_prior_scale": [0.05, 0.1, 0.5, 0.8, 1.0], "seasonality_prior_scale": [0.01, 0.1, 1.0, 10.0], "holidays_prior_scale": [0.1, 1.0, 10.0], "seasonality_mode": ["additive", "multiplicative"], } CV_TOP_N = 3 CV_INITIAL = "730 days" CV_PERIOD = "30 days" CV_HORIZON = "7 days" CV_PARALLEL = None print("Training Prophet + residual ML ensemble with expanded hyperparameter tuning...") def load_price_data(csv_path: Path = CSV_PATH) -> pd.DataFrame: df = pd.read_csv(csv_path, skiprows=[1]) df = df.dropna(subset=["Date"]).copy() df["Date"] = pd.to_datetime(df["Date"], errors="coerce") for col in ["Open", "High", "Low", "Close", "Volume"]: df[col] = pd.to_numeric(df[col], errors="coerce") return df.dropna().sort_values("Date").reset_index(drop=True) def add_technical_features(df: pd.DataFrame) -> pd.DataFrame: df = df.copy() df["Return"] = df["Close"].pct_change() df["MA7"] = df["Close"].rolling(7).mean() df["MA30"] = df["Close"].rolling(30).mean() df["Vol7"] = df["Close"].rolling(7).std() df["Volume_MA7"] = df["Volume"].rolling(7).mean() delta = df["Close"].diff() avg_gain = delta.clip(lower=0).rolling(14).mean() avg_loss = (-delta.clip(upper=0)).rolling(14).mean().replace(0, np.nan) rs = avg_gain / avg_loss df["RSI"] = 100 - (100 / (1 + rs)) ema12 = df["Close"].ewm(span=12, adjust=False).mean() ema26 = df["Close"].ewm(span=26, adjust=False).mean() df["MACD"] = ema12 - ema26 bb_middle = df["Close"].rolling(20).mean() bb_std = df["Close"].rolling(20).std() df["BB_upper"] = bb_middle + 2 * bb_std df["BB_middle"] = bb_middle df["BB_lower"] = bb_middle - 2 * bb_std return df.dropna().reset_index(drop=True) def make_prophet_frame(df: pd.DataFrame) -> pd.DataFrame: prophet_df = df[["Date", "Close", *REGRESSOR_COLUMNS]].rename( columns={"Date": "ds", "Close": "y"} ) return prophet_df.dropna().sort_values("ds").reset_index(drop=True) def create_model(params: dict) -> Prophet: model = Prophet( daily_seasonality=True, weekly_seasonality=True, yearly_seasonality=True, interval_width=0.95, uncertainty_samples=300, **params, ) for regressor in REGRESSOR_COLUMNS: model.add_regressor(regressor) return model def evaluate_predictions(actual: pd.Series, predicted: pd.Series) -> dict: actual_values = actual.to_numpy(dtype=float) predicted_values = predicted.to_numpy(dtype=float) mae = np.mean(np.abs(actual_values - predicted_values)) rmse = np.sqrt(np.mean((actual_values - predicted_values) ** 2)) mape = np.mean(np.abs((actual_values - predicted_values) / actual_values)) * 100 actual_direction = np.sign(np.diff(actual_values)) predicted_direction = np.sign(np.diff(predicted_values)) directional_accuracy = (actual_direction == predicted_direction).mean() * 100 return { "mae": float(mae), "rmse": float(rmse), "mape": float(mape), "directional_accuracy": float(directional_accuracy), "rating": float(max(0, 100 - mape)), } def tune_prophet_params(prophet_df: pd.DataFrame, test_size: int = 30) -> tuple[dict, pd.DataFrame]: train = prophet_df.iloc[:-test_size].copy() test = prophet_df.iloc[-test_size:].copy() results = [] keys = list(PARAM_GRID.keys()) total_candidates = int(np.prod([len(values) for values in PARAM_GRID.values()])) print(f"\nStage 1: broad holdout tuning across {total_candidates} Prophet candidates...") for values in product(*PARAM_GRID.values()): params = dict(zip(keys, values)) model = create_model(params) model.fit(train) future = test[["ds", *REGRESSOR_COLUMNS]].copy() forecast = model.predict(future) metrics = evaluate_predictions(test["y"], forecast["yhat"]) results.append( { **params, **{f"holdout_{name}": value for name, value in metrics.items()}, "mape": metrics["mape"], "rmse": metrics["rmse"], "directional_accuracy": metrics["directional_accuracy"], "rating": metrics["rating"], } ) print( "Holdout tuned", params, f"MAPE={metrics['mape']:.2f}%", f"DirAcc={metrics['directional_accuracy']:.1f}%", ) tuning_df = pd.DataFrame(results).sort_values(["mape", "rmse"]).reset_index(drop=True) cv_candidate_count = min(CV_TOP_N, len(tuning_df)) print( f"\nStage 2: Prophet cross-validation on top {cv_candidate_count} candidates " f"(initial={CV_INITIAL}, period={CV_PERIOD}, horizon={CV_HORIZON})..." ) for idx in range(cv_candidate_count): params = {key: tuning_df.loc[idx, key] for key in keys} cv_model = create_model(params) cv_model.fit(train) df_cv = cross_validation( cv_model, initial=CV_INITIAL, period=CV_PERIOD, horizon=CV_HORIZON, parallel=CV_PARALLEL, ) perf = performance_metrics(df_cv) cv_metrics = evaluate_predictions(df_cv["y"], df_cv["yhat"]) cv_mape = float(perf["mape"].mean() * 100) cv_rmse = float(perf["rmse"].mean()) cv_mae = float(perf["mae"].mean()) tuning_df.loc[idx, "cv_mape"] = cv_mape tuning_df.loc[idx, "cv_rmse"] = cv_rmse tuning_df.loc[idx, "cv_mae"] = cv_mae tuning_df.loc[idx, "cv_directional_accuracy"] = cv_metrics["directional_accuracy"] tuning_df.loc[idx, "cv_rating"] = max(0, 100 - cv_mape) print( "CV tuned", params, f"CV_MAPE={cv_mape:.2f}%", f"CV_DirAcc={cv_metrics['directional_accuracy']:.1f}%", ) cv_ready = tuning_df.dropna(subset=["cv_mape"]).copy() if not cv_ready.empty: cv_ready = cv_ready.sort_values(["cv_mape", "cv_rmse", "holdout_mape"]).reset_index(drop=True) best_params = {key: cv_ready.loc[0, key] for key in keys} tuning_df["selected_by_cv"] = False selected_mask = np.ones(len(tuning_df), dtype=bool) for key, value in best_params.items(): selected_mask &= tuning_df[key] == value tuning_df.loc[selected_mask, "selected_by_cv"] = True tuning_df = tuning_df.sort_values( ["selected_by_cv", "cv_mape", "mape", "rmse"], ascending=[False, True, True, True], ).reset_index(drop=True) else: best_params = {key: tuning_df.loc[0, key] for key in keys} tuning_df["selected_by_cv"] = False tuning_df.loc[0, "selected_by_cv"] = True return best_params, tuning_df def make_future_with_regressors(model: Prophet, prophet_df: pd.DataFrame, periods: int) -> pd.DataFrame: future = model.make_future_dataframe(periods=periods, freq="B") future = future.merge(prophet_df[["ds", *REGRESSOR_COLUMNS]], on="ds", how="left") latest_regressors = prophet_df[REGRESSOR_COLUMNS].iloc[-1] for col in REGRESSOR_COLUMNS: future[col] = future[col].fillna(latest_regressors[col]) return future def format_7day_forecast(forecast_df: pd.DataFrame) -> pd.DataFrame: pred = forecast_df.tail(7)[["ds", "yhat", "yhat_lower", "yhat_upper"]].copy() pred = pred.rename( columns={ "ds": "Date", "yhat": "Predicted_Close", "yhat_lower": "Lower_Bound", "yhat_upper": "Upper_Bound", } ) pred["Predicted_Close"] = pred["Predicted_Close"].round(2) pred["Lower_Bound"] = pred["Lower_Bound"].round(2) pred["Upper_Bound"] = pred["Upper_Bound"].round(2) pred["Date"] = pd.to_datetime(pred["Date"]).dt.strftime("%Y-%m-%d") return pred[["Date", "Predicted_Close", "Lower_Bound", "Upper_Bound"]] def make_residual_features(feature_df: pd.DataFrame, forecast_df: pd.DataFrame) -> pd.DataFrame: residual_features = feature_df[["ds", *REGRESSOR_COLUMNS]].merge( forecast_df[["ds", "yhat"]].rename(columns={"yhat": "prophet_yhat"}), on="ds", how="left", ) return residual_features[RESIDUAL_FEATURE_COLUMNS].copy() def train_residual_model(feature_df: pd.DataFrame, forecast_df: pd.DataFrame) -> GradientBoostingRegressor: residual_train = feature_df[["ds", "y"]].merge( forecast_df[["ds", "yhat"]], on="ds", how="left", ) residual_train["residual"] = residual_train["y"] - residual_train["yhat"] X_train = make_residual_features(feature_df, forecast_df) y_train = residual_train["residual"] residual_model = GradientBoostingRegressor( n_estimators=250, learning_rate=0.03, max_depth=2, subsample=0.85, random_state=42, ) residual_model.fit(X_train, y_train) return residual_model def apply_residual_model( forecast_df: pd.DataFrame, feature_df: pd.DataFrame, residual_model: GradientBoostingRegressor, ) -> pd.DataFrame: ensemble_forecast = forecast_df.copy() X = make_residual_features(feature_df, forecast_df) correction = residual_model.predict(X) ensemble_forecast["prophet_yhat"] = ensemble_forecast["yhat"] ensemble_forecast["residual_correction"] = correction ensemble_forecast["yhat"] = ensemble_forecast["prophet_yhat"] + correction ensemble_forecast["yhat_lower"] = ensemble_forecast["yhat_lower"] + correction ensemble_forecast["yhat_upper"] = ensemble_forecast["yhat_upper"] + correction return ensemble_forecast df = add_technical_features(load_price_data()) prophet_df = make_prophet_frame(df) print(f"Cleaned dataset: {len(prophet_df)} trading days | Latest close: ${prophet_df['y'].iloc[-1]:.2f}") best_params, tuning_results = tune_prophet_params(prophet_df, test_size=30) print(f"\nBest Prophet params: {best_params}") validation_train = prophet_df.iloc[:-30].copy() validation_test = prophet_df.iloc[-30:].copy() validation_model = create_model(best_params) validation_model.fit(validation_train) validation_train_forecast = validation_model.predict(validation_train[["ds", *REGRESSOR_COLUMNS]]) validation_residual_model = train_residual_model(validation_train, validation_train_forecast) validation_prophet_forecast = validation_model.predict(validation_test[["ds", *REGRESSOR_COLUMNS]]) validation_forecast = apply_residual_model( validation_prophet_forecast, validation_test[["ds", *REGRESSOR_COLUMNS]], validation_residual_model, ) backtest = validation_test[["ds", "y"]].merge( validation_forecast[ ["ds", "yhat", "yhat_lower", "yhat_upper", "prophet_yhat", "residual_correction"] ], on="ds", how="left", ) metrics = evaluate_predictions(backtest["y"], backtest["yhat"]) prophet_only_metrics = evaluate_predictions( validation_test["y"], validation_prophet_forecast["yhat"], ) final_model = create_model(best_params) final_model.fit(prophet_df) full_history_forecast = final_model.predict(prophet_df[["ds", *REGRESSOR_COLUMNS]]) residual_model = train_residual_model(prophet_df, full_history_forecast) full_future = make_future_with_regressors(final_model, prophet_df, periods=7) prophet_forecast = final_model.predict(full_future) forecast = apply_residual_model( prophet_forecast, full_future[["ds", *REGRESSOR_COLUMNS]], residual_model, ) future_7day = format_7day_forecast(forecast) future_7day.to_csv(FORECAST_CSV_PATH, index=False) backtest.to_csv(BACKTEST_CSV_PATH, index=False) def predict_7_days_prophet() -> pd.DataFrame: """Return the next 7 business-day NVDA close forecasts from the fitted v2 model.""" return format_7day_forecast(forecast) print("\nNVIDIA 7-DAY PRICE TREND PREDICTION") print(predict_7_days_prophet().to_markdown(index=False)) print("\n=== Final 30-Day Backtest (Prophet + Residual ML Ensemble) ===") print(f"MAE : ${metrics['mae']:.2f}") print(f"RMSE : ${metrics['rmse']:.2f}") print(f"MAPE : {metrics['mape']:.2f}%") print(f"Rating: {metrics['rating']:.2f}%") print(f"Directional accuracy: {metrics['directional_accuracy']:.1f}%") print(f"Prophet-only MAPE before residual correction: {prophet_only_metrics['mape']:.2f}%") print(f"Last close: ${df['Close'].iloc[-1]:.2f}") checkpoint = { "model_version": "prophet_v3_residual_ensemble", "prophet_model": final_model, "residual_model": residual_model, "last_close": float(df["Close"].iloc[-1]), "last_date": df["Date"].iloc[-1], "backtest_mape": metrics["mape"], "backtest_mae": metrics["mae"], "backtest_rmse": metrics["rmse"], "rating": metrics["rating"], "directional_accuracy": metrics["directional_accuracy"], "prophet_only_backtest_mape": prophet_only_metrics["mape"], "prophet_only_directional_accuracy": prophet_only_metrics["directional_accuracy"], "best_params": best_params, "tuning_strategy": { "stage_1": "expanded holdout grid search", "stage_2": "Prophet cross_validation on top holdout candidates", "cv_top_n": CV_TOP_N, "cv_initial": CV_INITIAL, "cv_period": CV_PERIOD, "cv_horizon": CV_HORIZON, }, "regressor_columns": REGRESSOR_COLUMNS, "residual_feature_columns": RESIDUAL_FEATURE_COLUMNS, "latest_regressors": prophet_df[REGRESSOR_COLUMNS].iloc[-1].to_dict(), "tuning_results": tuning_results.to_dict(orient="records"), } joblib.dump(checkpoint, MODEL_PATH) fig = final_model.plot(forecast) plt.title("NVIDIA Stock Price Forecast - Prophet v3 Residual Ensemble") plt.xlabel("Date") plt.ylabel("Close Price ($)") fig.savefig(FORECAST_PLOT_PATH, bbox_inches="tight") plt.close(fig) print(f"\nForecast saved as {FORECAST_CSV_PATH}") print(f"Backtest saved as {BACKTEST_CSV_PATH}") print(f"Forecast plot saved as {FORECAST_PLOT_PATH}") print(f"Upgraded model saved as {MODEL_PATH}")