124 lines
3.9 KiB
Python
124 lines
3.9 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 AlertType(str, Enum):
|
|
HIGH_DEMAND = "high_demand"
|
|
LOW_DEMAND = "low_demand"
|
|
STOCKOUT_RISK = "stockout_risk"
|
|
OVERPRODUCTION = "overproduction"
|
|
|
|
class ForecastRequest(BaseModel):
|
|
"""Request schema for generating forecasts"""
|
|
tenant_id: str = Field(..., description="Tenant ID")
|
|
product_name: str = Field(..., description="Product name")
|
|
location: str = Field(..., description="Location identifier")
|
|
forecast_date: date = Field(..., description="Date for which to generate forecast")
|
|
business_type: BusinessType = Field(BusinessType.INDIVIDUAL, description="Business model type")
|
|
|
|
# Optional context
|
|
include_weather: bool = Field(True, description="Include weather data in forecast")
|
|
include_traffic: bool = Field(True, description="Include traffic data in forecast")
|
|
confidence_level: float = Field(0.8, ge=0.5, le=0.95, description="Confidence level for intervals")
|
|
|
|
@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: str = Field(..., description="Tenant ID")
|
|
batch_name: str = Field(..., description="Batch name for tracking")
|
|
products: List[str] = Field(..., description="List of product names")
|
|
location: str = Field(..., description="Location identifier")
|
|
forecast_days: int = Field(7, ge=1, le=30, description="Number of days to forecast")
|
|
business_type: BusinessType = Field(BusinessType.INDIVIDUAL, description="Business model type")
|
|
|
|
# Options
|
|
include_weather: bool = Field(True, description="Include weather data")
|
|
include_traffic: bool = Field(True, description="Include traffic data")
|
|
confidence_level: float = Field(0.8, ge=0.5, le=0.95, description="Confidence level")
|
|
|
|
class ForecastResponse(BaseModel):
|
|
"""Response schema for forecast results"""
|
|
id: str
|
|
tenant_id: str
|
|
product_name: str
|
|
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 AlertResponse(BaseModel):
|
|
"""Response schema for forecast alerts"""
|
|
id: str
|
|
tenant_id: str
|
|
forecast_id: str
|
|
alert_type: str
|
|
severity: str
|
|
message: str
|
|
is_active: bool
|
|
created_at: datetime
|
|
acknowledged_at: Optional[datetime]
|
|
notification_sent: bool
|
|
|