Advanced Demand Forecasting System

System Architecture

A comprehensive demand forecasting system that combines multiple forecasting methods with hierarchical reconciliation and automated model selection.

Core Components

1. Data Pipeline

class ForecastDataPipeline:
    def __init__(self, config: Dict[str, Any]):
        self.feature_generator = FeatureGenerator(
            config['feature_config']
        )
        self.preprocessor = TimeSeriesPreprocessor(
            config['preprocessing_config']
        )
        self.validator = DataValidator(config['validation_config'])
        
    async def prepare_forecast_data(
        self,
        raw_data: pd.DataFrame
    ) -> ForecastData:
        # Validate input data
        await self.validator.validate_data(raw_data)
        
        # Generate features
        features = await self.feature_generator.generate_features(
            raw_data
        )
        
        # Preprocess time series
        processed_data = await self.preprocessor.process(
            raw_data,
            features
        )
        
        return ForecastData(
            time_series=processed_data['series'],
            features=processed_data['features'],
            metadata=processed_data['metadata']
        )

class FeatureGenerator:
    def __init__(self, config: Dict[str, Any]):
        self.calendar_processor = CalendarProcessor(
            config['calendar_config']
        )
        self.event_processor = EventProcessor(config['event_config'])
        self.lag_generator = LagFeatureGenerator(config['lag_config'])
        
    async def generate_features(
        self,
        data: pd.DataFrame
    ) -> pd.DataFrame:
        features = {}
        
        # Generate different types of features in parallel
        feature_tasks = [
            self.calendar_processor.process(data),
            self.event_processor.process(data),
            self.lag_generator.generate(data)
        ]
        
        results = await asyncio.gather(*feature_tasks)
        
        # Combine all features
        for result in results:
            features.update(result)
            
        return pd.DataFrame(features)

2. Forecasting Models

class ForecastingEngine:
    def __init__(self, config: Dict[str, Any]):
        self.models = {
            'deepar': DeepARModel(config['deepar_config']),
            'prophet': ProphetModel(config['prophet_config']),
            'lstm': HierarchicalLSTM(config['lstm_config']),
            'statistical': StatisticalModels(config['stat_config'])
        }
        self.model_selector = ModelSelector(config['selection_config'])
        self.reconciler = HierarchicalReconciliation(
            config['reconciliation_config']
        )
        
    async def generate_forecast(
        self,
        data: ForecastData,
        horizon: int
    ) -> ForecastResult:
        # Select best model for each series
        model_assignments = await self.model_selector.select_models(
            data,
            self.models
        )
        
        # Generate forecasts using assigned models
        forecasts = {}
        for series_id, model_name in model_assignments.items():
            model = self.models[model_name]
            forecasts[series_id] = await model.forecast(
                data.get_series(series_id),
                horizon
            )
            
        # Reconcile hierarchical forecasts
        reconciled_forecasts = await self.reconciler.reconcile(
            forecasts,
            data.hierarchy
        )
        
        return ForecastResult(
            point_forecasts=reconciled_forecasts['point'],
            intervals=reconciled_forecasts['intervals'],
            metrics=self._compute_metrics(reconciled_forecasts)
        )

class DeepARModel:
    def __init__(self, config: Dict[str, Any]):
        self.network = DeepARNetwork(
            input_size=config['input_size'],
            hidden_size=config['hidden_size'],
            num_layers=config['num_layers']
        )
        self.trainer = LightningTrainer(config['trainer_config'])
        
    async def forecast(
        self,
        series: pd.Series,
        horizon: int
    ) -> Dict[str, np.ndarray]:
        # Prepare data for DeepAR
        dataset = self._prepare_data(series)
        
        # Generate samples from the model
        samples = await self._generate_samples(
            dataset,
            horizon,
            num_samples=100
        )
        
        return {
            'mean': np.mean(samples, axis=0),
            'lower': np.percentile(samples, 10, axis=0),
            'upper': np.percentile(samples, 90, axis=0)
        }

3. Uncertainty Quantification

class UncertaintyEstimator:
    def __init__(self, config: Dict[str, Any]):
        self.distribution_fitter = DistributionFitter(
            config['distribution_config']
        )
        self.interval_calculator = PredictionIntervalCalculator(
            config['interval_config']
        )
        
    async def estimate_uncertainty(
        self,
        forecasts: np.ndarray,
        history: pd.Series
    ) -> Dict[str, np.ndarray]:
        # Fit error distribution
        error_dist = await self.distribution_fitter.fit(
            history,
            forecasts
        )
        
        # Calculate prediction intervals
        intervals = await self.interval_calculator.calculate(
            forecasts,
            error_dist,
            confidence_levels=[0.5, 0.8, 0.95]
        )
        
        return {
            'distributions': error_dist,
            'intervals': intervals
        }

class DistributionFitter:
    def __init__(self, config: Dict[str, Any]):
        self.distributions = [
            'normal',
            'student_t',
            'skewed_t'
        ]
        self.selection_criterion = config['criterion']
        
    async def fit(
        self,
        history: pd.Series,
        forecasts: np.ndarray
    ) -> Distribution:
        # Compute forecast errors
        errors = self._compute_errors(history, forecasts)
        
        # Fit multiple distributions
        fitted_dists = await asyncio.gather(*[
            self._fit_distribution(errors, dist_name)
            for dist_name in self.distributions
        ])
        
        # Select best fitting distribution
        return self._select_best_distribution(fitted_dists)

4. Forecast Monitoring

class ForecastMonitor:
    def __init__(self, config: Dict[str, Any]):
        self.metrics_calculator = ForecastMetrics(
            config['metrics_config']
        )
        self.anomaly_detector = ForecastAnomalyDetector(
            config['anomaly_config']
        )
        self.alert_manager = AlertManager(config['alert_config'])
        
    async def monitor_forecasts(
        self,
        forecasts: ForecastResult,
        actuals: pd.Series
    ) -> MonitoringReport:
        # Calculate forecast accuracy metrics
        metrics = await self.metrics_calculator.calculate(
            forecasts,
            actuals
        )
        
        # Detect forecast anomalies
        anomalies = await self.anomaly_detector.detect(
            forecasts,
            actuals
        )
        
        if anomalies:
            await self.alert_manager.send_alerts(anomalies)
            
        return MonitoringReport(
            metrics=metrics,
            anomalies=anomalies
        )

Usage Example

# Initialize forecasting system
config = {
    'data_config': {
        'feature_config': {
            'calendar_config': {'holidays': True},
            'lag_config': {'max_lags': 30}
        }
    },
    'model_config': {
        'deepar_config': {
            'hidden_size': 128,
            'num_layers': 3
        },
        'reconciliation_config': {
            'method': 'mint',
            'weights': 'ols'
        }
    }
}

forecasting_system = DemandForecastingSystem(config)

# Generate forecasts
forecast_data = await forecasting_system.prepare_data(historical_data)
forecasts = await forecasting_system.generate_forecast(
    forecast_data,
    horizon=30
)

# Monitor forecast accuracy
monitoring_report = await forecasting_system.monitor_forecasts(
    forecasts,
    actual_values
)

View Source Code | Documentation | Contributing Guidelines