""" Price Forecaster Predicts supplier price changes for opportunistic buying recommendations Identifies optimal timing for bulk purchases and price negotiation opportunities """ import pandas as pd import numpy as np from typing import Dict, List, Any, Optional, Tuple import structlog from datetime import datetime, timedelta from scipy import stats from sklearn.linear_model import LinearRegression from sklearn.ensemble import RandomForestRegressor import warnings warnings.filterwarnings('ignore') logger = structlog.get_logger() class PriceForecaster: """ Forecasts ingredient and product prices for opportunistic procurement. Capabilities: 1. Short-term price forecasting (1-4 weeks) 2. Seasonal price pattern detection 3. Price trend analysis 4. Buy/wait recommendations 5. Bulk purchase opportunity identification 6. Price volatility assessment 7. Supplier comparison for price optimization """ def __init__(self): self.price_models = {} self.seasonal_patterns = {} self.volatility_scores = {} async def forecast_price( self, tenant_id: str, ingredient_id: str, price_history: pd.DataFrame, forecast_horizon_days: int = 30, min_history_days: int = 180 ) -> Dict[str, Any]: """ Forecast future prices and generate procurement recommendations. Args: tenant_id: Tenant identifier ingredient_id: Ingredient/product identifier price_history: Historical price data with columns: - date - price_per_unit - quantity_purchased (optional) - supplier_id (optional) forecast_horizon_days: Days to forecast ahead (default 30) min_history_days: Minimum days of history required (default 180) Returns: Dictionary with price forecast and insights """ logger.info( "Forecasting prices", tenant_id=tenant_id, ingredient_id=ingredient_id, history_days=len(price_history), forecast_days=forecast_horizon_days ) # Validate input if len(price_history) < min_history_days: logger.warning( "Insufficient price history", ingredient_id=ingredient_id, days=len(price_history), required=min_history_days ) return self._insufficient_data_response( tenant_id, ingredient_id, price_history ) # Prepare data price_history = price_history.copy() price_history['date'] = pd.to_datetime(price_history['date']) price_history = price_history.sort_values('date') # Calculate price statistics price_stats = self._calculate_price_statistics(price_history) # Detect seasonal patterns seasonal_analysis = self._detect_seasonal_patterns(price_history) # Detect trends trend_analysis = self._analyze_price_trends(price_history) # Forecast future prices forecast = self._generate_price_forecast( price_history, forecast_horizon_days, seasonal_analysis, trend_analysis ) # Calculate volatility volatility = self._calculate_price_volatility(price_history) # Generate buy/wait recommendations recommendations = self._generate_procurement_recommendations( price_history, forecast, price_stats, volatility, trend_analysis ) # Identify bulk purchase opportunities bulk_opportunities = self._identify_bulk_opportunities( forecast, price_stats, volatility ) # Generate insights insights = self._generate_price_insights( tenant_id, ingredient_id, price_stats, forecast, recommendations, bulk_opportunities, trend_analysis, volatility ) # Store models self.seasonal_patterns[ingredient_id] = seasonal_analysis self.volatility_scores[ingredient_id] = volatility logger.info( "Price forecasting complete", ingredient_id=ingredient_id, avg_forecast_price=forecast['mean_forecast_price'], recommendation=recommendations['action'], insights_generated=len(insights) ) return { 'tenant_id': tenant_id, 'ingredient_id': ingredient_id, 'forecasted_at': datetime.utcnow().isoformat(), 'history_days': len(price_history), 'forecast_horizon_days': forecast_horizon_days, 'price_stats': price_stats, 'seasonal_analysis': seasonal_analysis, 'trend_analysis': trend_analysis, 'forecast': forecast, 'volatility': volatility, 'recommendations': recommendations, 'bulk_opportunities': bulk_opportunities, 'insights': insights } def _calculate_price_statistics( self, price_history: pd.DataFrame ) -> Dict[str, float]: """ Calculate comprehensive price statistics. Args: price_history: Historical price data Returns: Dictionary of price statistics """ prices = price_history['price_per_unit'].values # Basic statistics current_price = float(prices[-1]) mean_price = float(prices.mean()) std_price = float(prices.std()) cv_price = (std_price / mean_price) if mean_price > 0 else 0 # Price range min_price = float(prices.min()) max_price = float(prices.max()) price_range_pct = ((max_price - min_price) / mean_price * 100) if mean_price > 0 else 0 # Recent vs historical if len(prices) >= 60: recent_30d_mean = float(prices[-30:].mean()) historical_mean = float(prices[:-30].mean()) price_change_pct = ((recent_30d_mean - historical_mean) / historical_mean * 100) if historical_mean > 0 else 0 else: recent_30d_mean = current_price price_change_pct = 0 # Price momentum (last 7 days vs previous 7 days) if len(prices) >= 14: last_week = prices[-7:].mean() prev_week = prices[-14:-7].mean() momentum = ((last_week - prev_week) / prev_week * 100) if prev_week > 0 else 0 else: momentum = 0 return { 'current_price': current_price, 'mean_price': mean_price, 'std_price': std_price, 'cv_price': cv_price, 'min_price': min_price, 'max_price': max_price, 'price_range_pct': price_range_pct, 'recent_30d_mean': recent_30d_mean, 'price_change_30d_pct': price_change_pct, 'momentum_7d_pct': momentum, 'data_points': len(prices) } def _detect_seasonal_patterns( self, price_history: pd.DataFrame ) -> Dict[str, Any]: """ Detect seasonal price patterns. Args: price_history: Historical price data Returns: Seasonal pattern analysis """ # Extract month from date price_history = price_history.copy() price_history['month'] = price_history['date'].dt.month # Calculate average price per month monthly_avg = price_history.groupby('month')['price_per_unit'].agg(['mean', 'std', 'count']) overall_mean = price_history['price_per_unit'].mean() seasonal_patterns = {} has_seasonality = False month_names = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'Jun', 'Jul', 'Aug', 'Sep', 'Oct', 'Nov', 'Dec'] for month in range(1, 13): if month in monthly_avg.index and monthly_avg.loc[month, 'count'] >= 3: month_mean = monthly_avg.loc[month, 'mean'] deviation_pct = ((month_mean - overall_mean) / overall_mean * 100) if overall_mean > 0 else 0 seasonal_patterns[month_names[month-1]] = { 'month': month, 'avg_price': round(float(month_mean), 2), 'deviation_pct': round(float(deviation_pct), 2), 'sample_size': int(monthly_avg.loc[month, 'count']) } # Significant seasonality if >10% deviation if abs(deviation_pct) > 10: has_seasonality = True return { 'has_seasonality': has_seasonality, 'monthly_patterns': seasonal_patterns, 'overall_mean_price': round(float(overall_mean), 2) } def _analyze_price_trends( self, price_history: pd.DataFrame ) -> Dict[str, Any]: """ Analyze price trends using linear regression. Args: price_history: Historical price data Returns: Trend analysis """ # Create time index (days from start) price_history = price_history.copy() price_history['days_from_start'] = ( price_history['date'] - price_history['date'].min() ).dt.days X = price_history['days_from_start'].values.reshape(-1, 1) y = price_history['price_per_unit'].values # Fit linear regression model = LinearRegression() model.fit(X, y) # Calculate trend slope = float(model.coef_[0]) intercept = float(model.intercept_) r_squared = float(model.score(X, y)) # Trend direction and magnitude avg_price = y.mean() trend_pct_per_month = (slope * 30 / avg_price * 100) if avg_price > 0 else 0 # Classify trend if abs(trend_pct_per_month) < 2: trend_direction = 'stable' elif trend_pct_per_month > 2: trend_direction = 'increasing' else: trend_direction = 'decreasing' # Recent trend (last 90 days) if len(price_history) >= 90: recent_data = price_history.tail(90).copy() recent_X = recent_data['days_from_start'].values.reshape(-1, 1) recent_y = recent_data['price_per_unit'].values recent_model = LinearRegression() recent_model.fit(recent_X, recent_y) recent_slope = float(recent_model.coef_[0]) recent_trend_pct = (recent_slope * 30 / recent_y.mean() * 100) if recent_y.mean() > 0 else 0 else: recent_trend_pct = trend_pct_per_month return { 'trend_direction': trend_direction, 'trend_pct_per_month': round(trend_pct_per_month, 2), 'recent_trend_pct_per_month': round(recent_trend_pct, 2), 'slope': round(slope, 4), 'r_squared': round(r_squared, 3), 'is_accelerating': abs(recent_trend_pct) > abs(trend_pct_per_month) * 1.5 } def _generate_price_forecast( self, price_history: pd.DataFrame, forecast_days: int, seasonal_analysis: Dict[str, Any], trend_analysis: Dict[str, Any] ) -> Dict[str, Any]: """ Generate price forecast for specified horizon. Args: price_history: Historical price data forecast_days: Days to forecast seasonal_analysis: Seasonal patterns trend_analysis: Trend analysis Returns: Price forecast """ current_price = price_history['price_per_unit'].iloc[-1] current_date = price_history['date'].iloc[-1] # Simple forecast: current price + trend + seasonal adjustment trend_slope = trend_analysis['slope'] forecast_prices = [] forecast_dates = [] for day in range(1, forecast_days + 1): forecast_date = current_date + timedelta(days=day) forecast_dates.append(forecast_date) # Base forecast from trend base_forecast = current_price + (trend_slope * day) # Seasonal adjustment if seasonal_analysis['has_seasonality']: month_name = forecast_date.strftime('%b') if month_name in seasonal_analysis['monthly_patterns']: month_deviation = seasonal_analysis['monthly_patterns'][month_name]['deviation_pct'] seasonal_adjustment = base_forecast * (month_deviation / 100) base_forecast += seasonal_adjustment forecast_prices.append(base_forecast) forecast_prices = np.array(forecast_prices) # Calculate confidence intervals (±2 std) historical_std = price_history['price_per_unit'].std() lower_bound = forecast_prices - 2 * historical_std upper_bound = forecast_prices + 2 * historical_std return { 'forecast_dates': [d.strftime('%Y-%m-%d') for d in forecast_dates], 'forecast_prices': [round(float(p), 2) for p in forecast_prices], 'lower_bound': [round(float(p), 2) for p in lower_bound], 'upper_bound': [round(float(p), 2) for p in upper_bound], 'mean_forecast_price': round(float(forecast_prices.mean()), 2), 'min_forecast_price': round(float(forecast_prices.min()), 2), 'max_forecast_price': round(float(forecast_prices.max()), 2), 'confidence': self._calculate_forecast_confidence(price_history, trend_analysis) } def _calculate_forecast_confidence( self, price_history: pd.DataFrame, trend_analysis: Dict[str, Any] ) -> int: """Calculate confidence in price forecast (0-100).""" confidence = 50 # Base confidence # More data = higher confidence data_points = len(price_history) if data_points >= 365: confidence += 30 elif data_points >= 180: confidence += 20 else: confidence += 10 # Strong trend = higher confidence r_squared = trend_analysis['r_squared'] if r_squared > 0.7: confidence += 20 elif r_squared > 0.5: confidence += 10 # Low volatility = higher confidence cv = price_history['price_per_unit'].std() / price_history['price_per_unit'].mean() if cv < 0.1: confidence += 10 elif cv < 0.2: confidence += 5 return min(100, confidence) def _calculate_price_volatility( self, price_history: pd.DataFrame ) -> Dict[str, Any]: """ Calculate price volatility metrics. Args: price_history: Historical price data Returns: Volatility analysis """ prices = price_history['price_per_unit'].values # Coefficient of variation cv = float(prices.std() / prices.mean()) if prices.mean() > 0 else 0 # Price changes (day-to-day) price_changes = np.diff(prices) pct_changes = (price_changes / prices[:-1] * 100) # Volatility classification if cv < 0.1: volatility_level = 'low' elif cv < 0.2: volatility_level = 'medium' else: volatility_level = 'high' return { 'coefficient_of_variation': round(cv, 3), 'volatility_level': volatility_level, 'avg_daily_change_pct': round(float(np.abs(pct_changes).mean()), 2), 'max_daily_increase_pct': round(float(pct_changes.max()), 2), 'max_daily_decrease_pct': round(float(pct_changes.min()), 2) } def _generate_procurement_recommendations( self, price_history: pd.DataFrame, forecast: Dict[str, Any], price_stats: Dict[str, float], volatility: Dict[str, Any], trend_analysis: Dict[str, Any] ) -> Dict[str, Any]: """ Generate buy/wait recommendations based on forecast. Args: price_history: Historical data forecast: Price forecast price_stats: Price statistics volatility: Volatility analysis trend_analysis: Trend analysis Returns: Procurement recommendations """ current_price = price_stats['current_price'] forecast_mean = forecast['mean_forecast_price'] forecast_min = forecast['min_forecast_price'] # Calculate expected price change expected_change_pct = ((forecast_mean - current_price) / current_price * 100) if current_price > 0 else 0 # Decision logic with i18n-friendly reasoning codes if expected_change_pct < -5: # Price expected to drop >5% action = 'wait' reasoning_data = { 'type': 'decrease_expected', 'parameters': { 'change_pct': round(abs(expected_change_pct), 1), 'forecast_days': 30, 'current_price': round(current_price, 2), 'forecast_mean': round(forecast_mean, 2) } } urgency = 'low' elif expected_change_pct > 5: # Price expected to increase >5% action = 'buy_now' reasoning_data = { 'type': 'increase_expected', 'parameters': { 'change_pct': round(expected_change_pct, 1), 'forecast_days': 30, 'current_price': round(current_price, 2), 'forecast_mean': round(forecast_mean, 2) } } urgency = 'high' elif volatility['volatility_level'] == 'high': # High volatility - wait for dip action = 'wait_for_dip' reasoning_data = { 'type': 'high_volatility', 'parameters': { 'coefficient': round(volatility['coefficient_of_variation'], 2), 'volatility_level': volatility['volatility_level'], 'avg_daily_change_pct': round(volatility['avg_daily_change_pct'], 2) } } urgency = 'medium' elif current_price < price_stats['mean_price'] * 0.95: # Currently below average below_avg_pct = ((price_stats["mean_price"] - current_price) / price_stats["mean_price"] * 100) action = 'buy_now' reasoning_data = { 'type': 'below_average', 'parameters': { 'current_price': round(current_price, 2), 'mean_price': round(price_stats['mean_price'], 2), 'below_avg_pct': round(below_avg_pct, 1) } } urgency = 'medium' else: # Neutral action = 'normal_purchase' reasoning_data = { 'type': 'stable', 'parameters': { 'current_price': round(current_price, 2), 'forecast_mean': round(forecast_mean, 2), 'expected_change_pct': round(expected_change_pct, 2) } } urgency = 'low' # Optimal purchase timing min_price_index = forecast['forecast_prices'].index(forecast_min) optimal_date = forecast['forecast_dates'][min_price_index] return { 'action': action, 'reasoning_data': reasoning_data, 'urgency': urgency, 'expected_price_change_pct': round(expected_change_pct, 2), 'current_price': current_price, 'forecast_mean_price': forecast_mean, 'forecast_min_price': forecast_min, 'optimal_purchase_date': optimal_date, 'days_until_optimal': min_price_index + 1 } def _identify_bulk_opportunities( self, forecast: Dict[str, Any], price_stats: Dict[str, float], volatility: Dict[str, Any] ) -> Dict[str, Any]: """ Identify bulk purchase opportunities. Args: forecast: Price forecast price_stats: Price statistics volatility: Volatility analysis Returns: Bulk opportunity analysis """ current_price = price_stats['current_price'] forecast_max = forecast['max_forecast_price'] # Potential savings from bulk buy at current price if forecast_max > current_price: potential_savings_pct = ((forecast_max - current_price) / current_price * 100) if potential_savings_pct > 10: opportunity_level = 'high' elif potential_savings_pct > 5: opportunity_level = 'medium' else: opportunity_level = 'low' has_opportunity = potential_savings_pct > 5 else: potential_savings_pct = 0 opportunity_level = 'none' has_opportunity = False return { 'has_bulk_opportunity': has_opportunity, 'opportunity_level': opportunity_level, 'potential_savings_pct': round(potential_savings_pct, 2), 'recommended_bulk_quantity_months': 2 if has_opportunity and volatility['volatility_level'] != 'high' else 1 } def _generate_price_insights( self, tenant_id: str, ingredient_id: str, price_stats: Dict[str, float], forecast: Dict[str, Any], recommendations: Dict[str, Any], bulk_opportunities: Dict[str, Any], trend_analysis: Dict[str, Any], volatility: Dict[str, Any] ) -> List[Dict[str, Any]]: """ Generate actionable pricing insights. Returns: List of insights """ insights = [] # Insight 1: Buy now recommendation if recommendations['action'] == 'buy_now': insights.append({ 'type': 'recommendation', 'priority': recommendations['urgency'], 'category': 'procurement', 'title': f'Buy Now: Price Increasing {recommendations["expected_price_change_pct"]:.1f}%', 'reasoning_data': recommendations['reasoning_data'], 'impact_type': 'cost_avoidance', 'impact_value': abs(recommendations['expected_price_change_pct']), 'impact_unit': 'percentage', 'confidence': forecast['confidence'], 'metrics_json': { 'ingredient_id': ingredient_id, 'current_price': price_stats['current_price'], 'forecast_price': forecast['mean_forecast_price'], 'expected_change_pct': recommendations['expected_price_change_pct'], 'optimal_date': recommendations['optimal_purchase_date'] }, 'actionable': True, 'recommendation_actions': [ { 'label': 'Purchase Now', 'action': 'create_purchase_order', 'params': { 'ingredient_id': ingredient_id, 'priority': 'high' } } ], 'source_service': 'procurement', 'source_model': 'price_forecaster' }) # Insight 2: Wait recommendation elif recommendations['action'] == 'wait': insights.append({ 'type': 'recommendation', 'priority': 'medium', 'category': 'procurement', 'title': f'Wait to Buy: Price Decreasing {abs(recommendations["expected_price_change_pct"]):.1f}%', 'reasoning_data': { **recommendations['reasoning_data'], 'optimal_purchase_date': recommendations['optimal_purchase_date'], 'days_until_optimal': recommendations['days_until_optimal'] }, 'impact_type': 'cost_savings', 'impact_value': abs(recommendations['expected_price_change_pct']), 'impact_unit': 'percentage', 'confidence': forecast['confidence'], 'metrics_json': { 'ingredient_id': ingredient_id, 'current_price': price_stats['current_price'], 'forecast_min_price': forecast['min_forecast_price'], 'optimal_date': recommendations['optimal_purchase_date'], 'days_until_optimal': recommendations['days_until_optimal'] }, 'actionable': True, 'recommendation_actions': [ { 'label': 'Delay Purchase', 'action': 'delay_purchase_order', 'params': { 'ingredient_id': ingredient_id, 'delay_days': recommendations['days_until_optimal'] } } ], 'source_service': 'procurement', 'source_model': 'price_forecaster' }) # Insight 3: Bulk opportunity if bulk_opportunities['has_bulk_opportunity']: insights.append({ 'type': 'optimization', 'priority': bulk_opportunities['opportunity_level'], 'category': 'procurement', 'title': f'Bulk Buy Opportunity: Save {bulk_opportunities["potential_savings_pct"]:.1f}%', 'description': f'Current price is favorable. Purchasing {bulk_opportunities["recommended_bulk_quantity_months"]} months supply now could save {bulk_opportunities["potential_savings_pct"]:.1f}% vs future prices.', 'impact_type': 'cost_savings', 'impact_value': bulk_opportunities['potential_savings_pct'], 'impact_unit': 'percentage', 'confidence': forecast['confidence'], 'metrics_json': { 'ingredient_id': ingredient_id, 'current_price': price_stats['current_price'], 'forecast_max_price': forecast['max_forecast_price'], 'savings_pct': bulk_opportunities['potential_savings_pct'], 'recommended_months_supply': bulk_opportunities['recommended_bulk_quantity_months'] }, 'actionable': True, 'recommendation_actions': [ { 'label': 'Create Bulk Order', 'action': 'create_bulk_purchase_order', 'params': { 'ingredient_id': ingredient_id, 'months_supply': bulk_opportunities['recommended_bulk_quantity_months'] } } ], 'source_service': 'procurement', 'source_model': 'price_forecaster' }) # Insight 4: High volatility warning if volatility['volatility_level'] == 'high': insights.append({ 'type': 'alert', 'priority': 'medium', 'category': 'procurement', 'title': f'High Price Volatility: CV={volatility["coefficient_of_variation"]:.2f}', 'description': f'Ingredient {ingredient_id} shows high price volatility with {volatility["avg_daily_change_pct"]:.1f}% average daily change. Consider alternative suppliers or hedge strategies.', 'impact_type': 'risk_warning', 'impact_value': volatility['coefficient_of_variation'], 'impact_unit': 'cv_score', 'confidence': 90, 'metrics_json': { 'ingredient_id': ingredient_id, 'volatility_level': volatility['volatility_level'], 'cv': volatility['coefficient_of_variation'], 'avg_daily_change_pct': volatility['avg_daily_change_pct'] }, 'actionable': True, 'recommendation_actions': [ { 'label': 'Find Alternative Suppliers', 'action': 'search_alternative_suppliers', 'params': {'ingredient_id': ingredient_id} } ], 'source_service': 'procurement', 'source_model': 'price_forecaster' }) # Insight 5: Strong price trend if abs(trend_analysis['trend_pct_per_month']) > 5: direction = 'increasing' if trend_analysis['trend_pct_per_month'] > 0 else 'decreasing' insights.append({ 'type': 'insight', 'priority': 'medium', 'category': 'procurement', 'title': f'Strong Price Trend: {direction.title()} {abs(trend_analysis["trend_pct_per_month"]):.1f}%/month', 'description': f'Ingredient {ingredient_id} prices are {direction} at {abs(trend_analysis["trend_pct_per_month"]):.1f}% per month. Plan procurement strategy accordingly.', 'impact_type': 'trend_warning', 'impact_value': abs(trend_analysis['trend_pct_per_month']), 'impact_unit': 'pct_per_month', 'confidence': int(trend_analysis['r_squared'] * 100), 'metrics_json': { 'ingredient_id': ingredient_id, 'trend_direction': trend_analysis['trend_direction'], 'trend_pct_per_month': trend_analysis['trend_pct_per_month'], 'r_squared': trend_analysis['r_squared'] }, 'actionable': False, 'source_service': 'procurement', 'source_model': 'price_forecaster' }) return insights def _insufficient_data_response( self, tenant_id: str, ingredient_id: str, price_history: pd.DataFrame ) -> Dict[str, Any]: """Return response when insufficient data available.""" return { 'tenant_id': tenant_id, 'ingredient_id': ingredient_id, 'forecasted_at': datetime.utcnow().isoformat(), 'history_days': len(price_history), 'forecast_horizon_days': 0, 'price_stats': {}, 'seasonal_analysis': {'has_seasonality': False}, 'trend_analysis': {}, 'forecast': {}, 'volatility': {}, 'recommendations': { 'action': 'insufficient_data', 'reasoning_data': { 'type': 'insufficient_data', 'parameters': { 'history_days': len(price_history), 'min_required_days': 180 } }, 'urgency': 'low' }, 'bulk_opportunities': {'has_bulk_opportunity': False}, 'insights': [] } def get_seasonal_patterns(self, ingredient_id: str) -> Optional[Dict[str, Any]]: """Get cached seasonal patterns for an ingredient.""" return self.seasonal_patterns.get(ingredient_id) def get_volatility_score(self, ingredient_id: str) -> Optional[Dict[str, Any]]: """Get cached volatility score for an ingredient.""" return self.volatility_scores.get(ingredient_id)