AI in Demand Forecasting: Transforming Consumer Goods and Distribution Supply Chains

import pandas as pd
import numpy as np
from sklearn.ensemble import RandomForestRegressor, GradientBoostingRegressor
from sklearn.neural_network import MLPRegressor
from sklearn.preprocessing import StandardScaler, LabelEncoder
from sklearn.model_selection import TimeSeriesSplit, GridSearchCV
from sklearn.metrics import mean_absolute_error, mean_squared_error
import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense, Dropout
import warnings
warnings.filterwarnings('ignore')

class AIdemandForecaster:
    def __init__(self, forecast_horizon=30):
        self.forecast_horizon = forecast_horizon
        self.models = {}
        self.scalers = {}
        self.feature_importance = {}
        self.is_trained = False
        
    def prepare_features(self, data, target_column='demand'):
        """Prepare comprehensive feature set for AI models"""
        df = data.copy()
        
        # Time-based features
        df['date'] = pd.to_datetime(df['date'])
        df['year'] = df['date'].dt.year
        df['month'] = df['date'].dt.month
        df['day'] = df['date'].dt.day
        df['day_of_week'] = df['date'].dt.dayofweek
        df['day_of_year'] = df['date'].dt.dayofyear
        df['week_of_year'] = df['date'].dt.isocalendar().week
        df['quarter'] = df['date'].dt.quarter
        df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)
        df['is_month_end'] = df['date'].dt.is_month_end.astype(int)
        df['is_month_start'] = df['date'].dt.is_month_start.astype(int)
        
        # Seasonal features
        df['season'] = df['month'].map({12: 'Winter', 1: 'Winter', 2: 'Winter',
                                        3: 'Spring', 4: 'Spring', 5: 'Spring',
                                        6: 'Summer', 7: 'Summer', 8: 'Summer',
                                        9: 'Fall', 10: 'Fall', 11: 'Fall'})
        
        # Lag features
        for lag in [1, 7, 14, 30, 90]:
            df[f'demand_lag_{lag}'] = df[target_column].shift(lag)
            
        # Rolling statistics
        for window in [7, 14, 30]:
            df[f'demand_mean_{window}d'] = df[target_column].rolling(window=window).mean()
            df[f'demand_std_{window}d'] = df[target_column].rolling(window=window).std()
            df[f'demand_min_{window}d'] = df[target_column].rolling(window=window).min()
            df[f'demand_max_{window}d'] = df[target_column].rolling(window=window).max()
            
        # Exponentially weighted moving averages
        df['demand_ewm_7'] = df[target_column].ewm(span=7).mean()
        df['demand_ewm_30'] = df[target_column].ewm(span=30).mean()
        
        # Trend features
        df['demand_trend_7d'] = df[target_column].diff(7)
        df['demand_trend_30d'] = df[target_column].diff(30)
        
        # Cyclical features
        df['month_sin'] = np.sin(2 * np.pi * df['month'] / 12)
        df['month_cos'] = np.cos(2 * np.pi * df['month'] / 12)
        df['day_sin'] = np.sin(2 * np.pi * df['day_of_week'] / 7)
        df['day_cos'] = np.cos(2 * np.pi * df['day_of_week'] / 7)
        
        # External factors (if available)
        if 'temperature' in df.columns:
            df['temp_lag_1'] = df['temperature'].shift(1)
            df['temp_mean_7d'] = df['temperature'].rolling(window=7).mean()
            
        if 'promotion' in df.columns:
            df['promo_lag_1'] = df['promotion'].shift(1)
            df['promo_impact'] = df['promotion'] * df[target_column].shift(1)
            
        if 'competitor_price' in df.columns:
            df['price_ratio'] = df['price'] / df['competitor_price']
            df['price_change'] = df['price'].pct_change()
            
        return df.dropna()
    
    def train_ensemble_models(self, prepared_data, target_column='demand'):
        """Train multiple AI models for ensemble forecasting"""
        # Separate features and target
        feature_columns = [col for col in prepared_data.columns 
                          if col not in [target_column, 'date'] and not col.startswith('Unnamed')]
        
        X = prepared_data[feature_columns]
        y = prepared_data[target_column]
        
        # Encode categorical features
        categorical_features = X.select_dtypes(include=['object']).columns
        for col in categorical_features:
            le = LabelEncoder()
            X[col] = le.fit_transform(X[col].astype(str))
        
        # Scale features for neural networks
        self.scalers['features'] = StandardScaler()
        X_scaled = self.scalers['features'].fit_transform(X)
        
        # Time series split for validation
        tscv = TimeSeriesSplit(n_splits=5)
        
        # Train Random Forest
        rf_params = {
            'n_estimators': [100, 200, 300],
            'max_depth': [10, 20, None],
            'min_samples_split': [2, 5, 10]
        }
        
        rf = RandomForestRegressor(random_state=42)
        rf_grid = GridSearchCV(rf, rf_params, cv=tscv, scoring='neg_mean_squared_error', n_jobs=-1)
        rf_grid.fit(X, y)
        self.models['random_forest'] = rf_grid.best_estimator_
        
        # Store feature importance
        self.feature_importance['random_forest'] = dict(zip(
            feature_columns, 
            self.models['random_forest'].feature_importances_
        ))
        
        # Train Gradient Boosting
        gb_params = {
            'n_estimators': [100, 200],
            'learning_rate': [0.05, 0.1, 0.2],
            'max_depth': [3, 5, 7]
        }
        
        gb = GradientBoostingRegressor(random_state=42)
        gb_grid = GridSearchCV(gb, gb_params, cv=tscv, scoring='neg_mean_squared_error', n_jobs=-1)
        gb_grid.fit(X, y)
        self.models['gradient_boosting'] = gb_grid.best_estimator_
        
        # Train Neural Network
        mlp = MLPRegressor(
            hidden_layer_sizes=(100, 50),
            activation='relu',
            solver='adam',
            alpha=0.001,
            learning_rate='adaptive',
            max_iter=1000,
            random_state=42
        )
        mlp.fit(X_scaled, y)
        self.models['neural_network'] = mlp
        
        # Train LSTM model
        self.models['lstm'] = self._train_lstm_model(X_scaled, y)
        
        self.feature_columns = feature_columns
        self.is_trained = True
        
        return self._evaluate_models(X, X_scaled, y)
    
    def _train_lstm_model(self, X_scaled, y, sequence_length=30):
        """Train LSTM model for time series forecasting"""
        # Prepare sequences for LSTM
        def create_sequences(data, target, seq_length):
            X_seq, y_seq = [], []
            for i in range(seq_length, len(data)):
                X_seq.append(data[i-seq_length:i])
                y_seq.append(target.iloc[i])
            return np.array(X_seq), np.array(y_seq)
        
        X_seq, y_seq = create_sequences(X_scaled, y, sequence_length)
        
        # Build LSTM model
        model = Sequential([
            LSTM(50, return_sequences=True, input_shape=(sequence_length, X_scaled.shape)),
            Dropout(0.2),
            LSTM(50, return_sequences=False),
            Dropout(0.2),
            Dense(25),
            Dense(1)
        ])
        
        model.compile(optimizer='adam', loss='mse')
        model.fit(X_seq, y_seq, epochs=50, batch_size=32, verbose=0)
        
        return model
    
    def _evaluate_models(self, X, X_scaled, y):
        """Evaluate all trained models"""
        results = {}
        
        for model_name, model in self.models.items():
            if model_name == 'lstm':
                continue  # LSTM evaluation requires sequence preparation
                
            if model_name == 'neural_network':
                y_pred = model.predict(X_scaled)
            else:
                y_pred = model.predict(X)
            
            mae = mean_absolute_error(y, y_pred)
            rmse = np.sqrt(mean_squared_error(y, y_pred))
            mape = np.mean(np.abs((y - y_pred) / y)) * 100
            
            results[model_name] = {
                'MAE': mae,
                'RMSE': rmse,
                'MAPE': mape
            }
        
        return results
    
    def predict_demand(self, input_data, ensemble_method='weighted_average'):
        """Generate demand forecasts using ensemble of AI models"""
        if not self.is_trained:
            raise ValueError("Models must be trained before making predictions")
        
        # Prepare input features
        X = input_data[self.feature_columns]
        
        # Encode categorical features
        categorical_features = X.select_dtypes(include=['object']).columns
        for col in categorical_features:
            # Handle unseen categories
            X[col] = X[col].astype(str)
        
        X_scaled = self.scalers['features'].transform(X)
        
        # Get predictions from each model
        predictions = {}
        
        predictions['random_forest'] = self.models['random_forest'].predict(X)
        predictions['gradient_boosting'] = self.models['gradient_boosting'].predict(X)
        predictions['neural_network'] = self.models['neural_network'].predict(X_scaled)
        
        # Ensemble predictions
        if ensemble_method == 'simple_average':
            final_prediction = np.mean(list(predictions.values()), axis=0)
        elif ensemble_method == 'weighted_average':
            # Weight by inverse RMSE (better models get higher weight)
            weights = {'random_forest': 0.4, 'gradient_boosting': 0.35, 'neural_network': 0.25}
            final_prediction = sum(weights[model] * pred for model, pred in predictions.items())
        else:
            final_prediction = predictions['random_forest']  # Default to best performing
        
        return {
            'ensemble_forecast': final_prediction,
            'individual_forecasts': predictions,
            'confidence_intervals': self._calculate_confidence_intervals(predictions)
        }
    
    def _calculate_confidence_intervals(self, predictions, confidence=0.95):
        """Calculate confidence intervals for ensemble predictions"""
        pred_array = np.array(list(predictions.values()))
        mean_pred = np.mean(pred_array, axis=0)
        std_pred = np.std(pred_array, axis=0)
        
        # Use t-distribution for small sample sizes
        from scipy import stats
        t_value = stats.t.ppf((1 + confidence) / 2, len(predictions) - 1)
        
        margin_of_error = t_value * std_pred / np.sqrt(len(predictions))
        
        return {
            'lower_bound': mean_pred - margin_of_error,
            'upper_bound': mean_pred + margin_of_error,
            'confidence_level': confidence
        }
    
    def generate_forecast_report(self, historical_data, forecast_period_days=30):
        """Generate comprehensive demand forecast report"""
        if not self.is_trained:
            raise ValueError("Models must be trained before generating reports")
        
        # Generate future dates
        last_date = pd.to_datetime(historical_data['date']).max()
        future_dates = pd.date_range(start=last_date + pd.Timedelta(days=1), 
                                   periods=forecast_period_days)
        
        # Create future feature matrix (simplified - in practice would need external data)
        future_data = pd.DataFrame({'date': future_dates})
        future_prepared = self.prepare_features(pd.concat([historical_data, future_data]))
        future_features = future_prepared.tail(forecast_period_days)
        
        # Generate forecasts
        forecast_results = self.predict_demand(future_features)
        
        # Create report
        report = {
            'forecast_period': {
                'start_date': future_dates[0].strftime('%Y-%m-%d'),
                'end_date': future_dates[-1].strftime('%Y-%m-%d'),
                'total_days': forecast_period_days
            },
            'forecast_summary': {
                'total_predicted_demand': float(np.sum(forecast_results['ensemble_forecast'])),
                'average_daily_demand': float(np.mean(forecast_results['ensemble_forecast'])),
                'peak_demand_day': future_dates[np.argmax(forecast_results['ensemble_forecast'])].strftime('%Y-%m-%d'),
                'peak_demand_value': float(np.max(forecast_results['ensemble_forecast']))
            },
            'model_performance': self.feature_importance,
            'daily_forecasts': [
                {
                    'date': date.strftime('%Y-%m-%d'),
                    'predicted_demand': float(demand),
                    'confidence_lower': float(lower),
                    'confidence_upper': float(upper)
                }
                for date, demand, lower, upper in zip(
                    future_dates,
                    forecast_results['ensemble_forecast'],
                    forecast_results['confidence_intervals']['lower_bound'],
                    forecast_results['confidence_intervals']['upper_bound']
                )
            ]
        }
        
        return report

