Is Your SME Losing Sales? Machine Learning Demand Forecasting to the Rescue

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

def prepare_sales_data(sales_file):
    """
    Prepare sales data for demand forecasting models
    """
    # Load sales data
    df = pd.read_csv(sales_file)
    
    # Convert date to datetime
    df['date'] = pd.to_datetime(df['date'])
    
    # Add temporal features
    df['year'] = df['date'].dt.year
    df['month'] = df['date'].dt.month
    df['day_of_week'] = df['date'].dt.dayofweek
    df['day_of_month'] = df['date'].dt.day
    df['week_of_year'] = df['date'].dt.isocalendar().week
    
    # Identify holidays (example for Spain)
    spanish_holidays = [
        '2024-01-01', '2024-01-06', '2024-04-01', '2024-05-01',
        '2024-08-15', '2024-10-12', '2024-11-01', '2024-12-06',
        '2024-12-08', '2024-12-25'
    ]
    
    df['is_holiday'] = df['date'].dt.strftime('%Y-%m-%d').isin(spanish_holidays)
    
    # Detect high demand periods
    df['is_black_friday'] = (
        (df['month'] == 11) & 
        (df['day_of_week'] == 4) & 
        (df['day_of_month'] >= 22) & 
        (df['day_of_month'] <= 28)
    )
    
    df['is_christmas'] = (
        (df['month'] == 12) & 
        (df['day_of_month'] >= 15)
    )
    
    # Calculate lag features (previous sales)
    for product in df['product_id'].unique():
        mask = df['product_id'] == product
        df.loc[mask, 'sales_lag_7'] = df.loc[mask, 'units_sold'].shift(7)
        df.loc[mask, 'sales_lag_30'] = df.loc[mask, 'units_sold'].shift(30)
        df.loc[mask, 'moving_avg_7'] = df.loc[mask, 'units_sold'].rolling(7).mean()
        df.loc[mask, 'moving_avg_30'] = df.loc[mask, 'units_sold'].rolling(30).mean()
    
    # Calculate growth trend
    df['trend_7d'] = (df['moving_avg_7'] - df['sales_lag_7']) / df['sales_lag_7']
    
    # Detect seasonal products
    df['coef_variation'] = df.groupby('product_id')['units_sold'].transform(
        lambda x: x.std() / x.mean() if x.mean() > 0 else 0
    )
    
    # Mark seasonal products (high variability)
    df['is_seasonal'] = df['coef_variation'] > 0.5
    
    # Clean null values
    df = df.fillna(0)
    
    return df

# Usage example
# prepared_data = prepare_sales_data('historical_sales.csv')
# print(f"Prepared data: {prepared_data.shape[0]} records with {prepared_data.shape[1]} features")

from sklearn.linear_model import LinearRegression
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_absolute_error, mean_squared_error
import numpy as np

def simple_regression_model(data):
    """
    Implements a linear regression model for demand forecasting
    """
    # Select relevant features
    features = [
        'day_of_week', 'month', 'is_holiday', 'is_black_friday',
        'sales_lag_7', 'moving_avg_30', 'price'
    ]
    
    X = data[features]
    y = data['units_sold']
    
    # Split into training and test
    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=42, shuffle=False
    )
    
    # Train model
    model = LinearRegression()
    model.fit(X_train, y_train)
    
    # Make predictions
    y_pred = model.predict(X_test)
    
    # Evaluate performance
    mae = mean_absolute_error(y_test, y_pred)
    rmse = np.sqrt(mean_squared_error(y_test, y_pred))
    
    # Calculate accuracy as percentage
    mape = np.mean(np.abs((y_test - y_pred) / y_test)) * 100
    
    print(f"Mean Absolute Error: {mae:.2f} units")
    print(f"Root Mean Square Error: {rmse:.2f} units")
    print(f"Mean Absolute Percentage Error: {mape:.1f}%")
    print(f"Model accuracy: {100-mape:.1f}%")
    
    # Show feature importance
    importance = pd.DataFrame({
        'feature': features,
        'coefficient': model.coef_,
        'abs_importance': np.abs(model.coef_)
    }).sort_values('abs_importance', ascending=False)
    
    print("\nFeature importance:")
    print(importance)
    
    return model, mae, rmse, mape

