Coffee Shop LSTM Testing

import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout, Bidirectional, LayerNormalization
import matplotlib.pyplot as plt

# Load the coffee shop data
data = pd.read_csv("coffee_shop_revenue.csv")

# Handle outliers in Daily_Revenue
revenue_lower, revenue_upper = np.percentile(data["Daily_Revenue"], [1, 99])
data["Daily_Revenue"] = np.clip(data["Daily_Revenue"], revenue_lower, revenue_upper)

# Add temporal features
dates = pd.date_range(start='2024-01-01', periods=len(data), freq='D')
data["Day_of_Week"] = dates.dayofweek
data["Month_of_Year"] = dates.month
data["Is_Holiday"] = ((data["Month_of_Year"] == 12) & (dates.day.isin([24, 25, 31])) | 
                      (data["Month_of_Year"] == 1) & (dates.day == 1)).astype(int)

# Define features and target
features = ["Number_of_Customers_Per_Day", "Average_Order_Value", "Operating_Hours_Per_Day", 
            "Number_of_Employees", "Marketing_Spend_Per_Day", "Location_Foot_Traffic", 
            "Day_of_Week", "Month_of_Year", "Is_Holiday"]
target = "Daily_Revenue"

# Add interaction terms
data["Customer_Order_Interaction"] = data["Number_of_Customers_Per_Day"] * data["Average_Order_Value"]
data["Marketing_Foot_Traffic_Interaction"] = data["Marketing_Spend_Per_Day"] * data["Location_Foot_Traffic"]
features.extend(["Customer_Order_Interaction", "Marketing_Foot_Traffic_Interaction"])

# Add high marketing spend indicator
marketing_threshold = np.percentile(data["Marketing_Spend_Per_Day"], 75)
data["High_Marketing_Spend"] = (data["Marketing_Spend_Per_Day"] > marketing_threshold).astype(int)
features.append("High_Marketing_Spend")

# Add lagged and moving average features
data["lagged_revenue_1"] = data["Daily_Revenue"].shift(1).fillna(data["Daily_Revenue"].mean())
data["lagged_revenue_7"] = data["Daily_Revenue"].shift(7).fillna(data["Daily_Revenue"].mean())
data["lagged_revenue_14"] = data["Daily_Revenue"].shift(14).fillna(data["Daily_Revenue"].mean())
data["lagged_revenue_30"] = data["Daily_Revenue"].shift(30).fillna(data["Daily_Revenue"].mean())
data["ma_revenue_7"] = data["Daily_Revenue"].rolling(window=7).mean().fillna(data["Daily_Revenue"].mean())
features.extend(["lagged_revenue_1", "lagged_revenue_7", "lagged_revenue_14", "lagged_revenue_30", "ma_revenue_7"])

# Initialize the scaler
scaler = MinMaxScaler()
scaled_data = scaler.fit_transform(data[features + [target]])

# Debug scaler shape
print(f"Scaled data shape: {scaled_data.shape}")
print(f"Scaler min_ shape: {scaler.min_.shape}")
print(f"Scaler scale_ shape: {scaler.scale_.shape}")

# Prepare validation sequences
lookback = 30
val_size = 200
train_data = scaled_data[:-val_size]
val_data = scaled_data[-val_size-lookback:]

val_X, val_y = [], []
for i in range(lookback, len(val_data)):
    val_X.append(val_data[i - lookback:i, :15])  
    val_y.append(val_data[i, 15])  
val_X = np.array(val_X)
val_y = np.array(val_y)

# Build the LSTM model
model = Sequential()
model.add(Bidirectional(LSTM(64, activation='tanh', input_shape=(lookback, 15), return_sequences=True)))
model.add(LayerNormalization())
model.add(Dropout(0.1))
model.add(LSTM(64, activation='tanh', return_sequences=False))
model.add(LayerNormalization())
model.add(Dropout(0.1))
model.add(Dense(32, activation='relu'))
model.add(Dense(1))

# Build model with input shape
model.build(input_shape=(None, lookback, 15))

# Load the trained weights
model.load_weights("lstm_weights_coffee.weights.h5")

# Make predictions
predictions = model.predict(val_X, verbose=0)

# Inverse transform predictions
pred_scaled_array = np.zeros((len(predictions), len(features) + 1))  
pred_scaled_array[:, -1] = predictions.flatten()
pred_unscaled = scaler.inverse_transform(pred_scaled_array)[:, -1]

# Inverse transform actual values
actual_scaled_array = np.zeros((len(val_y), len(features) + 1))
actual_scaled_array[:, -1] = val_y
actual_unscaled = scaler.inverse_transform(actual_scaled_array)[:, -1]

# Calculate errors
absolute_errors = np.abs(pred_unscaled - actual_unscaled)
mae = np.mean(absolute_errors)
rmse = np.sqrt(np.mean((pred_unscaled - actual_unscaled) ** 2))

# Print results
print("Validation Set Results:")
print(f"Mean Absolute Error (MAE): {mae:.2f}")
print(f"Root Mean Squared Error (RMSE): {rmse:.2f}")
print("\nSample Predictions (First 5):")
for i in range(min(5, len(pred_unscaled))):
    print(f"Prediction {i+1}: Predicted = {pred_unscaled[i]:.2f}, Actual = {actual_unscaled[i]:.2f}, Error = {absolute_errors[i]:.2f}")

# Plot actual vs predicted values and display
plt.figure(figsize=(12, 6))
dates = dates[-val_size:]  # Get dates for the test set
plt.plot(dates, actual_unscaled, label='Actual Daily Revenue', color='blue', marker='o')
plt.plot(dates, pred_unscaled, label='Predicted Daily Revenue', color='red', linestyle='--', marker='x')
plt.xlabel('Date')
plt.ylabel('Daily Revenue (USD)')
plt.title('Actual vs Predicted Daily Revenue')
plt.legend()
plt.grid(True)
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()
plt.close()