Sadat1210's picture
Update app.py
374f290 verified
Raw History Blame Contribute Delete
27.6 kB
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("""
<div class="main-header">
<h1>๐Ÿ“ˆ AAPL Stock Price Time-Series Forecasting</h1>
<h3>DataSynthis ML Job Task - SET B</h3>
</div>
""")
gr.Markdown("""
This application implements and compares **three time-series forecasting models** for Apple Inc. (AAPL) stock price prediction:
- ๐Ÿ”ด **ARIMA** (AutoRegressive Integrated Moving Average) - Traditional statistical model
- ๐ŸŸ  **LSTM** (Long Short-Term Memory) - Deep learning neural network
- ๐ŸŸฃ **Prophet** - Facebook's robust forecasting tool
---
""")
with gr.Tabs() as tabs:
# TAB 1: Exploratory Data Analysis
with gr.Tab("๐Ÿ“Š Exploratory Data Analysis"):
gr.Markdown("""
### Dataset Overview & Statistical Analysis
Explore the AAPL stock price data with comprehensive visualizations and statistical summaries.
""")
eda_btn = gr.Button("๐Ÿ” Generate EDA Report", variant="primary", size="lg")
eda_plot = gr.Plot(label="EDA Visualizations")
with gr.Row():
with gr.Column():
eda_stats = gr.Markdown()
with gr.Column():
eda_stationarity = gr.Markdown()
eda_btn.click(
fn=generate_eda,
inputs=[],
outputs=[eda_plot, eda_stats, eda_stationarity]
)
# TAB 2: Future Prediction
with gr.Tab("๐Ÿ”ฎ Future Prediction"):
gr.Markdown("""
### Predict Future Stock Prices
Select the number of days you want to predict and click **Generate Predictions** to see forecasts from all three models.
""")
with gr.Row():
with gr.Column(scale=1):
days_slider = gr.Slider(
minimum=7,
maximum=90,
value=30,
step=1,
label="๐Ÿ“… Days to Predict",
info="Choose between 7 and 90 days"
)
predict_btn = gr.Button("๐Ÿš€ Generate Predictions", variant="primary", size="lg")
gr.Markdown("#### Quick Options:")
with gr.Row():
btn_7 = gr.Button("1 Week", size="sm")
btn_30 = gr.Button("1 Month", size="sm")
btn_60 = gr.Button("2 Months", size="sm")
btn_90 = gr.Button("3 Months", size="sm")
btn_7.click(lambda: 7, outputs=days_slider)
btn_30.click(lambda: 30, outputs=days_slider)
btn_60.click(lambda: 60, outputs=days_slider)
btn_90.click(lambda: 90, outputs=days_slider)
prediction_plot = gr.Plot(label="๐Ÿ“Š Forecast Chart")
prediction_summary = gr.Markdown()
predict_btn.click(
fn=run_prediction,
inputs=[days_slider],
outputs=[prediction_plot, prediction_summary]
)
# TAB 3: Model Evaluation
with gr.Tab("๐Ÿ“Š Model Evaluation"):
gr.Markdown("""
### Rolling Window Evaluation
This evaluation uses the **last 90 days** as test data to compare model performance.
**Evaluation Metrics:**
- **RMSE**: Root Mean Squared Error (lower is better)
- **MAE**: Mean Absolute Error (lower is better)
- **MAPE**: Mean Absolute Percentage Error (lower is better)
- **Rยฒ**: R-squared score (closer to 1 is better)
Click the button below to run the evaluation:
""")
eval_btn = gr.Button("๐Ÿ”ฌ Run Evaluation", variant="primary", size="lg")
evaluation_plot = gr.Plot(label="๐Ÿ“Š Model Comparison Chart")
metrics_output = gr.Markdown()
eval_btn.click(
fn=run_evaluation,
inputs=[],
outputs=[evaluation_plot, metrics_output]
)
# TAB 3: About
with gr.Tab("โ„น๏ธ About"):
gr.Markdown("""
## ๐Ÿ“ Project Information
### Models Implemented
#### 1๏ธโƒฃ ARIMA (5,1,0)
- **Type**: Statistical time-series model
- **Strengths**: Fast, interpretable, good for linear trends
- **Use Case**: Short-term predictions, linear patterns
- **Training Time**: < 1 second
#### 2๏ธโƒฃ LSTM Neural Network
- **Type**: Deep learning recurrent neural network
- **Architecture**: 2 LSTM layers (50 units each) with dropout
- **Strengths**: Captures complex non-linear patterns
- **Use Case**: Medium to long-term predictions
- **Training Time**: ~30-60 seconds
#### 3๏ธโƒฃ Prophet
- **Type**: Additive regression model by Facebook
- **Strengths**: Handles seasonality, robust to missing data
- **Use Case**: Business forecasting, interpretable components
- **Training Time**: ~5-10 seconds
---
### Dataset Information
- **Stock**: Apple Inc. (AAPL)
- **Features**: Date, Open, High, Low, Close, Volume
- **Target Variable**: Close Price
- **Data Source**: Historical stock data (CSV)
---
### Technical Stack
- **Framework**: Gradio 4.44.0
- **Deep Learning**: TensorFlow 2.15.0 / Keras
- **Statistical Models**: Statsmodels 0.14.0
- **Forecasting**: Prophet 1.1.5
- **Data Processing**: Pandas, NumPy, Scikit-learn
- **Visualization**: Plotly
---
### Deployment
- **Platform**: Hugging Face Spaces
- **Space Name**: DataSynthis_ML_JobTask
---
### ๐Ÿ‘จโ€๐Ÿ’ป Author
**Md. Sadat Ahmed Rafi**
Task: Time-Series Forecasting โ€“ Stock Prices
---
### ๐Ÿ“„ License
This project is created for the DataSynthis ML Job Task evaluation.
""")
gr.Markdown("""
---
<div style="text-align: center; color: #666; padding: 20px;">
<p><strong>DataSynthis ML Job Task - SET B</strong></p>
<p>Time-Series Forecasting โ€“ Stock Prices</p>
<p>Deployed on Hugging Face Spaces | Powered by Gradio</p>
</div>
""")
# ============================================================================
# LAUNCH
# ============================================================================
if __name__ == "__main__":
demo.launch()