# Function to predict future sales
def predict_future_demand(model, base_data, future_days=30):
    """
    Predicts demand for the next N days
    """
    # Get the last record as base
    last_record = base_data.iloc[-1].copy()
    predictions = []
    
    for day in range(1, future_days + 1):
        # Update date
        future_date = last_record['date'] + timedelta(days=day)
        
        # Update temporal features
        last_record['day_of_week'] = future_date.dayofweek
        last_record['month'] = future_date.month
        last_record['day_of_month'] = future_date.day
        
        # Prepare data for prediction
        X_future = last_record[[
            'day_of_week', 'month', 'is_holiday', 'is_black_friday',
            'sales_lag_7', 'moving_avg_30', 'price'
        ]].values.reshape(1, -1)
        
        # Predict
        prediction = model.predict(X_future)[0]
        predictions.append({
            'date': future_date,
            'predicted_demand': max(0, int(prediction))  # Cannot be negative
        })
        
        # Update lag features for next iteration
        last_record['sales_lag_7'] = prediction
    
    return pd.DataFrame(predictions)

# Example integration with Shopify to get sales data
import shopify
import pandas as pd
from datetime import datetime, timedelta

def extract_shopify_data(api_key, password, shop_name, historical_days=365):
    """
    Extract sales data from Shopify for ML training
    """
    # Configure connection
    shopify.ShopifyResource.set_site(f"https://{api_key}:{password}@{shop_name}.myshopify.com/admin")
    
    # Start date
    start_date = datetime.now() - timedelta(days=historical_days)
    
    # Get orders
    orders = shopify.Order.find(
        status='any',
        created_at_min=start_date.isoformat(),
        limit=250
    )
    
    sales_data = []
    
    for order in orders:
        for item in order.line_items:
            sales_data.append({
                'date': pd.to_datetime(order.created_at).date(),
                'product_id': item.product_id,
                'variant_id': item.variant_id,
                'sku': item.sku,
                'product_name': item.name,
                'units_sold': item.quantity,
                'unit_price': float(item.price),
                'total_sales': float(item.price) * item.quantity,
                'discount': float(getattr(item, 'total_discount', 0))
            })
    
    df = pd.DataFrame(sales_data)
    
    # Aggregate by day and product
    df_aggregated = df.groupby(['date', 'product_id', 'sku']).agg({
        'units_sold': 'sum',
        'unit_price': 'mean',
        'total_sales': 'sum',
        'product_name': 'first'
    }).reset_index()
    
    return df_aggregated

# Example integration with WooCommerce
from woocommerce import API

def extract_woocommerce_data(url, consumer_key, consumer_secret, historical_days=365):
    """
    Extract sales data from WooCommerce
    """
    wcapi = API(
        url=url,
        consumer_key=consumer_key,
        consumer_secret=consumer_secret,
        version="wc/v3",
        timeout=30
    )
    
    # Start date
    start_date = datetime.now() - timedelta(days=historical_days)
    
    # Get orders
    orders = wcapi.get("orders", params={
        "after": start_date.isoformat(),
        "status": "completed",
        "per_page": 100
    })
    
    sales_data = []
    
    for order in orders.json():
        for item in order['line_items']:
            sales_data.append({
                'date': pd.to_datetime(order['date_created']).date(),
                'product_id': item['product_id'],
                'sku': item['sku'],
                'product_name': item['name'],
                'units_sold': item['quantity'],
                'unit_price': float(item['price']),
                'total_sales': float(item['total'])
            })
    
    return pd.DataFrame(sales_data)

# Unified function for multiple platforms
def get_sales_data(platform, **kwargs):
    """
    Unified function to extract data from different platforms
    """
    if platform == 'shopify':
        return extract_shopify_data(**kwargs)
    elif platform == 'woocommerce':
        return extract_woocommerce_data(**kwargs)
    else:
        raise ValueError(f"Platform {platform} not supported")

import pandas as pd
import matplotlib.pyplot as plt
import seaborn as sns
from datetime import datetime, timedelta

