import gradio as gr
import pandas as pd
import numpy as np
import torch
import torch.nn as nn
from torch.utils.data import DataLoader, TensorDataset
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import r2_score, mean_squared_error, mean_absolute_error
import matplotlib.pyplot as plt
import os
# Load and preprocess data
df = pd.read_csv("haddonfield_chocolate_sales_2016_2022.csv")
features = ['temperature', 'holiday', 'town_event', 'day_of_week', 'season_code']
target = 'sales'
# Add lag feature (sales 7 days ago)
df['sales_lag_7'] = df['sales'].shift(7)
df.dropna(inplace=True)
features.append('sales_lag_7')
scaler_X = MinMaxScaler()
scaler_y = MinMaxScaler()
X_scaled = scaler_X.fit_transform(df[features])
y_scaled = scaler_y.fit_transform(df[[target]])
# Sequence creation
def create_sequences(X, y, window_size=60):
X_seq, y_seq = [], []
for i in range(len(X) - window_size):
X_seq.append(X[i:i+window_size])
y_seq.append(y[i+window_size])
return np.array(X_seq), np.array(y_seq)
window_size = 60
X_seq, y_seq = create_sequences(X_scaled, y_scaled, window_size)
# Train/test split
split_index = int(0.8 * len(X_seq))
X_train, X_test = X_seq[:split_index], X_seq[split_index:]
y_train, y_test = y_seq[:split_index], y_seq[split_index:]
# Convert to tensors
torch_X_train = torch.tensor(X_train, dtype=torch.float32)
torch_y_train = torch.tensor(y_train, dtype=torch.float32)
torch_X_test = torch.tensor(X_test, dtype=torch.float32)
torch_y_test = torch.tensor(y_test, dtype=torch.float32)
train_loader = DataLoader(TensorDataset(torch_X_train, torch_y_train), batch_size=32, shuffle=True)
# Define LSTM model
class LSTMModel(nn.Module):
def __init__(self, input_size, hidden_size=64, num_layers=2):
super(LSTMModel, self).__init__()
self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
self.fc = nn.Linear(hidden_size, 1)
def forward(self, x):
lstm_out, _ = self.lstm(x)
return self.fc(lstm_out[:, -1, :])
input_size = X_train.shape[2]
model = LSTMModel(input_size)
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
latest_evaluation_result = ""
def evaluate_model(model, X_test, y_test):
global latest_evaluation_result
model.eval()
with torch.no_grad():
preds = model(X_test).numpy()
true_vals = y_test.numpy()
preds_inv = scaler_y.inverse_transform(preds)
true_inv = scaler_y.inverse_transform(true_vals)
r2 = r2_score(true_inv, preds_inv)
rmse = np.sqrt(mean_squared_error(true_inv, preds_inv))
mae = mean_absolute_error(true_inv, preds_inv)
mape = np.mean(np.abs((true_inv - preds_inv) / true_inv)) * 100
latest_evaluation_result = (
f"📊 Model Evaluation Metrics:\n"
f"• R² Score: {r2:.4f}\n"
f"• RMSE: {rmse:.2f}\n"
f"• MAE: {mae:.2f}\n"
f"• MAPE: {mape:.2f}%"
)
plt.figure(figsize=(10,5))
plt.plot(true_inv, label='Actual Sales')
plt.plot(preds_inv, label='Predicted Sales')
plt.xlabel('Test Sample Index')
plt.ylabel('Sales ($)')
plt.title('Predicted vs Actual Chocolate Sales')
plt.legend()
plt.tight_layout()
fig = plt.gcf()
plt.close()
return latest_evaluation_result, fig
# Train or load model
if os.path.exists("lstm_model.pt"):
model.load_state_dict(torch.load("lstm_model.pt"))
model.eval()
latest_evaluation_result, _ = evaluate_model(model, torch_X_test, torch_y_test)
else:
for epoch in range(50):
model.train()
epoch_loss = 0
for X_batch, y_batch in train_loader:
optimizer.zero_grad()
output = model(X_batch)
loss = criterion(output, y_batch)
loss.backward()
optimizer.step()
epoch_loss += loss.item()
print(f"Epoch {epoch+1}/50, Loss: {epoch_loss:.4f}")
torch.save(model.state_dict(), "lstm_model.pt")
latest_evaluation_result, _ = evaluate_model(model, torch_X_test, torch_y_test)
# Prediction function
def predict_sales_and_evaluate(temp, holiday, event, weekday, season_code):
recent = X_scaled[-(window_size-1):].tolist()
sales_lag = df[target].iloc[-7]
inp = [temp, holiday, event, weekday, season_code, sales_lag]
inp_scaled = scaler_X.transform([inp])[0]
recent.append(inp_scaled)
seq = torch.tensor([recent], dtype=torch.float32)
model.eval()
with torch.no_grad():
pred = model(seq).numpy()
pred_inv = scaler_y.inverse_transform(pred)[0][0]
eval_txt, fig = evaluate_model(model, torch_X_test, torch_y_test)
return round(pred_inv,2), eval_txt, fig
# Gradio app
with gr.Blocks() as demo:
# Logo top-left
gr.HTML("""
"""
)
with gr.Tab("📈 Predict Sales"):
gr.Markdown("### 🍫 Chocolate Sales Predictor (LSTM)")
gr.Markdown("Predict next-day chocolate sales based on weather, holidays, and events in Haddonfield, NJ.")
temp = gr.Slider(0, 100, label="Temperature (°F)", value=70)
holiday = gr.Radio([0,1], label="Holiday?")
event = gr.Radio([0,1], label="Town Event?")
weekday = gr.Slider(0,6, step=1, label="Day of Week (0=Mon)")
season_code = gr.Dropdown([0,1,2,3], label="Season (0=Winter, 3=Fall)")
out_num = gr.Number(label="Predicted Sales ($)")
out_text = gr.Textbox(label="Latest Evaluation Metrics", lines=6)
out_plot = gr.Plot()
gr.Button("Predict").click(
fn=predict_sales_and_evaluate,
inputs=[temp, holiday, event, weekday, season_code],
outputs=[out_num, out_text, out_plot]
)
with gr.Tab("📊 Model Accuracy"):
gr.Markdown("### 🔍 Evaluate LSTM Accuracy on Test Set")
eval_out = gr.Textbox(label="Evaluation Results", lines=6, value=latest_evaluation_result)
eval_plot = gr.Plot()
gr.Button("Re-run Evaluation").click(
fn=lambda: evaluate_model(model, torch_X_test, torch_y_test),
inputs=[],
outputs=[eval_out, eval_plot]
)
# Footer
gr.HTML("""
"""
)
demo.launch()
