Prophet + XGBoost Ensemble with Real-time Data Pipeline

Solution for the use case Inventory Demand Forecasting with Dynamic Pricing.

AI tools

Facebook Prophet, XGBoost, Apache Kafka, Redis, Python, Scikit-learn

Language

Python

Accuracy

82%

Published

2025-07-05

Code (Python)

import pandas as pd
import numpy as np
from prophet import Prophet
import xgboost as xgb
from sklearn.ensemble import VotingRegressor
from kafka import KafkaConsumer
import redis
import requests

class DemandForecastingSystem:
    def __init__(self):
        self.prophet_model = Prophet(
            yearly_seasonality=True,
            weekly_seasonality=True,
            daily_seasonality=True,
            changepoint_prior_scale=0.05
        )
        self.xgb_model = xgb.XGBRegressor(
            n_estimators=100,
            max_depth=6,
            learning_rate=0.1
        )
        self.redis_client = redis.Redis(host="localhost", port=6379)
        
    def prepare_features(self, sales_data, external_data):
        """Prepare features for demand forecasting"""
        features = sales_data.copy()
        
        # Time-based features
        features["day_of_week"] = features["date"].dt.dayofweek
        features["month"] = features["date"].dt.month
        features["quarter"] = features["date"].dt.quarter
        features["is_weekend"] = features["day_of_week"].isin([5, 6])
        
        # Weather features
        features = features.merge(external_data["weather"], on="date", how="left")
        features["temp_category"] = pd.cut(features["temperature"], 
                                         bins=[-np.inf, 0, 15, 25, np.inf], 
                                         labels=["cold", "cool", "warm", "hot"])
        
        # Holiday and event features
        features = features.merge(external_data["events"], on="date", how="left")
        features["is_holiday"] = features["event_type"] == "holiday"
        features["is_local_event"] = features["event_type"] == "local"
        
        # Competitor pricing features
        features = features.merge(external_data["competitor_prices"], on="date", how="left")
        features["price_vs_competitor"] = features["our_price"] / features["competitor_avg_price"]
        
        return features
    
    def train_ensemble_model(self, training_data):
        """Train Prophet + XGBoost ensemble"""
        
        # Prepare Prophet data
        prophet_data = training_data[["date", "sales"]].rename(columns={"date": "ds", "sales": "y"})
        
        # Add external regressors to Prophet
        prophet_data["temperature"] = training_data["temperature"]
        prophet_data["is_holiday"] = training_data["is_holiday"].astype(int)
        prophet_data["competitor_price"] = training_data["competitor_avg_price"]
        
        self.prophet_model.add_regressor("temperature")
        self.prophet_model.add_regressor("is_holiday")
        self.prophet_model.add_regressor("competitor_price")
        
        # Train Prophet
        self.prophet_model.fit(prophet_data)
        
        # Prepare XGBoost features
        feature_cols = [col for col in training_data.columns 
                       if col not in ["date", "sales", "sku", "store_id"]]
        X = training_data[feature_cols]
        y = training_data["sales"]
        
        # Train XGBoost
        self.xgb_model.fit(X, y)
        
    def forecast_demand(self, sku, store_id, forecast_horizon=7):
        """Generate demand forecast for specific SKU and store"""
        
        # Get latest data
        latest_data = self.get_latest_data(sku, store_id)
        
        # Prophet forecast
        future_prophet = self.create_prophet_future(latest_data, forecast_horizon)
        prophet_forecast = self.prophet_model.predict(future_prophet)
        
        # XGBoost forecast  
        future_features = self.prepare_future_features(latest_data, forecast_horizon)
        xgb_forecast = self.xgb_model.predict(future_features)
        
        # Ensemble prediction (weighted average)
        ensemble_forecast = 0.6 * prophet_forecast["yhat"].values + 0.4 * xgb_forecast
        
        return {
            "sku": sku,
            "store_id": store_id,
            "forecast": ensemble_forecast.tolist(),
            "proph
…

Prompts

FORECAST_PROMPT = """
Generate demand forecast analysis for the following product and location:

Product: {product_name} (SKU: {sku})
Location: {store_location}
Historical Data: {sales_history}
External Factors: {weather_data}, {events_data}, {competitor_data}

Consider these factors:
1. Seasonal patterns and trends
2. Weather impact on demand
3. Local events and holidays
4. Competitor pricing changes
5. Promotional calendar

Provide forecast with confidence intervals and key drivers:
{
  "7_day_forecast": [daily_predictions],
  "confidence_intervals": {"lower": [], "upper": []},
  "key_drivers": ["weather", "event", "trend"],
  "recommended_actions": ["restock", "adjust_price", "promotion"]
}
"""

PRICING_PROMPT = """
Based on demand forecast and business constraints, recommend optimal pricing:

Current Price: ${current_price}
Demand Forecast: {forecast_data}
Inventory Level: {current_stock}
Cost: ${unit_cost}
Competitor Prices: {competitor_pricing}

Optimize for:
- Revenue maximization
- Inventory turnover
- Market positioning
- Customer perception

Return pricing recommendation:
{
  "recommended_price": 29.99,
  "price_change": "+5.2%",
  "expected_impact": {
    "demand_change": "-8%",
    "revenue_change": "+12%"
  },
  "rationale": "Demand spike predicted due to weather event"
}
"""