# Example usage:
# forecaster = AIdemandForecaster()
# prepared_data = forecaster.prepare_features(historical_sales_data)
# model_results = forecaster.train_ensemble_models(prepared_data)
# forecast_report = forecaster.generate_forecast_report(historical_sales_data, 30)

class RealTimeDemandSensor {
  constructor() {
    this.dataStreams = new Map();
    this.models = new Map();
    this.alerts = [];
    this.updateInterval = 60000; // 1 minute
    this.thresholds = {
      demand_spike: 1.5,
      demand_drop: 0.7,
      forecast_deviation: 0.2
    };
  }

  // Initialize data stream connections
  initializeDataStreams() {
    const streamTypes = [
      'pos_sales',
      'ecommerce_traffic',
      'weather_data',
      'social_sentiment',
      'inventory_levels',
      'promotional_activity'
    ];

    streamTypes.forEach(streamType => {
      this.dataStreams.set(streamType, {
        status: 'active',
        lastUpdate: new Date(),
        dataBuffer: [],
        processingQueue: []
      });
    });

    this.startRealTimeProcessing();
  }

  // Process incoming real-time data
  async processIncomingData(streamType, data) {
    const stream = this.dataStreams.get(streamType);
    if (!stream) return;

    // Add to buffer with timestamp
    const dataPoint = {
      timestamp: new Date(),
      data: data,
      processed: false
    };

    stream.dataBuffer.push(dataPoint);
    stream.lastUpdate = new Date();

    // Trigger immediate processing for critical streams
    if (['pos_sales', 'inventory_levels'].includes(streamType)) {
      await this.processDataBuffer(streamType);
    }
  }

