import gradio as gr import pandas as pd import numpy as np from datetime import datetime, timedelta from plotly.subplots import make_subplots import plotly.graph_objects as go from prophet.serialize import model_from_json from statsmodels.tsa.arima.model import ARIMA from statsmodels.tsa.stattools import adfuller from sklearn.preprocessing import MinMaxScaler from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score import tensorflow as tf from tensorflow.keras.models import Sequential, load_model from tensorflow.keras.layers import LSTM, Dense, Dropout import pickle import warnings warnings.filterwarnings('ignore') np.random.seed(42) tf.random.set_seed(42) # ============================================================================ # DATA LOADING # ============================================================================ def load_stock_data(): """Load AAPL stock data from CSV file""" try: df = pd.read_csv('AAPL_stock_data.csv', parse_dates=['Date']) df = df.sort_values('Date').reset_index(drop=True) df['Returns'] = df['Close'].pct_change() print(f"Loaded {len(df)} records from AAPL_stock_data.csv") print(f"Date Range: {df['Date'].min()} to {df['Date'].max()}") return df except Exception as e: print(f"Error loading CSV: {e}") raise Exception("Please upload AAPL_stock_data.csv file") # ============================================================================ # METRICS CALCULATION # ============================================================================ def calculate_metrics(y_true, y_pred): """Calculate evaluation metrics""" rmse = np.sqrt(mean_squared_error(y_true, y_pred)) mae = mean_absolute_error(y_true, y_pred) mape = np.mean(np.abs((y_true - y_pred) / y_true)) * 100 r2 = r2_score(y_true, y_pred) return rmse, mae, mape, r2 # ============================================================================ # STATIONARITY TEST # ============================================================================ def test_stationarity(timeseries): """Perform Augmented Dickey-Fuller test""" result = adfuller(timeseries.dropna(), autolag='AIC') return { 'statistic': result[0], 'pvalue': result[1], 'is_stationary': result[1] <= 0.05, 'critical_values': result[4] } # ============================================================================ # MODEL 1: ARIMA # ============================================================================ def load_arima_model(): """Load pre-trained ARIMA model""" try: with open('models/arima_model.pkl', 'rb') as f: model = pickle.load(f) print("ARIMA model loaded successfully") return model except Exception as e: print(f"ARIMA Load Error: {e}") return None # ============================================================================ # MODEL 2: PROPHET # ============================================================================ def load_prophet_model(): """Load pre-trained Prophet model""" try: # Load the model from JSON with open('models/prophet_model.json', 'r') as f: model = model_from_json(f.read()) print("Prophet model loaded successfully") return model except Exception as e: print(f"Prophet Load Error: {e}") return None # ============================================================================ # MODEL 3: LSTM # ============================================================================ def load_lstm_model(): """Load pre-trained LSTM model and scaler""" try: # Load the model lstm_model = load_model('models/lstm_model.h5') # Load the scaler with open('models/lstm_scaler.pkl', 'rb') as f: scaler = pickle.load(f) print("LSTM model and scaler loaded successfully") return lstm_model, scaler except Exception as e: print(f"LSTM Load Error: {e}") return None, None # ============================================================================ # EDA FUNCTION # ============================================================================ def generate_eda(): """Generate exploratory data analysis""" try: df = load_stock_data() # Create subplots fig = make_subplots( rows=2, cols=2, subplot_titles=('Stock Close Price Over Time', 'Trading Volume', 'Daily Returns', 'Returns Distribution'), specs=[[{"secondary_y": False}, {"secondary_y": False}], [{"secondary_y": False}, {"secondary_y": False}]] ) # Close Price fig.add_trace( go.Scatter(x=df['Date'], y=df['Close'], mode='lines', name='Close Price', line=dict(color='blue', width=1.5)), row=1, col=1 ) # Volume fig.add_trace( go.Bar(x=df['Date'], y=df['Volume'], name='Volume', marker=dict(color='green', opacity=0.6)), row=1, col=2 ) # Daily Returns fig.add_trace( go.Scatter(x=df['Date'], y=df['Returns'], mode='lines', name='Returns', line=dict(color='red', width=0.8)), row=2, col=1 ) fig.add_hline(y=0, line_dash="dash", line_color="black", row=2, col=1) # Returns Distribution fig.add_trace( go.Histogram(x=df['Returns'].dropna(), nbinsx=50, name='Distribution', marker=dict(color='purple', opacity=0.7)), row=2, col=2 ) fig.update_layout(height=800, showlegend=False, template='plotly_white') fig.update_xaxes(title_text="Date", row=1, col=1) fig.update_xaxes(title_text="Date", row=1, col=2) fig.update_xaxes(title_text="Date", row=2, col=1) fig.update_xaxes(title_text="Returns", row=2, col=2) fig.update_yaxes(title_text="Price ($)", row=1, col=1) fig.update_yaxes(title_text="Volume", row=1, col=2) fig.update_yaxes(title_text="Returns", row=2, col=1) fig.update_yaxes(title_text="Frequency", row=2, col=2) # Statistics stats = f""" ### 📊 Dataset Statistics **Price Statistics:** - Mean Close Price: ${df['Close'].mean():.2f} - Median Close Price: ${df['Close'].median():.2f} - Min Close Price: ${df['Close'].min():.2f} - Max Close Price: ${df['Close'].max():.2f} - Standard Deviation: ${df['Close'].std():.2f} **Returns Statistics:** - Mean Daily Return: {df['Returns'].mean()*100:.4f}% - Volatility (Std): {df['Returns'].std()*100:.4f}% - Min Return: {df['Returns'].min()*100:.2f}% - Max Return: {df['Returns'].max()*100:.2f}% **Dataset Info:** - Total Records: {len(df)} - Date Range: {df['Date'].min().strftime('%Y-%m-%d')} to {df['Date'].max().strftime('%Y-%m-%d')} - Missing Values: {df.isnull().sum().sum()} """ # Stationarity test stationarity_result = test_stationarity(df['Close']) stationarity_text = f""" ### 🔬 Stationarity Test (Augmented Dickey-Fuller) - **ADF Statistic:** {stationarity_result['statistic']:.6f} - **p-value:** {stationarity_result['pvalue']:.6f} - **Result:** {'✅ Series is STATIONARY' if stationarity_result['is_stationary'] else '❌ Series is NON-STATIONARY'} **Critical Values:** """ for key, value in stationarity_result['critical_values'].items(): stationarity_text += f"- {key}: {value:.3f}\n" return fig, stats, stationarity_text except Exception as e: error_msg = f"### ❌ Error\n\n{str(e)}" return None, error_msg, error_msg # ============================================================================ # ROLLING WINDOW EVALUATION # ============================================================================ def rolling_window_evaluation(df, test_size=90): """Perform rolling window evaluation for all models""" print("\n" + "="*70) print("RUNNING MODEL EVALUATION") print("="*70) train_data = df[:-test_size].copy() test_data = df[-test_size:].copy() print(f"Training set: {len(train_data)} days") print(f"Test set: {len(test_data)} days") results = { 'ARIMA': {'predictions': [], 'metrics': {}}, 'LSTM': {'predictions': [], 'metrics': {}}, 'Prophet': {'predictions': [], 'metrics': {}} } # ARIMA arima_model = load_arima_model() if arima_model: arima_forecast = arima_model.forecast(steps=test_size) results['ARIMA']['predictions'] = arima_forecast rmse, mae, mape, r2 = calculate_metrics(test_data['Close'].values, arima_forecast) results['ARIMA']['metrics'] = {'RMSE': rmse, 'MAE': mae, 'MAPE': mape, 'R2': r2} print(f"ARIMA - RMSE: {rmse:.2f}, MAE: {mae:.2f}, MAPE: {mape:.2f}%, R2: {r2:.4f}") # Prophet prophet_model = load_prophet_model() if prophet_model: future = prophet_model.make_future_dataframe(periods=test_size) forecast = prophet_model.predict(future) prophet_forecast = forecast['yhat'].iloc[-test_size:].values results['Prophet']['predictions'] = prophet_forecast rmse, mae, mape, r2 = calculate_metrics(test_data['Close'].values, prophet_forecast) results['Prophet']['metrics'] = {'RMSE': rmse, 'MAE': mae, 'MAPE': mape, 'R2': r2} print(f"Prophet - RMSE: {rmse:.2f}, MAE: {mae:.2f}, MAPE: {mape:.2f}%, R2: {r2:.4f}") # LSTM lstm_model, scaler = load_lstm_model() if lstm_model and scaler: all_data = df['Close'].values test_inputs = all_data[len(train_data) - 60:] # Using look_back=60 as in original code test_inputs = scaler.transform(test_inputs.reshape(-1, 1)) X_test = [] for i in range(60, len(test_inputs)): X_test.append(test_inputs[i-60:i, 0]) X_test = np.array(X_test) X_test = X_test.reshape((X_test.shape[0], X_test.shape[1], 1)) lstm_predictions = lstm_model.predict(X_test, verbose=0) lstm_predictions = scaler.inverse_transform(lstm_predictions) results['LSTM']['predictions'] = lstm_predictions.flatten() rmse, mae, mape, r2 = calculate_metrics(test_data['Close'].values, lstm_predictions.flatten()) results['LSTM']['metrics'] = {'RMSE': rmse, 'MAE': mae, 'MAPE': mape, 'R2': r2} print(f"LSTM - RMSE: {rmse:.2f}, MAE: {mae:.2f}, MAPE: {mape:.2f}%, R2: {r2:.4f}") print("="*70 + "\n") return results, train_data, test_data # ============================================================================ # FUTURE PREDICTIONS # ============================================================================ def predict_future(df, days_to_predict=30): """Make future predictions with all three models""" print("\n" + "="*70) print(f"GENERATING {days_to_predict}-DAY PREDICTIONS") print("="*70) full_data = df['Close'].values predictions = {} # ARIMA arima_model = load_arima_model() if arima_model: arima_forecast = arima_model.forecast(steps=days_to_predict) predictions['ARIMA'] = arima_forecast # Prophet prophet_model = load_prophet_model() if prophet_model: future = prophet_model.make_future_dataframe(periods=days_to_predict) forecast = prophet_model.predict(future) predictions['Prophet'] = forecast['yhat'].iloc[-days_to_predict:].values # LSTM lstm_model, scaler = load_lstm_model() if lstm_model and scaler: last_sequence = scaler.transform(full_data[-60:].reshape(-1, 1)) # Using look_back=60 lstm_predictions = [] for _ in range(days_to_predict): X_test = last_sequence.reshape((1, 60, 1)) pred = lstm_model.predict(X_test, verbose=0) lstm_predictions.append(pred[0, 0]) last_sequence = np.append(last_sequence[1:], pred).reshape(-1, 1) lstm_predictions = scaler.inverse_transform(np.array(lstm_predictions).reshape(-1, 1)) predictions['LSTM'] = lstm_predictions.flatten() print("="*70 + "\n") return predictions # ============================================================================ # VISUALIZATION FUNCTIONS # ============================================================================ def create_comparison_plot(results, train_data, test_data): """Create comparison plot for model evaluation""" fig = go.Figure() fig.add_trace(go.Scatter( x=train_data['Date'], y=train_data['Close'], mode='lines', name='Training Data', line=dict(color='blue', width=1.5) )) fig.add_trace(go.Scatter( x=test_data['Date'], y=test_data['Close'], mode='lines', name='Actual Test Data', line=dict(color='green', width=3) )) colors = {'ARIMA': 'red', 'LSTM': 'orange', 'Prophet': 'purple'} for model_name in ['ARIMA', 'LSTM', 'Prophet']: if len(results[model_name]['predictions']) > 0: fig.add_trace(go.Scatter( x=test_data['Date'], y=results[model_name]['predictions'], mode='lines', name=f'{model_name} Prediction', line=dict(color=colors[model_name], dash='dash', width=2) )) fig.update_layout( title='AAPL Stock Price - Model Comparison (Rolling Window Evaluation)', xaxis_title='Date', yaxis_title='Stock Price ($)', hovermode='x unified', height=600, template='plotly_white', legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1) ) return fig def create_future_plot(df, predictions, days_to_predict): """Create plot for future predictions""" fig = go.Figure() recent_data = df.tail(180) fig.add_trace(go.Scatter( x=recent_data['Date'], y=recent_data['Close'], mode='lines', name='Historical Data', line=dict(color='blue', width=2) )) last_date = df['Date'].iloc[-1] future_dates = pd.date_range(start=last_date + timedelta(days=1), periods=days_to_predict) colors = {'ARIMA': 'red', 'LSTM': 'orange', 'Prophet': 'purple'} for model_name, preds in predictions.items(): fig.add_trace(go.Scatter( x=future_dates, y=preds, mode='lines+markers', name=f'{model_name} Forecast', line=dict(color=colors[model_name], dash='dash', width=2), marker=dict(size=4) )) fig.update_layout( title=f'AAPL Stock Price Forecast - Next {days_to_predict} Days', xaxis_title='Date', yaxis_title='Stock Price ($)', hovermode='x unified', height=600, template='plotly_white', legend=dict(orientation="h", yanchor="bottom", y=1.02, xanchor="right", x=1) ) return fig # ============================================================================ # FORMATTING FUNCTIONS # ============================================================================ def format_metrics_table(results): """Format metrics as markdown table""" table = "### 📊 Model Performance Comparison\n\n" table += "| Model | RMSE | MAE | MAPE (%) | R² Score |\n" table += "|-------|------|-----|----------|----------|\n" for model_name in ['ARIMA', 'LSTM', 'Prophet']: metrics = results[model_name]['metrics'] if metrics: table += f"| **{model_name}** | {metrics['RMSE']:.2f} | {metrics['MAE']:.2f} | {metrics['MAPE']:.2f} | {metrics['R2']:.4f} |\n" best_model = min( results.keys(), key=lambda x: results[x]['metrics'].get('RMSE', float('inf')) if results[x]['metrics'] else float('inf') ) table += f"\n### 🏆 Best Model: **{best_model}** (Lowest RMSE)\n\n" table += "#### Model Characteristics:\n\n" table += "- **ARIMA**: Statistical model, captures linear trends and autocorrelation\n" table += "- **LSTM**: Deep learning model, captures complex non-linear patterns\n" table += "- **Prophet**: Hybrid model, robust to missing data and seasonality\n\n" table += "#### Evaluation Metrics:\n\n" table += "- **RMSE** (Root Mean Squared Error): Lower is better, penalizes large errors\n" table += "- **MAE** (Mean Absolute Error): Average prediction error in dollars\n" table += "- **MAPE** (Mean Absolute Percentage Error): Prediction error as percentage\n" table += "- **R²** (R-squared): Proportion of variance explained (closer to 1 is better)\n" return table def format_prediction_summary(df, predictions, days_to_predict): """Format prediction summary""" current_price = df['Close'].iloc[-1] current_date = df['Date'].iloc[-1].strftime('%Y-%m-%d') summary = f"### 📈 AAPL Stock Price Predictions ({days_to_predict} days ahead)\n\n" summary += f"**Current Price** (as of {current_date}): **${current_price:.2f}**\n\n" summary += "| Model | Predicted Price | Change ($) | Change (%) |\n" summary += "|-------|----------------|------------|------------|\n" for model_name, preds in predictions.items(): pred_price = preds[-1] change = pred_price - current_price pct_change = (change / current_price) * 100 emoji = "📈" if change > 0 else "📉" summary += f"| **{model_name}** {emoji} | ${pred_price:.2f} | ${change:+.2f} | {pct_change:+.2f}% |\n" avg_prediction = np.mean([preds[-1] for preds in predictions.values()]) avg_change = avg_prediction - current_price avg_pct_change = (avg_change / current_price) * 100 summary += f"\n**Average Prediction**: ${avg_prediction:.2f} ({avg_pct_change:+.2f}%)\n\n" summary += "#### ⚠️ Disclaimer\n" summary += "*These predictions are for educational purposes only. Stock market predictions are inherently uncertain. " summary += "Past performance does not guarantee future results. Always consult with financial advisors before making investment decisions.*\n" return summary # ============================================================================ # MAIN INTERFACE FUNCTIONS # ============================================================================ def run_evaluation(): """Run model evaluation""" try: df = load_stock_data() results, train_data, test_data = rolling_window_evaluation(df, test_size=90) plot = create_comparison_plot(results, train_data, test_data) metrics_table = format_metrics_table(results) return plot, metrics_table except Exception as e: error_msg = f"### ❌ Error\n\n{str(e)}\n\nPlease make sure AAPL_stock_data.csv is uploaded." return None, error_msg def run_prediction(days_to_predict): """Run future prediction""" try: df = load_stock_data() predictions = predict_future(df, int(days_to_predict)) plot = create_future_plot(df, predictions, int(days_to_predict)) summary = format_prediction_summary(df, predictions, int(days_to_predict)) return plot, summary except Exception as e: error_msg = f"### ❌ Error\n\n{str(e)}\n\nPlease make sure AAPL_stock_data.csv is uploaded." return None, error_msg # ============================================================================ # GRADIO INTERFACE # ============================================================================ custom_css = """ .gradio-container { font-family: 'Arial', sans-serif; } .main-header { text-align: center; padding: 20px; background: linear-gradient(90deg, #667eea 0%, #764ba2 100%); color: white; border-radius: 10px; margin-bottom: 20px; } """ with gr.Blocks(title="Stock Price Forecasting - DataSynthis_ML_JobTask", theme=gr.themes.Soft(), css=custom_css) as demo: gr.Markdown("""
DataSynthis ML Job Task - SET B
Time-Series Forecasting – Stock Prices
Deployed on Hugging Face Spaces | Powered by Gradio