train_nvidia_ml_model.py

# 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}")