  // Process data buffer for a specific stream
  async processDataBuffer(streamType) {
    const stream = this.dataStreams.get(streamType);
    const unprocessedData = stream.dataBuffer.filter(d => !d.processed);

    if (unprocessedData.length === 0) return;

    // Aggregate data for analysis
    const aggregatedData = this.aggregateStreamData(unprocessedData, streamType);
    
    // Apply real-time model
    const prediction = await this.applyRealTimeModel(streamType, aggregatedData);
    
    // Check for significant changes
    await this.detectAnomalies(streamType, prediction, aggregatedData);
    
    // Mark data as processed
    unprocessedData.forEach(d => d.processed = true);
    
    // Trigger forecast update if needed
    if (this.shouldUpdateForecast(streamType, prediction)) {
      await this.updateDemandForecast(streamType, prediction);
    }
  }

  aggregateStreamData(dataPoints, streamType) {
    const now = new Date();
    const oneHourAgo = new Date(now.getTime() - 60 * 60 * 1000);
    
    // Filter recent data
    const recentData = dataPoints.filter(d => d.timestamp >= oneHourAgo);
    
    switch (streamType) {
      case 'pos_sales':
        return {
          totalSales: recentData.reduce((sum, d) => sum + (d.data.amount || 0), 0),
          transactionCount: recentData.length,
          averageTransaction: recentData.length > 0 ? 
            recentData.reduce((sum, d) => sum + (d.data.amount || 0), 0) / recentData.length : 0,
          topProducts: this.getTopProducts(recentData)
        };
        
      case 'ecommerce_traffic':
        return {
          pageViews: recentData.reduce((sum, d) => sum + (d.data.views || 0), 0),
          uniqueVisitors: new Set(recentData.map(d => d.data.userId)).size,
          conversionRate: this.calculateConversionRate(recentData),
          cartAdditions: recentData.filter(d => d.data.action === 'add_to_cart').length
        };
        
      case 'social_sentiment':
        const sentiments = recentData.map(d => d.data.sentiment);
        return {
          averageSentiment: sentiments.reduce((sum, s) => sum + s, 0) / sentiments.length,
          mentionCount: recentData.length,
          trendingTopics: this.extractTrendingTopics(recentData),
          sentimentTrend: this.calculateSentimentTrend(recentData)
        };
        
      case 'weather_data':
        return {
          currentTemp: recentData[recentData.length - 1]?.data.temperature,
          precipitation: recentData.some(d => d.data.precipitation > 0),
          forecast: this.extractWeatherForecast(recentData),
          seasonalFactor: this.calculateSeasonalFactor(recentData)
        };
        
      default:
        return { count: recentData.length, latestData: recentData[recentData.length - 1] };
    }
  }