class MLMonitoring:
    def __init__(self):
        self.historical_metrics = []
        
    def calculate_mape(self, y_real, y_predicted):
        """
        Calculate Mean Absolute Percentage Error
        """
        return np.mean(np.abs((y_real - y_predicted) / y_real)) * 100
    
    def calculate_bias(self, y_real, y_predicted):
        """
        Calculate prediction bias (tendency to over/underestimate)
        """
        return np.mean((y_predicted - y_real) / y_real) * 100
    
    def evaluate_daily_model(self, date, predictions, actual_sales):
        """
        Evaluate model performance for a specific day
        """
        mape = self.calculate_mape(actual_sales, predictions)
        bias = self.calculate_bias(actual_sales, predictions)
        
        # Calculate additional metrics
        mae = np.mean(np.abs(actual_sales - predictions))
        rmse = np.sqrt(np.mean((actual_sales - predictions)**2))
        
        metrics = {
            'date': date,
            'mape': mape,
            'bias': bias,
            'mae': mae,
            'rmse': rmse,
            'accuracy': max(0, 100 - mape),
            'num_products': len(predictions)
        }
        
        self.historical_metrics.append(metrics)
        return metrics
    
    def generate_weekly_report(self):
        """
        Generate weekly performance report
        """
        if len(self.historical_metrics) < 7:
            print("Insufficient data for weekly report")
            return
        
        # Get last 7 days
        df_metrics = pd.DataFrame(self.historical_metrics[-7:])
        
        # Calculate weekly statistics
        report = {
            'average_mape': df_metrics['mape'].mean(),
            'median_mape': df_metrics['mape'].median(),
            'average_bias': df_metrics['bias'].mean(),
            'average_accuracy': df_metrics['accuracy'].mean(),
            'days_meeting_target': (df_metrics['mape'] < 15).sum(),
            'improvement_trend': df_metrics['mape'].iloc[-3:].mean() < df_metrics['mape'].iloc[:3].mean()
        }
        
        # Generate alert if necessary
        if report['average_mape'] > 20:
            print("⚠️ ALERT: Accuracy below target")
            print(f"Average MAPE: {report['average_mape']:.1f}% (target: <15%)")
        
        return report
    
    def metrics_dashboard(self):
        """
        Create visual metrics dashboard
        """
        df = pd.DataFrame(self.historical_metrics)
        
        fig, axes = plt.subplots(2, 2, figsize=(15, 10))
        
        # MAPE over time
        axes[0,0].plot(df['date'], df['mape'])
        axes[0,0].axhline(y=15, color='r', linestyle='--', label='Target')
        axes[0,0].set_title('MAPE Evolution')
        axes[0,0].set_ylabel('MAPE (%)')
        axes[0,0].legend()
        
        # Bias over time
        axes[0,1].plot(df['date'], df['bias'])
        axes[0,1].axhline(y=0, color='g', linestyle='-', alpha=0.3)
        axes[0,1].axhline(y=5, color='r', linestyle='--', label='Limit')
        axes[0,1].axhline(y=-5, color='r', linestyle='--')
        axes[0,1].set_title('Prediction Bias')
        axes[0,1].set_ylabel('Bias (%)')
        axes[0,1].legend()
        
        # Error distribution
        axes[1,0].hist(df['mape'], bins=20, alpha=0.7)
        axes[1,0].axvline(x=15, color='r', linestyle='--', label='Target')
        axes[1,0].set_title('Error Distribution')
        axes[1,0].set_xlabel('MAPE (%)')
        axes[1,0].legend()
        
        # Accuracy by day of week
        df['day_of_week'] = pd.to_datetime(df['date']).dt.day_name()
        accuracy_by_day = df.groupby('day_of_week')['accuracy'].mean()
        axes[1,1].bar(accuracy_by_day.index, accuracy_by_day.values)
        axes[1,1].set_title('Accuracy by Day of Week')
        axes[1,1].set_ylabel('Accuracy (%)')
        axes[1,1].tick_params(axis='x', rotation=45)
        
        plt.tight_layout()
        plt.show()
        
        return fig

# Usage example
monitor = MLMonitoring()

# Simulate daily evaluation
for i in range(30):
    date = datetime.now() - timedelta(days=30-i)
    # Simulate predictions vs actual sales
    predictions = np.random.normal(100, 20, 50)
    actual_sales = predictions + np.random.normal(0, 15, 50)
    
    metrics = monitor.evaluate_daily_model(date, predictions, actual_sales)
    
report = monitor.generate_weekly_report()
print(f"Average accuracy last week: {report['average_accuracy']:.1f}%")