273 lines
10 KiB
Python
273 lines
10 KiB
Python
# ================================================================
|
|
# services/forecasting/app/schemas/forecasts.py
|
|
# ================================================================
|
|
"""
|
|
Forecast schemas for request/response validation
|
|
"""
|
|
|
|
from pydantic import BaseModel, Field, validator
|
|
from datetime import datetime, date
|
|
from typing import Optional, List, Dict, Any
|
|
from enum import Enum
|
|
|
|
class BusinessType(str, Enum):
|
|
INDIVIDUAL = "individual"
|
|
CENTRAL_WORKSHOP = "central_workshop"
|
|
|
|
|
|
class ForecastRequest(BaseModel):
|
|
"""Request schema for generating forecasts"""
|
|
inventory_product_id: str = Field(..., description="Inventory product UUID reference")
|
|
# product_name: str = Field(..., description="Product name") # DEPRECATED - use inventory_product_id
|
|
forecast_date: date = Field(..., description="Starting date for forecast")
|
|
forecast_days: int = Field(1, ge=1, le=30, description="Number of days to forecast")
|
|
location: str = Field(..., description="Location identifier")
|
|
|
|
# Optional parameters - internally handled
|
|
confidence_level: float = Field(0.8, ge=0.5, le=0.95, description="Confidence level")
|
|
|
|
@validator('forecast_date')
|
|
def validate_forecast_date(cls, v):
|
|
if v < date.today():
|
|
raise ValueError("Forecast date cannot be in the past")
|
|
return v
|
|
|
|
class BatchForecastRequest(BaseModel):
|
|
"""Request schema for batch forecasting"""
|
|
tenant_id: Optional[str] = None # Optional, can be from path parameter
|
|
batch_name: str = Field(..., description="Batch name for tracking")
|
|
inventory_product_ids: List[str] = Field(..., description="List of inventory product IDs")
|
|
forecast_days: int = Field(7, ge=1, le=30, description="Number of days to forecast")
|
|
|
|
class ForecastResponse(BaseModel):
|
|
"""Response schema for forecast results"""
|
|
id: str
|
|
tenant_id: str
|
|
inventory_product_id: str # Reference to inventory service
|
|
# product_name: str # Can be fetched from inventory service if needed for display
|
|
location: str
|
|
forecast_date: datetime
|
|
|
|
# Predictions
|
|
predicted_demand: float
|
|
confidence_lower: float
|
|
confidence_upper: float
|
|
confidence_level: float
|
|
|
|
# Model info
|
|
model_id: str
|
|
model_version: str
|
|
algorithm: str
|
|
|
|
# Context
|
|
business_type: str
|
|
is_holiday: bool
|
|
is_weekend: bool
|
|
day_of_week: int
|
|
|
|
# External factors
|
|
weather_temperature: Optional[float]
|
|
weather_precipitation: Optional[float]
|
|
weather_description: Optional[str]
|
|
traffic_volume: Optional[int]
|
|
|
|
# Metadata
|
|
created_at: datetime
|
|
processing_time_ms: Optional[int]
|
|
features_used: Optional[Dict[str, Any]]
|
|
|
|
class BatchForecastResponse(BaseModel):
|
|
"""Response schema for batch forecast requests"""
|
|
id: str
|
|
tenant_id: str
|
|
batch_name: str
|
|
status: str
|
|
total_products: int
|
|
completed_products: int
|
|
failed_products: int
|
|
|
|
# Timing
|
|
requested_at: datetime
|
|
completed_at: Optional[datetime]
|
|
processing_time_ms: Optional[int]
|
|
|
|
# Results
|
|
forecasts: Optional[List[ForecastResponse]]
|
|
error_message: Optional[str]
|
|
|
|
class MultiDayForecastResponse(BaseModel):
|
|
"""Response schema for multi-day forecast results"""
|
|
tenant_id: str = Field(..., description="Tenant ID")
|
|
inventory_product_id: str = Field(..., description="Inventory product ID")
|
|
forecast_start_date: date = Field(..., description="Start date of forecast period")
|
|
forecast_days: int = Field(..., description="Number of forecasted days")
|
|
forecasts: List[ForecastResponse] = Field(..., description="Daily forecasts")
|
|
total_predicted_demand: float = Field(..., description="Total demand across all days")
|
|
average_confidence_level: float = Field(..., description="Average confidence across all days")
|
|
processing_time_ms: int = Field(..., description="Total processing time")
|
|
|
|
|
|
# ================================================================
|
|
# SCENARIO SIMULATION SCHEMAS - PROFESSIONAL/ENTERPRISE ONLY
|
|
# ================================================================
|
|
|
|
class ScenarioType(str, Enum):
|
|
"""Types of scenarios available for simulation"""
|
|
WEATHER = "weather" # Weather impact (heatwave, cold snap, rain, etc.)
|
|
COMPETITION = "competition" # New competitor opening nearby
|
|
EVENT = "event" # Local event (festival, sports, concert, etc.)
|
|
PRICING = "pricing" # Price changes
|
|
PROMOTION = "promotion" # Promotional campaigns
|
|
HOLIDAY = "holiday" # Holiday periods
|
|
SUPPLY_DISRUPTION = "supply_disruption" # Supply chain issues
|
|
CUSTOM = "custom" # Custom user-defined scenario
|
|
|
|
|
|
class WeatherScenario(BaseModel):
|
|
"""Weather scenario parameters"""
|
|
temperature_change: Optional[float] = Field(None, ge=-30, le=30, description="Temperature change in °C")
|
|
precipitation_change: Optional[float] = Field(None, ge=0, le=100, description="Precipitation change in mm")
|
|
weather_type: Optional[str] = Field(None, description="Weather type (heatwave, cold_snap, rainy, etc.)")
|
|
|
|
|
|
class CompetitionScenario(BaseModel):
|
|
"""Competition scenario parameters"""
|
|
new_competitors: int = Field(1, ge=1, le=10, description="Number of new competitors")
|
|
distance_km: float = Field(0.5, ge=0.1, le=10, description="Distance from location in km")
|
|
estimated_market_share_loss: float = Field(0.1, ge=0, le=0.5, description="Estimated market share loss (0-50%)")
|
|
|
|
|
|
class EventScenario(BaseModel):
|
|
"""Event scenario parameters"""
|
|
event_type: str = Field(..., description="Type of event (festival, sports, concert, etc.)")
|
|
expected_attendance: int = Field(..., ge=0, description="Expected attendance")
|
|
distance_km: float = Field(0.5, ge=0, le=50, description="Distance from location in km")
|
|
duration_days: int = Field(1, ge=1, le=30, description="Duration in days")
|
|
|
|
|
|
class PricingScenario(BaseModel):
|
|
"""Pricing scenario parameters"""
|
|
price_change_percent: float = Field(..., ge=-50, le=100, description="Price change percentage")
|
|
affected_products: Optional[List[str]] = Field(None, description="List of affected product IDs")
|
|
|
|
|
|
class PromotionScenario(BaseModel):
|
|
"""Promotion scenario parameters"""
|
|
discount_percent: float = Field(..., ge=0, le=75, description="Discount percentage")
|
|
promotion_type: str = Field(..., description="Type of promotion (bogo, discount, bundle, etc.)")
|
|
expected_traffic_increase: float = Field(0.2, ge=0, le=2, description="Expected traffic increase (0-200%)")
|
|
|
|
|
|
class ScenarioSimulationRequest(BaseModel):
|
|
"""Request schema for scenario simulation - PROFESSIONAL/ENTERPRISE ONLY"""
|
|
scenario_name: str = Field(..., min_length=3, max_length=200, description="Name for this scenario")
|
|
scenario_type: ScenarioType = Field(..., description="Type of scenario to simulate")
|
|
inventory_product_ids: List[str] = Field(..., min_items=1, description="Products to simulate")
|
|
start_date: date = Field(..., description="Simulation start date")
|
|
duration_days: int = Field(7, ge=1, le=30, description="Simulation duration in days")
|
|
|
|
# Scenario-specific parameters (one should be provided based on scenario_type)
|
|
weather_params: Optional[WeatherScenario] = None
|
|
competition_params: Optional[CompetitionScenario] = None
|
|
event_params: Optional[EventScenario] = None
|
|
pricing_params: Optional[PricingScenario] = None
|
|
promotion_params: Optional[PromotionScenario] = None
|
|
|
|
# Custom scenario parameters
|
|
custom_multipliers: Optional[Dict[str, float]] = Field(
|
|
None,
|
|
description="Custom multipliers for baseline forecast (e.g., {'demand': 1.2, 'traffic': 0.8})"
|
|
)
|
|
|
|
# Comparison settings
|
|
include_baseline: bool = Field(True, description="Include baseline forecast for comparison")
|
|
|
|
@validator('start_date')
|
|
def validate_start_date(cls, v):
|
|
if v < date.today():
|
|
raise ValueError("Simulation start date cannot be in the past")
|
|
return v
|
|
|
|
|
|
class ScenarioImpact(BaseModel):
|
|
"""Impact of scenario on a specific product"""
|
|
inventory_product_id: str
|
|
baseline_demand: float
|
|
simulated_demand: float
|
|
demand_change_percent: float
|
|
confidence_range: tuple[float, float]
|
|
impact_factors: Dict[str, Any] # Breakdown of what drove the change
|
|
|
|
|
|
class ScenarioSimulationResponse(BaseModel):
|
|
"""Response schema for scenario simulation"""
|
|
id: str = Field(..., description="Simulation ID")
|
|
tenant_id: str
|
|
scenario_name: str
|
|
scenario_type: ScenarioType
|
|
|
|
# Simulation parameters
|
|
start_date: date
|
|
end_date: date
|
|
duration_days: int
|
|
|
|
# Results
|
|
baseline_forecasts: Optional[List[ForecastResponse]] = Field(
|
|
None,
|
|
description="Baseline forecasts (if requested)"
|
|
)
|
|
scenario_forecasts: List[ForecastResponse] = Field(..., description="Forecasts with scenario applied")
|
|
|
|
# Impact summary
|
|
total_baseline_demand: float
|
|
total_scenario_demand: float
|
|
overall_impact_percent: float
|
|
product_impacts: List[ScenarioImpact]
|
|
|
|
# Insights and recommendations
|
|
insights: List[str] = Field(..., description="AI-generated insights about the scenario")
|
|
recommendations: List[str] = Field(..., description="Actionable recommendations")
|
|
risk_level: str = Field(..., description="Risk level: low, medium, high")
|
|
|
|
# Metadata
|
|
created_at: datetime
|
|
processing_time_ms: int
|
|
|
|
class Config:
|
|
json_schema_extra = {
|
|
"example": {
|
|
"id": "scenario_123",
|
|
"tenant_id": "tenant_456",
|
|
"scenario_name": "Summer Heatwave Impact",
|
|
"scenario_type": "weather",
|
|
"overall_impact_percent": 15.5,
|
|
"insights": [
|
|
"Cold beverages expected to increase by 45%",
|
|
"Bread products may decrease by 8% due to reduced appetite",
|
|
"Ice cream demand projected to surge by 120%"
|
|
],
|
|
"recommendations": [
|
|
"Increase cold beverage inventory by 40%",
|
|
"Reduce bread production by 10%",
|
|
"Stock additional ice cream varieties"
|
|
],
|
|
"risk_level": "medium"
|
|
}
|
|
}
|
|
|
|
|
|
class ScenarioComparisonRequest(BaseModel):
|
|
"""Request to compare multiple scenarios"""
|
|
scenario_ids: List[str] = Field(..., min_items=2, max_items=5, description="Scenario IDs to compare")
|
|
|
|
|
|
class ScenarioComparisonResponse(BaseModel):
|
|
"""Response comparing multiple scenarios"""
|
|
scenarios: List[ScenarioSimulationResponse]
|
|
comparison_matrix: Dict[str, Dict[str, Any]]
|
|
best_case_scenario_id: str
|
|
worst_case_scenario_id: str
|
|
recommended_action: str
|
|
|
|
|