  async applyRealTimeModel(streamType, aggregatedData) {
    // Simulate ML model application
    const baselineForecast = await this.getBaselineForecast(streamType);
    
    switch (streamType) {
      case 'pos_sales':
        // Adjust forecast based on current sales velocity
        const salesVelocity = aggregatedData.totalSales / aggregatedData.transactionCount;
        const velocityFactor = salesVelocity > baselineForecast.averageTransaction ? 1.1 : 0.95;
        return {
          adjustedForecast: baselineForecast.value * velocityFactor,
          confidence: 0.85,
          factors: { salesVelocity, transactionCount: aggregatedData.transactionCount }
        };
        
      case 'social_sentiment':
        // Adjust forecast based on sentiment
        const sentimentMultiplier = Math.max(0.7, Math.min(1.3, 1 + (aggregatedData.averageSentiment - 0.5)));
        return {
          adjustedForecast: baselineForecast.value * sentimentMultiplier,
          confidence: 0.72,
          factors: { sentiment: aggregatedData.averageSentiment, mentions: aggregatedData.mentionCount }
        };
        
      case 'weather_data':
        // Weather-based adjustments
        const weatherImpact = this.calculateWeatherImpact(aggregatedData);
        return {
          adjustedForecast: baselineForecast.value * weatherImpact,
          confidence: 0.88,
          factors: { weatherImpact, temperature: aggregatedData.currentTemp }
        };
        
      default:
        return {
          adjustedForecast: baselineForecast.value,
          confidence: 0.75,
          factors: {}
        };
    }
  }

