386 lines
12 KiB
Python
386 lines
12 KiB
Python
"""
|
|
Scenario Planning System
|
|
What-if analysis for demand forecasting
|
|
"""
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
from typing import Dict, List, Any, Optional
|
|
from datetime import datetime, date, timedelta
|
|
import structlog
|
|
from enum import Enum
|
|
|
|
logger = structlog.get_logger()
|
|
|
|
|
|
class ScenarioType(str, Enum):
|
|
"""Types of scenarios"""
|
|
BASELINE = "baseline"
|
|
OPTIMISTIC = "optimistic"
|
|
PESSIMISTIC = "pessimistic"
|
|
CUSTOM = "custom"
|
|
PROMOTION = "promotion"
|
|
EVENT = "event"
|
|
WEATHER = "weather"
|
|
PRICE_CHANGE = "price_change"
|
|
|
|
|
|
class ScenarioPlanner:
|
|
"""
|
|
Scenario planning for demand forecasting.
|
|
|
|
Allows testing "what-if" scenarios:
|
|
- What if we run a promotion?
|
|
- What if there's a local festival?
|
|
- What if weather is unusually bad?
|
|
- What if we change prices?
|
|
"""
|
|
|
|
def __init__(self, base_forecaster=None):
|
|
"""
|
|
Initialize scenario planner.
|
|
|
|
Args:
|
|
base_forecaster: Base forecaster to use for baseline predictions
|
|
"""
|
|
self.base_forecaster = base_forecaster
|
|
|
|
async def create_scenario(
|
|
self,
|
|
tenant_id: str,
|
|
inventory_product_id: str,
|
|
scenario_name: str,
|
|
scenario_type: ScenarioType,
|
|
start_date: date,
|
|
end_date: date,
|
|
adjustments: Dict[str, Any]
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Create a forecast scenario with adjustments.
|
|
|
|
Args:
|
|
tenant_id: Tenant identifier
|
|
inventory_product_id: Product identifier
|
|
scenario_name: Name for the scenario
|
|
scenario_type: Type of scenario
|
|
start_date: Scenario start date
|
|
end_date: Scenario end date
|
|
adjustments: Dictionary of adjustments to apply
|
|
|
|
Returns:
|
|
Scenario forecast results
|
|
"""
|
|
logger.info(
|
|
"Creating forecast scenario",
|
|
tenant_id=tenant_id,
|
|
inventory_product_id=inventory_product_id,
|
|
scenario_name=scenario_name,
|
|
scenario_type=scenario_type
|
|
)
|
|
|
|
# Generate baseline forecast first
|
|
baseline_forecast = await self._generate_baseline_forecast(
|
|
tenant_id=tenant_id,
|
|
inventory_product_id=inventory_product_id,
|
|
start_date=start_date,
|
|
end_date=end_date
|
|
)
|
|
|
|
# Apply scenario adjustments
|
|
scenario_forecast = self._apply_scenario_adjustments(
|
|
baseline_forecast=baseline_forecast,
|
|
adjustments=adjustments,
|
|
scenario_type=scenario_type
|
|
)
|
|
|
|
# Calculate impact
|
|
impact_analysis = self._calculate_scenario_impact(
|
|
baseline_forecast=baseline_forecast,
|
|
scenario_forecast=scenario_forecast
|
|
)
|
|
|
|
return {
|
|
'scenario_id': f"scenario_{tenant_id}_{inventory_product_id}_{datetime.now().strftime('%Y%m%d%H%M%S')}",
|
|
'scenario_name': scenario_name,
|
|
'scenario_type': scenario_type,
|
|
'tenant_id': tenant_id,
|
|
'inventory_product_id': inventory_product_id,
|
|
'date_range': {
|
|
'start': start_date.isoformat(),
|
|
'end': end_date.isoformat()
|
|
},
|
|
'baseline_forecast': baseline_forecast,
|
|
'scenario_forecast': scenario_forecast,
|
|
'impact_analysis': impact_analysis,
|
|
'adjustments_applied': adjustments,
|
|
'created_at': datetime.now().isoformat()
|
|
}
|
|
|
|
async def compare_scenarios(
|
|
self,
|
|
scenarios: List[Dict[str, Any]]
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Compare multiple scenarios side-by-side.
|
|
|
|
Args:
|
|
scenarios: List of scenario results from create_scenario()
|
|
|
|
Returns:
|
|
Comparison analysis
|
|
"""
|
|
if len(scenarios) < 2:
|
|
return {'error': 'Need at least 2 scenarios to compare'}
|
|
|
|
comparison = {
|
|
'scenarios_compared': len(scenarios),
|
|
'scenario_names': [s['scenario_name'] for s in scenarios],
|
|
'comparison_metrics': {}
|
|
}
|
|
|
|
# Extract total demand for each scenario
|
|
for scenario in scenarios:
|
|
scenario_name = scenario['scenario_name']
|
|
scenario_forecast = scenario['scenario_forecast']
|
|
|
|
total_demand = sum(f['predicted_demand'] for f in scenario_forecast)
|
|
|
|
comparison['comparison_metrics'][scenario_name] = {
|
|
'total_demand': total_demand,
|
|
'avg_daily_demand': total_demand / len(scenario_forecast) if scenario_forecast else 0,
|
|
'peak_demand': max(f['predicted_demand'] for f in scenario_forecast) if scenario_forecast else 0
|
|
}
|
|
|
|
# Determine best and worst scenarios
|
|
total_demands = {
|
|
name: metrics['total_demand']
|
|
for name, metrics in comparison['comparison_metrics'].items()
|
|
}
|
|
|
|
comparison['best_scenario'] = max(total_demands, key=total_demands.get)
|
|
comparison['worst_scenario'] = min(total_demands, key=total_demands.get)
|
|
|
|
comparison['demand_range'] = {
|
|
'min': min(total_demands.values()),
|
|
'max': max(total_demands.values()),
|
|
'spread': max(total_demands.values()) - min(total_demands.values())
|
|
}
|
|
|
|
return comparison
|
|
|
|
async def _generate_baseline_forecast(
|
|
self,
|
|
tenant_id: str,
|
|
inventory_product_id: str,
|
|
start_date: date,
|
|
end_date: date
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Generate baseline forecast without adjustments.
|
|
|
|
Args:
|
|
tenant_id: Tenant identifier
|
|
inventory_product_id: Product identifier
|
|
start_date: Start date
|
|
end_date: End date
|
|
|
|
Returns:
|
|
List of daily forecasts
|
|
"""
|
|
# Generate date range
|
|
dates = []
|
|
current_date = start_date
|
|
while current_date <= end_date:
|
|
dates.append(current_date)
|
|
current_date += timedelta(days=1)
|
|
|
|
# Placeholder forecast (in real implementation, call forecasting service)
|
|
baseline = []
|
|
for forecast_date in dates:
|
|
baseline.append({
|
|
'date': forecast_date.isoformat(),
|
|
'predicted_demand': 100, # Placeholder
|
|
'confidence_lower': 80,
|
|
'confidence_upper': 120
|
|
})
|
|
|
|
return baseline
|
|
|
|
def _apply_scenario_adjustments(
|
|
self,
|
|
baseline_forecast: List[Dict[str, Any]],
|
|
adjustments: Dict[str, Any],
|
|
scenario_type: ScenarioType
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Apply adjustments to baseline forecast.
|
|
|
|
Args:
|
|
baseline_forecast: Baseline forecast data
|
|
adjustments: Adjustments to apply
|
|
scenario_type: Type of scenario
|
|
|
|
Returns:
|
|
Adjusted forecast
|
|
"""
|
|
scenario_forecast = []
|
|
|
|
for day_forecast in baseline_forecast:
|
|
adjusted_forecast = day_forecast.copy()
|
|
|
|
# Apply different adjustment types
|
|
if 'demand_multiplier' in adjustments:
|
|
# Multiply demand by factor
|
|
multiplier = adjustments['demand_multiplier']
|
|
adjusted_forecast['predicted_demand'] *= multiplier
|
|
adjusted_forecast['confidence_lower'] *= multiplier
|
|
adjusted_forecast['confidence_upper'] *= multiplier
|
|
|
|
if 'demand_offset' in adjustments:
|
|
# Add/subtract fixed amount
|
|
offset = adjustments['demand_offset']
|
|
adjusted_forecast['predicted_demand'] += offset
|
|
adjusted_forecast['confidence_lower'] += offset
|
|
adjusted_forecast['confidence_upper'] += offset
|
|
|
|
if 'event_impact' in adjustments:
|
|
# Apply event-specific impact
|
|
event_multiplier = adjustments['event_impact']
|
|
adjusted_forecast['predicted_demand'] *= event_multiplier
|
|
|
|
if 'weather_impact' in adjustments:
|
|
# Apply weather adjustments
|
|
weather_factor = adjustments['weather_impact']
|
|
adjusted_forecast['predicted_demand'] *= weather_factor
|
|
|
|
if 'price_elasticity' in adjustments and 'price_change_percent' in adjustments:
|
|
# Apply price elasticity
|
|
elasticity = adjustments['price_elasticity']
|
|
price_change = adjustments['price_change_percent']
|
|
demand_change = -elasticity * price_change # Negative correlation
|
|
adjusted_forecast['predicted_demand'] *= (1 + demand_change)
|
|
|
|
# Ensure non-negative demand
|
|
adjusted_forecast['predicted_demand'] = max(0, adjusted_forecast['predicted_demand'])
|
|
adjusted_forecast['confidence_lower'] = max(0, adjusted_forecast['confidence_lower'])
|
|
|
|
scenario_forecast.append(adjusted_forecast)
|
|
|
|
return scenario_forecast
|
|
|
|
def _calculate_scenario_impact(
|
|
self,
|
|
baseline_forecast: List[Dict[str, Any]],
|
|
scenario_forecast: List[Dict[str, Any]]
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Calculate impact of scenario vs baseline.
|
|
|
|
Args:
|
|
baseline_forecast: Baseline forecast
|
|
scenario_forecast: Scenario forecast
|
|
|
|
Returns:
|
|
Impact analysis
|
|
"""
|
|
baseline_total = sum(f['predicted_demand'] for f in baseline_forecast)
|
|
scenario_total = sum(f['predicted_demand'] for f in scenario_forecast)
|
|
|
|
difference = scenario_total - baseline_total
|
|
percent_change = (difference / baseline_total * 100) if baseline_total > 0 else 0
|
|
|
|
return {
|
|
'baseline_total_demand': baseline_total,
|
|
'scenario_total_demand': scenario_total,
|
|
'absolute_difference': difference,
|
|
'percent_change': percent_change,
|
|
'impact_category': self._categorize_impact(percent_change),
|
|
'days_analyzed': len(baseline_forecast)
|
|
}
|
|
|
|
def _categorize_impact(self, percent_change: float) -> str:
|
|
"""Categorize impact magnitude"""
|
|
if abs(percent_change) < 5:
|
|
return "minimal"
|
|
elif abs(percent_change) < 15:
|
|
return "moderate"
|
|
elif abs(percent_change) < 30:
|
|
return "significant"
|
|
else:
|
|
return "major"
|
|
|
|
def generate_predefined_scenarios(
|
|
self,
|
|
base_scenario: Dict[str, Any]
|
|
) -> List[Dict[str, Any]]:
|
|
"""
|
|
Generate common predefined scenarios for comparison.
|
|
|
|
Args:
|
|
base_scenario: Base scenario parameters
|
|
|
|
Returns:
|
|
List of scenario configurations
|
|
"""
|
|
scenarios = []
|
|
|
|
# Baseline scenario
|
|
scenarios.append({
|
|
'scenario_name': 'Baseline',
|
|
'scenario_type': ScenarioType.BASELINE,
|
|
'adjustments': {}
|
|
})
|
|
|
|
# Optimistic scenario
|
|
scenarios.append({
|
|
'scenario_name': 'Optimistic',
|
|
'scenario_type': ScenarioType.OPTIMISTIC,
|
|
'adjustments': {
|
|
'demand_multiplier': 1.2, # 20% increase
|
|
'description': '+20% demand increase'
|
|
}
|
|
})
|
|
|
|
# Pessimistic scenario
|
|
scenarios.append({
|
|
'scenario_name': 'Pessimistic',
|
|
'scenario_type': ScenarioType.PESSIMISTIC,
|
|
'adjustments': {
|
|
'demand_multiplier': 0.8, # 20% decrease
|
|
'description': '-20% demand decrease'
|
|
}
|
|
})
|
|
|
|
# Promotion scenario
|
|
scenarios.append({
|
|
'scenario_name': 'Promotion Campaign',
|
|
'scenario_type': ScenarioType.PROMOTION,
|
|
'adjustments': {
|
|
'demand_multiplier': 1.5, # 50% increase
|
|
'description': '50% promotion boost'
|
|
}
|
|
})
|
|
|
|
# Bad weather scenario
|
|
scenarios.append({
|
|
'scenario_name': 'Bad Weather',
|
|
'scenario_type': ScenarioType.WEATHER,
|
|
'adjustments': {
|
|
'weather_impact': 0.7, # 30% decrease
|
|
'description': 'Bad weather reduces foot traffic'
|
|
}
|
|
})
|
|
|
|
# Price increase scenario
|
|
scenarios.append({
|
|
'scenario_name': 'Price Increase 10%',
|
|
'scenario_type': ScenarioType.PRICE_CHANGE,
|
|
'adjustments': {
|
|
'price_elasticity': 1.2, # Elastic demand
|
|
'price_change_percent': 0.10, # 10% price increase
|
|
'description': '10% price increase with elastic demand'
|
|
}
|
|
})
|
|
|
|
return scenarios
|