  async detectAnomalies(streamType, prediction, aggregatedData) {
    const baseline = await this.getBaselineForecast(streamType);
    const deviation = Math.abs(prediction.adjustedForecast - baseline.value) / baseline.value;
    
    if (deviation > this.thresholds.forecast_deviation) {
      const alert = {
        timestamp: new Date(),
        type: 'FORECAST_ANOMALY',
        streamType: streamType,
        severity: deviation > 0.5 ? 'HIGH' : 'MEDIUM',
        details: {
          predicted: prediction.adjustedForecast,
          baseline: baseline.value,
          deviation: deviation,
          confidence: prediction.confidence,
          factors: prediction.factors
        }
      };
      
      this.alerts.push(alert);
      await this.notifyStakeholders(alert);
    }

    // Check for demand spikes or drops
    if (streamType === 'pos_sales') {
      const currentRate = aggregatedData.totalSales / (aggregatedData.transactionCount || 1);
      const expectedRate = baseline.averageTransaction || currentRate;
      
      if (currentRate > expectedRate * this.thresholds.demand_spike) {
        await this.triggerInventoryAlert('DEMAND_SPIKE', streamType, {
          currentRate,
          expectedRate,
          multiplier: currentRate / expectedRate
        });
      } else if (currentRate < expectedRate * this.thresholds.demand_drop) {
        await this.triggerInventoryAlert('DEMAND_DROP', streamType, {
          currentRate,
          expectedRate,
          multiplier: currentRate / expectedRate
        });
      }
    }
  }

  async updateDemandForecast(streamType, prediction) {
    const forecastUpdate = {
      timestamp: new Date(),
      source: streamType,
      newForecast: prediction.adjustedForecast,
      confidence: prediction.confidence,
      adjustmentFactors: prediction.factors
    };
    
    // Update central forecasting system
    await this.publishForecastUpdate(forecastUpdate);
    
    console.log(`Forecast updated based on ${streamType}:`, forecastUpdate);
  }

  shouldUpdateForecast(streamType, prediction) {
    // Update forecast if confidence is high and deviation is significant
    return prediction.confidence > 0.8 && 
           Math.abs(prediction.adjustedForecast) > this.thresholds.forecast_deviation;
  }

  startRealTimeProcessing() {
    setInterval(async () => {
      for (const [streamType, stream] of this.dataStreams) {
        if (stream.dataBuffer.some(d => !d.processed)) {
          await this.processDataBuffer(streamType);
        }
      }
      
      // Clean up old data
      this.cleanupOldData();
    }, this.updateInterval);
  }

  cleanupOldData() {
    const cutoffTime = new Date(Date.now() - 24 * 60 * 60 * 1000); // 24 hours ago
    
    for (const [streamType, stream] of this.dataStreams) {
      stream.dataBuffer = stream.dataBuffer.filter(d => d.timestamp > cutoffTime);
    }
    
    this.alerts = this.alerts.filter(a => a.timestamp > cutoffTime);
  }

  // Utility methods
  getTopProducts(salesData) {
    const productCounts = {};
    salesData.forEach(d => {
      if (d.data.productId) {
        productCounts[d.data.productId] = (productCounts[d.data.productId] || 0) + 1;
      }
    });
    
    return Object.entries(productCounts)
      .sort(([,a], [,b]) => b - a)
      .slice(0, 5)
      .map(([productId, count]) => ({ productId, count }));
  }

  calculateConversionRate(trafficData) {
    const visitors = trafficData.filter(d => d.data.action === 'visit').length;
    const purchases = trafficData.filter(d => d.data.action === 'purchase').length;
    return visitors > 0 ? purchases / visitors : 0;
  }

  calculateWeatherImpact(weatherData) {
    // Simplified weather impact calculation
    const temp = weatherData.currentTemp || 20;
    const hasRain = weatherData.precipitation;
    
    let impact = 1.0;
    
    // Temperature adjustments (seasonal products)
    if (temp > 25) impact *= 1.1; // Hot weather boosts cold drinks
    if (temp < 5) impact *= 1.05;  // Cold weather boosts warm products
    if (hasRain) impact *= 0.95;   // Rain reduces foot traffic
    
    return Math.max(0.8, Math.min(1.3, impact));
  }

  async getBaselineForecast(streamType) {
    // Simulate baseline forecast retrieval
    return {
      value: 100 + Math.random() * 50,
      averageTransaction: 25 + Math.random() * 10
    };
  }

  async notifyStakeholders(alert) {
    console.log('ALERT:', alert.type, alert.severity, alert.details);
    // In production, send to monitoring systems, emails, etc.
  }

  async triggerInventoryAlert(type, source, data) {
    console.log(`INVENTORY ALERT: ${type} detected in ${source}`, data);
    // In production, trigger inventory management systems
  }

  async publishForecastUpdate(update) {
    console.log('Publishing forecast update:', update);
    // In production, update central forecasting database
  }

  getDashboard() {
    const now = new Date();
    return {
      timestamp: now,
      activeStreams: this.dataStreams.size,
      recentAlerts: this.alerts.filter(a => 
        (now - a.timestamp) < 3600000 // Last hour
      ).length,
      streamStatus: Array.from(this.dataStreams.entries()).map(([name, stream]) => ({
        name,
        status: stream.status,
        lastUpdate: stream.lastUpdate,
        bufferSize: stream.dataBuffer.length,
        unprocessedCount: stream.dataBuffer.filter(d => !d.processed).length
      }))
    };
  }
}

// Example usage:
// const demandSensor = new RealTimeDemandSensor();
// demandSensor.initializeDataStreams();
// 
// // Simulate incoming data
// setInterval(() => {
//   demandSensor.processIncomingData('pos_sales', {
//     amount: Math.random() * 100,
//     productId: 'PROD_' + Math.floor(Math.random() * 10)
//   });
// }, 5000);