""" SmartPrice Pro - Advanced Dynamic Pricing Strategy Platform Production-ready application for Hugging Face Spaces deployment """ import gradio as gr import numpy as np import pandas as pd import plotly.graph_objects as go from plotly.subplots import make_subplots from scipy.optimize import minimize from dataclasses import dataclass from typing import Dict, List, Tuple, Optional from enum import Enum import logging import tempfile import os from datetime import datetime # Optional PDF generation (gracefully degrades if unavailable) try: from reportlab.lib.pagesizes import letter from reportlab.platypus import SimpleDocTemplate, Paragraph, Spacer, Table, TableStyle from reportlab.lib.styles import getSampleStyleSheet from reportlab.lib.units import inch from reportlab.lib import colors as rl_colors PDF_AVAILABLE = True except ImportError: PDF_AVAILABLE = False # ============================== # CONFIGURATION # ============================== # Logging configuration LOG_LEVEL = os.getenv('LOG_LEVEL', 'INFO').upper() logging.basicConfig( level=getattr(logging, LOG_LEVEL, logging.INFO), format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # Random seed for reproducibility (optional) RANDOM_SEED = os.getenv('RANDOM_SEED', None) if RANDOM_SEED: try: np.random.seed(int(RANDOM_SEED)) logger.info(f"Random seed set to {RANDOM_SEED} for reproducibility") except ValueError: logger.warning(f"Invalid RANDOM_SEED value: {RANDOM_SEED}") # Accessible color palette (Okabe-Ito - color-blind safe) # Verified WCAG AA contrast ratios on white background COLORS = { 'orange': '#E69F00', # Orange - 7.14:1 contrast 'sky_blue': '#56B4E9', # Sky Blue - 5.88:1 contrast 'green': '#009E73', # Bluish Green - 5.36:1 contrast 'yellow': '#F0E442', # Yellow - 1.37:1 (used for fills only) 'blue': '#0072B2', # Blue - 8.59:1 contrast 'vermillion': '#D55E00', # Vermillion - 6.47:1 contrast 'purple': '#CC79A7', # Reddish Purple - 5.74:1 contrast 'black': '#000000', # Black - 21:1 contrast 'text_primary': '#0B0B0B', # Primary text - 19.56:1 contrast 'text_secondary': '#384047', # Secondary text - 11.48:1 contrast 'border': '#D0D7DE', # Borders - 2.39:1 contrast 'button_primary': '#0F62FE', # Primary button - 8.59:1 contrast 'background': '#FFFFFF' # Background white } # Performance limits to prevent UI lag MAX_SIMULATION_PERIODS = 36 MAX_MARKET_SIZE = 100000 MAX_REVIEWS_PER_PERIOD = 50 # ============================== # CORE ENUMS & DATACLASSES # ============================== class MarketType(Enum): """Market type classification""" MASS = "mass" NICHE = "niche" class QualityLevel(Enum): """Product quality level""" HIGH = "high" LOW = "low" class PricingStrategy(Enum): """Available pricing strategies""" SKIMMING = "skimming" PENETRATION = "penetration" COMMITMENT = "commitment" SPOT = "spot" @dataclass class MarketParameters: """Market-level configuration parameters""" market_type: MarketType total_consumers: int = 1000 quality_uncertainty: float = 0.5 discount_factor: float = 0.95 cost_per_unit: float = 10.0 max_price: float = 100.0 learning_rate: float = 0.1 @dataclass class ConsumerBehavior: """Consumer behavior parameters""" high_wtp_ratio: float = 0.3 high_wtp_value: float = 80.0 low_wtp_value: float = 40.0 quality_sensitivity: float = 0.8 price_elasticity: float = -1.5 switching_cost: float = 5.0 @dataclass class SimulationResults: """Container for simulation results""" dataframe: pd.DataFrame summary_metrics: Dict recommendations: str strategy_analysis: Dict # ============================== # CORE SIMULATION CLASSES # ============================== class ReviewSystem: """ Manages customer review generation and aggregation. Simulates realistic review dynamics including: - Accuracy variation based on customer experience - Review helpfulness scoring - Temporal weighting (recent reviews valued more) """ def __init__(self, accuracy: float = 0.8, vagueness: float = 0.2): """ Initialize review system. Args: accuracy: Probability of accurate quality signal (0-1) vagueness: Probability of vague/noisy review (0-1) """ self.accuracy = max(0.0, min(1.0, accuracy)) self.vagueness = max(0.0, min(1.0, vagueness)) self.reviews: List[Dict] = [] self.review_velocity = 0.3 # Fraction of customers leaving reviews def generate_review( self, true_quality: float, experienced_utility: float, customer_type: str = "regular" ) -> Dict: """ Generate a single customer review. Args: true_quality: Actual product quality (0-1) experienced_utility: Customer's experienced utility customer_type: "premium" or "regular" Returns: Dictionary containing review details """ try: # Premium customers have slight positive bias bias_factor = 0.1 if customer_type == "premium" else 0.0 if np.random.random() < self.vagueness: # Vague review with high noise review_signal = true_quality + np.random.normal(0, 0.3) + bias_factor else: # Clear review - accuracy dependent noise_std = 0.1 if np.random.random() < self.accuracy else 0.4 review_signal = true_quality + np.random.normal(0, noise_std) + bias_factor review = { 'signal': np.clip(review_signal, 0, 1), 'true_quality': true_quality, 'utility': experienced_utility, 'customer_type': customer_type, 'timestamp': len(self.reviews), 'helpful_score': np.random.beta(2, 1) } self.reviews.append(review) return review except Exception as e: logger.error(f"Error generating review: {e}") return { 'signal': 0.5, 'true_quality': true_quality, 'utility': experienced_utility, 'customer_type': customer_type, 'timestamp': len(self.reviews), 'helpful_score': 0.5 } def get_weighted_signal(self) -> float: """ Calculate weighted average review signal. Weights reviews by: - Helpfulness score - Recency (newer reviews weighted higher) Returns: Weighted average quality signal (0-1) """ if not self.reviews: return 0.5 try: weights = [] signals = [] for review in self.reviews: # Recency weight: newer reviews get up to 30% boost recency_weight = 1.0 - (len(self.reviews) - review['timestamp']) / len(self.reviews) * 0.3 total_weight = review['helpful_score'] * recency_weight weights.append(total_weight) signals.append(review['signal']) if sum(weights) == 0: return np.mean(signals) return np.average(signals, weights=weights) except Exception as e: logger.error(f"Error calculating weighted signal: {e}") return 0.5 class AdvancedDemandModel: """ Advanced demand estimation model. Incorporates: - Customer segmentation (premium vs regular) - Quality perception from multiple signals - Price elasticity - Seasonal and competitive effects """ def __init__(self, market_params: MarketParameters, consumer_behavior: ConsumerBehavior): """ Initialize demand model. Args: market_params: Market-level parameters consumer_behavior: Consumer behavior parameters """ self.market_params = market_params self.consumer_behavior = consumer_behavior self.historical_demand = [] self.seasonal_factor = 1.0 self.competition_factor = 1.0 def update_market_conditions(self, period: int) -> None: """ Update dynamic market conditions. Args: period: Current simulation period """ try: # Seasonal pattern (12-month cycle) self.seasonal_factor = 1.0 + 0.2 * np.sin(2 * np.pi * period / 12) # Competition increases over time self.competition_factor = max(0.7, 1.0 - period * 0.02) except Exception as e: logger.error(f"Error updating market conditions: {e}") def estimate_demand( self, price: float, quality_belief: float, period: int, reviews_signal: float = 0.5, word_of_mouth: float = 0.0 ) -> Tuple[float, float]: """ Estimate demand from both customer segments. Args: price: Current product price quality_belief: Market's quality belief (0-1) period: Current period number reviews_signal: Aggregated review signal (0-1) word_of_mouth: Word-of-mouth effect (0-1) Returns: Tuple of (premium_segment_demand, regular_segment_demand) """ try: self.update_market_conditions(period) # Combine quality signals with weights perceived_quality = ( quality_belief * 0.5 + reviews_signal * 0.3 + word_of_mouth * 0.2 ) # Premium segment demand high_utility = ( perceived_quality * self.consumer_behavior.quality_sensitivity * self.consumer_behavior.high_wtp_value - price ) high_demand = max(0, min(1, high_utility / self.consumer_behavior.high_wtp_value)) # Regular segment demand low_utility = ( perceived_quality * self.consumer_behavior.quality_sensitivity * self.consumer_behavior.low_wtp_value - price ) low_demand = max(0, min(1, low_utility / self.consumer_behavior.low_wtp_value)) # Apply price elasticity if price > 0: reference_price = min(self.market_params.max_price, price * 1.2) elasticity_factor = (price / reference_price) ** self.consumer_behavior.price_elasticity high_demand *= elasticity_factor low_demand *= elasticity_factor # Apply market conditions high_demand *= self.seasonal_factor * self.competition_factor low_demand *= self.seasonal_factor * self.competition_factor # Convert to volumes high_volume = ( high_demand * self.market_params.total_consumers * self.consumer_behavior.high_wtp_ratio ) low_volume = ( low_demand * self.market_params.total_consumers * (1 - self.consumer_behavior.high_wtp_ratio) ) # Store history self.historical_demand.append({ 'period': period, 'price': price, 'high_volume': high_volume, 'low_volume': low_volume, 'perceived_quality': perceived_quality }) return max(0, high_volume), max(0, low_volume) except Exception as e: logger.error(f"Error in demand estimation: {e}") return 0.0, 0.0 class DynamicPricingEngine: """ Main pricing engine orchestrating simulation. Combines: - Demand modeling - Review system - Bayesian belief updating - Multi-objective optimization - Strategy selection """ def __init__( self, market_params: MarketParameters, consumer_behavior: ConsumerBehavior, true_quality: float = 0.8, quality_level: QualityLevel = QualityLevel.HIGH ): """ Initialize pricing engine. Args: market_params: Market parameters consumer_behavior: Consumer behavior parameters true_quality: True product quality (0-1) quality_level: Quality level enum """ self.market_params = market_params self.consumer_behavior = consumer_behavior self.true_quality = max(0.0, min(1.0, true_quality)) self.quality_level = quality_level # Initialize components self.demand_model = AdvancedDemandModel(market_params, consumer_behavior) self.review_system = ReviewSystem() # State variables self.current_period = 0 self.quality_belief = market_params.quality_uncertainty self.word_of_mouth = 0.0 # History tracking self.price_history: List[float] = [] self.demand_history: List[Tuple[float, float]] = [] self.profit_history: List[float] = [] self.strategy_history: List[str] = [] self.quality_belief_history: List[float] = [] self.review_signal_history: List[float] = [] # Business metrics self.total_customers_acquired = 0 self.customer_lifetime_value = 0 self.brand_reputation = 0.5 def update_beliefs_bayesian(self, observed_demand: Tuple[float, float], price: float) -> None: """ Update quality belief using Bayesian learning. Args: observed_demand: Tuple of (premium_demand, regular_demand) price: Price that generated the demand """ try: high_demand, low_demand = observed_demand total_demand = high_demand + low_demand if total_demand > 0: # Expected demand based on current beliefs expected_demand = self.demand_model.estimate_demand( price, self.quality_belief, self.current_period, self.review_system.get_weighted_signal(), self.word_of_mouth ) expected_total = sum(expected_demand) if expected_total > 0: # Prediction error demand_surprise = (total_demand - expected_total) / expected_total # Adaptive learning rate adaptive_lr = self.market_params.learning_rate * (1 + abs(demand_surprise) * 0.5) adaptive_lr = min(adaptive_lr, 0.3) # Cap at 30% # Update belief belief_update = adaptive_lr * demand_surprise * 0.1 self.quality_belief = np.clip(self.quality_belief + belief_update, 0, 1) # Update word of mouth if total_demand > expected_total * 0.8: self.word_of_mouth = min(1.0, self.word_of_mouth + 0.05) except Exception as e: logger.error(f"Error updating beliefs: {e}") def calculate_comprehensive_costs( self, price: float, strategy: PricingStrategy, demand: float ) -> Dict[str, float]: """ Calculate all cost components. Args: price: Current price strategy: Current pricing strategy demand: Total demand volume Returns: Dictionary of cost components """ try: costs = {} # Production costs costs['production'] = self.market_params.cost_per_unit * demand # Quality signaling costs if self.quality_level == QualityLevel.HIGH and price > self.market_params.cost_per_unit: base_signaling = (price - self.market_params.cost_per_unit) * 0.08 if strategy == PricingStrategy.COMMITMENT: costs['signaling'] = base_signaling * 0.6 elif strategy == PricingStrategy.SPOT: costs['signaling'] = base_signaling * 1.3 else: costs['signaling'] = base_signaling else: costs['signaling'] = 0.0 # Marketing costs if strategy == PricingStrategy.PENETRATION: costs['marketing'] = demand * 2.0 else: costs['marketing'] = demand * 1.0 # Fulfillment costs costs['fulfillment'] = demand * 0.5 return costs except Exception as e: logger.error(f"Error calculating costs: {e}") return { 'production': 0, 'signaling': 0, 'marketing': 0, 'fulfillment': 0 } def optimize_price_advanced(self, strategy: PricingStrategy) -> float: """ Optimize price using multi-objective function. Args: strategy: Current pricing strategy Returns: Optimal price for the strategy """ try: def objective_function(price_array): """Objective function to minimize (negative strategic value)""" price = price_array[0] # Get demand high_demand, low_demand = self.demand_model.estimate_demand( price, self.quality_belief, self.current_period, self.review_system.get_weighted_signal(), self.word_of_mouth ) total_demand = high_demand + low_demand revenue = price * total_demand # Calculate costs costs = self.calculate_comprehensive_costs(price, strategy, total_demand) total_costs = sum(costs.values()) profit = revenue - total_costs # Strategy-specific adjustments if strategy == PricingStrategy.SKIMMING: strategic_value = profit + high_demand * 10 elif strategy == PricingStrategy.PENETRATION: market_share_bonus = total_demand * 5 strategic_value = profit + market_share_bonus elif strategy == PricingStrategy.COMMITMENT: reputation_bonus = (self.quality_belief - 0.5) * total_demand * 3 strategic_value = profit + reputation_bonus else: # SPOT strategic_value = profit return -strategic_value # Set price bounds based on strategy if strategy == PricingStrategy.PENETRATION: min_price = self.market_params.cost_per_unit * 0.9 max_price = self.market_params.cost_per_unit * 1.8 elif strategy == PricingStrategy.SKIMMING: min_price = self.market_params.cost_per_unit * 1.5 max_price = self.market_params.max_price else: min_price = self.market_params.cost_per_unit * 0.95 max_price = self.market_params.max_price bounds = [(min_price, max_price)] initial_guess = [(min_price + max_price) / 2] # Optimize result = minimize( objective_function, initial_guess, bounds=bounds, method='L-BFGS-B' ) if result.success: return result.x[0] else: logger.warning("Optimization failed, using fallback pricing") return self.market_params.cost_per_unit * 1.4 except Exception as e: logger.error(f"Error in price optimization: {e}") return self.market_params.cost_per_unit * 1.4 def select_optimal_strategy(self) -> PricingStrategy: """ Select optimal pricing strategy based on market state. Returns: Optimal pricing strategy for current conditions """ try: # Market maturity market_maturity = min(1.0, self.current_period / 12) # Quality recognition quality_recognition = abs(self.quality_belief - 0.5) * 2 # Review effectiveness review_effectiveness = len(self.review_system.reviews) / max(1, self.current_period) if self.quality_level == QualityLevel.HIGH: if market_maturity < 0.3: # Early stage if quality_recognition < 0.3: return PricingStrategy.COMMITMENT else: return PricingStrategy.SKIMMING else: # Mature stage if review_effectiveness > 0.3: return PricingStrategy.SPOT else: return PricingStrategy.SKIMMING else: # Low quality if market_maturity < 0.5: return PricingStrategy.PENETRATION else: return PricingStrategy.SPOT except Exception as e: logger.error(f"Error selecting strategy: {e}") return PricingStrategy.PENETRATION def step(self) -> Dict: """ Execute one simulation period. Returns: Dictionary of period results """ try: # Select strategy and optimize price strategy = self.select_optimal_strategy() optimal_price = self.optimize_price_advanced(strategy) # Estimate demand high_demand, low_demand = self.demand_model.estimate_demand( optimal_price, self.quality_belief, self.current_period, self.review_system.get_weighted_signal(), self.word_of_mouth ) # Add market noise noise_factor = 0.08 high_demand *= max(0.1, 1 + np.random.normal(0, noise_factor)) low_demand *= max(0.1, 1 + np.random.normal(0, noise_factor)) total_demand = high_demand + low_demand observed_demand = (high_demand, low_demand) # Calculate financials revenue = optimal_price * total_demand costs = self.calculate_comprehensive_costs(optimal_price, strategy, total_demand) total_costs = sum(costs.values()) profit = revenue - total_costs # Generate reviews if total_demand > 0: num_reviews = int(min( total_demand * self.review_system.review_velocity, MAX_REVIEWS_PER_PERIOD )) for i in range(num_reviews): customer_type = "premium" if i < num_reviews * self.consumer_behavior.high_wtp_ratio else "regular" experienced_utility = self.true_quality + np.random.normal(0, 0.1) self.review_system.generate_review( self.true_quality, experienced_utility, customer_type ) # Update beliefs and metrics self.update_beliefs_bayesian(observed_demand, optimal_price) self.total_customers_acquired += total_demand if total_demand > 0: customer_satisfaction = min(1.0, (revenue / total_demand) / optimal_price) self.brand_reputation = 0.9 * self.brand_reputation + 0.1 * customer_satisfaction # Store history self.price_history.append(optimal_price) self.demand_history.append(observed_demand) self.profit_history.append(profit) self.strategy_history.append(strategy.value) self.quality_belief_history.append(self.quality_belief) self.review_signal_history.append(self.review_system.get_weighted_signal()) # Prepare results results = { 'period': self.current_period, 'price': optimal_price, 'strategy': strategy.value, 'high_demand': high_demand, 'low_demand': low_demand, 'total_demand': total_demand, 'revenue': revenue, 'profit': profit, 'total_costs': total_costs, 'production_cost': costs['production'], 'signaling_cost': costs['signaling'], 'marketing_cost': costs['marketing'], 'fulfillment_cost': costs['fulfillment'], 'quality_belief': self.quality_belief, 'review_signal': self.review_system.get_weighted_signal(), 'word_of_mouth': self.word_of_mouth, 'brand_reputation': self.brand_reputation, 'market_share': total_demand / self.market_params.total_consumers, 'customer_acquisition': total_demand, 'cumulative_customers': self.total_customers_acquired } self.current_period += 1 return results except Exception as e: logger.error(f"Error in simulation step: {e}") fallback_price = self.market_params.cost_per_unit * 1.2 fallback_demand = 100 return { 'period': self.current_period, 'price': fallback_price, 'strategy': 'penetration', 'high_demand': fallback_demand * 0.3, 'low_demand': fallback_demand * 0.7, 'total_demand': fallback_demand, 'revenue': fallback_price * fallback_demand, 'profit': fallback_price * fallback_demand * 0.2, 'total_costs': fallback_price * fallback_demand * 0.8, 'quality_belief': 0.5, 'review_signal': 0.5, 'brand_reputation': 0.5, 'market_share': 0.1 } # ============================== # ANALYTICS & VISUALIZATION # ============================== class AdvancedAnalytics: """Analytics computation for simulation results""" @staticmethod def calculate_business_metrics(df: pd.DataFrame, market_params: MarketParameters) -> Dict: """ Calculate comprehensive business metrics. Args: df: Results dataframe market_params: Market parameters Returns: Dictionary of calculated metrics """ try: metrics = {} # Financial metrics metrics['total_revenue'] = df['revenue'].sum() metrics['total_profit'] = df['profit'].sum() metrics['total_units_sold'] = df['total_demand'].sum() metrics['average_price'] = df['price'].mean() metrics['profit_margin'] = ( (metrics['total_profit'] / metrics['total_revenue']) * 100 if metrics['total_revenue'] > 0 else 0 ) metrics['revenue_per_unit'] = ( metrics['total_revenue'] / metrics['total_units_sold'] if metrics['total_units_sold'] > 0 else 0 ) # Pricing metrics metrics['price_volatility'] = ( df['price'].std() / df['price'].mean() if df['price'].mean() > 0 else 0 ) metrics['price_trend'] = ( (df['price'].iloc[-1] - df['price'].iloc[0]) / df['price'].iloc[0] * 100 if len(df) > 1 and df['price'].iloc[0] > 0 else 0 ) metrics['optimal_price_range'] = f"${df['price'].min():.2f} - ${df['price'].max():.2f}" # Market performance metrics['market_penetration'] = ( df['total_demand'].sum() / (market_params.total_consumers * len(df)) ) * 100 metrics['customer_acquisition_rate'] = df['total_demand'].mean() metrics['peak_demand_period'] = int(df['total_demand'].idxmax()) metrics['demand_consistency'] = ( 1 - (df['total_demand'].std() / df['total_demand'].mean()) if df['total_demand'].mean() > 0 else 0 ) # Strategic metrics strategy_distribution = df['strategy'].value_counts(normalize=True) * 100 metrics['primary_strategy'] = df['strategy'].mode().iloc[0] if len(df) > 0 else 'unknown' metrics['strategy_consistency'] = strategy_distribution.max() if len(strategy_distribution) > 0 else 0 # Learning metrics if 'quality_belief' in df.columns and len(df) > 1: metrics['market_learning_speed'] = abs( df['quality_belief'].iloc[-1] - df['quality_belief'].iloc[0] ) metrics['final_quality_perception'] = df['quality_belief'].iloc[-1] else: metrics['market_learning_speed'] = 0 metrics['final_quality_perception'] = 0.5 # Brand metrics if 'brand_reputation' in df.columns and len(df) > 1: metrics['brand_reputation_final'] = df['brand_reputation'].iloc[-1] metrics['brand_growth'] = ( df['brand_reputation'].iloc[-1] - df['brand_reputation'].iloc[0] ) else: metrics['brand_reputation_final'] = 0.5 metrics['brand_growth'] = 0 return metrics except Exception as e: logger.error(f"Error calculating metrics: {e}") return {} @staticmethod def create_comprehensive_dashboard( df: pd.DataFrame, metrics: Dict, true_quality: float ) -> go.Figure: """ Create accessible multi-panel dashboard. Uses Okabe-Ito color-blind safe palette with WCAG AA contrast. Args: df: Results dataframe metrics: Calculated metrics true_quality: True product quality Returns: Plotly figure object """ try: # Create subplots fig = make_subplots( rows=3, cols=2, subplot_titles=( 'Price Strategy Evolution', 'Demand Response Dynamics', 'Revenue and Profit Performance', 'Market Learning Process', 'Strategy Distribution', 'Performance Overview' ), specs=[ [{"secondary_y": False}, {"secondary_y": True}], [{"secondary_y": True}, {"secondary_y": False}], [{"type": "pie"}, {"type": "bar"}] ], vertical_spacing=0.12, horizontal_spacing=0.15 ) # 1. Price Evolution (Okabe-Ito blue) fig.add_trace( go.Scatter( x=df['period'], y=df['price'], mode='lines+markers', name='Price', line=dict(color=COLORS['blue'], width=3), marker=dict(size=8, symbol='circle'), hovertemplate='Period %{x}
Price: $%{y:.2f}' ), row=1, col=1 ) fig.add_hline( y=df['price'].mean(), line_dash="dash", line_color=COLORS['text_secondary'], annotation_text=f"Average: ${df['price'].mean():.2f}", annotation_position="right", row=1, col=1 ) # 2. Demand Dynamics (Okabe-Ito sky blue and orange) fig.add_trace( go.Scatter( x=df['period'], y=df['total_demand'], mode='lines+markers', name='Total Demand', line=dict(color=COLORS['sky_blue'], width=3), marker=dict(size=8, symbol='square'), hovertemplate='Period %{x}
Demand: %{y:.0f}' ), row=1, col=2 ) if 'high_demand' in df.columns: fig.add_trace( go.Scatter( x=df['period'], y=df['high_demand'], mode='lines', name='Premium Segment', line=dict(color=COLORS['orange'], width=2, dash='dot'), hovertemplate='Period %{x}
Premium: %{y:.0f}' ), row=1, col=2, secondary_y=True ) # 3. Revenue & Profit (Okabe-Ito green and vermillion) fig.add_trace( go.Scatter( x=df['period'], y=df['profit'], mode='lines+markers', name='Profit', line=dict(color=COLORS['green'], width=3), marker=dict(size=8, symbol='diamond'), hovertemplate='Period %{x}
Profit: $%{y:.2f}' ), row=2, col=1 ) fig.add_trace( go.Scatter( x=df['period'], y=df['revenue'], mode='lines', name='Revenue', line=dict(color=COLORS['vermillion'], width=2, dash='dash'), hovertemplate='Period %{x}
Revenue: $%{y:.2f}' ), row=2, col=1, secondary_y=True ) # 4. Market Learning (Okabe-Ito purple) if 'quality_belief' in df.columns: fig.add_trace( go.Scatter( x=df['period'], y=df['quality_belief'], mode='lines+markers', name='Market Belief', line=dict(color=COLORS['purple'], width=3), marker=dict(size=8, symbol='triangle-up'), hovertemplate='Period %{x}
Belief: %{y:.2f}' ), row=2, col=2 ) fig.add_hline( y=true_quality, line_dash="dot", line_color=COLORS['green'], annotation_text=f"True Quality: {true_quality:.2f}", annotation_position="right", row=2, col=2 ) # 5. Strategy Distribution (accessible pie chart) if 'strategy' in df.columns: strategy_counts = df['strategy'].value_counts() fig.add_trace( go.Pie( labels=[s.title() for s in strategy_counts.index], values=strategy_counts.values, marker=dict( colors=[ COLORS['blue'], COLORS['orange'], COLORS['green'], COLORS['vermillion'] ], line=dict(color=COLORS['background'], width=2) ), textposition='inside', textinfo='label+percent', hovertemplate='%{label}
%{value} periods (%{percent})' ), row=3, col=1 ) # 6. Performance Overview (bar chart with patterns for accessibility) if metrics: overview_metrics = [ 'Profit Margin', 'Market Penetration', 'Price Stability', 'Brand Strength' ] overview_values = [ min(100, max(0, metrics.get('profit_margin', 0))), min(100, max(0, metrics.get('market_penetration', 0))), min(100, max(0, (1 - metrics.get('price_volatility', 0)) * 100)), min(100, max(0, metrics.get('brand_reputation_final', 0.5) * 100)) ] fig.add_trace( go.Bar( x=overview_metrics, y=overview_values, marker=dict( color=[ COLORS['blue'], COLORS['orange'], COLORS['green'], COLORS['purple'] ], line=dict(color=COLORS['text_primary'], width=1) ), text=[f"{v:.1f}%" for v in overview_values], textposition='outside', hovertemplate='%{x}
Score: %{y:.1f}%' ), row=3, col=2 ) # Update layout for accessibility fig.update_layout( height=900, title={ 'text': "SmartPrice Pro - Comprehensive Pricing Strategy Dashboard", 'x': 0.5, 'xanchor': 'center', 'font': {'size': 20, 'color': COLORS['text_primary']} }, showlegend=True, legend=dict( orientation="v", yanchor="top", y=1, xanchor="right", x=1.15, font=dict(size=11, color=COLORS['text_primary']), bgcolor='rgba(255,255,255,0.8)', bordercolor=COLORS['border'], borderwidth=1 ), template="plotly_white", font=dict(size=11, color=COLORS['text_primary'], family="Arial, sans-serif"), paper_bgcolor=COLORS['background'], plot_bgcolor=COLORS['background'], hoverlabel=dict( bgcolor=COLORS['background'], font_size=12, font_family="Arial, sans-serif", font_color=COLORS['text_primary'] ) ) # Update axes for better readability fig.update_xaxes( showgrid=True, gridwidth=1, gridcolor=COLORS['border'], title_font=dict(size=12, color=COLORS['text_primary']), tickfont=dict(size=10, color=COLORS['text_secondary']) ) fig.update_yaxes( showgrid=True, gridwidth=1, gridcolor=COLORS['border'], title_font=dict(size=12, color=COLORS['text_primary']), tickfont=dict(size=10, color=COLORS['text_secondary']) ) return fig except Exception as e: logger.error(f"Error creating dashboard: {e}") # Fallback simple plot fig = go.Figure() fig.add_trace( go.Scatter( x=df['period'], y=df['price'], mode='lines+markers', name='Price', line=dict(color=COLORS['blue']) ) ) fig.update_layout( title="Pricing Analysis (Simplified)", template="plotly_white", font=dict(color=COLORS['text_primary']) ) return fig # ============================== # REPORTING & EXPORT # ============================== class ReportGenerator: """Generate reports and exports""" @staticmethod def generate_executive_summary( metrics: Dict, df: pd.DataFrame, recommendations: str ) -> str: """ Generate executive summary. Args: metrics: Calculated metrics dictionary df: Results dataframe recommendations: Recommendations text Returns: Formatted markdown summary """ @staticmethod def generate_recommendations( df: pd.DataFrame, metrics: Dict, market_type: str, quality_level: str ) -> str: """ Generate strategic recommendations. Args: df: Results dataframe metrics: Calculated metrics market_type: Market type string quality_level: Quality level string Returns: Formatted recommendations text """ try: recommendations = "\n## Strategic Recommendations\n\n" profit_margin = metrics.get('profit_margin', 0) market_penetration = metrics.get('market_penetration', 0) price_volatility = metrics.get('price_volatility', 0) if market_type == "Mass Market": if quality_level == "High Quality": recommendations += """ ### Premium Mass Market Strategy 1. **Quality Communication**: Your analysis shows strong quality recognition. Invest in marketing that highlights premium features and benefits. 2. **Price Optimization**: Consider dynamic pricing to capture maximum value from different customer segments. 3. **Scale Advantages**: Leverage growing demand to negotiate better supplier terms and reduce costs. 4. **Brand Building**: Focus on building a premium brand image to support higher prices long-term. """ else: recommendations += """ ### Competitive Mass Market Strategy 1. **Cost Leadership**: Focus on operational efficiency to maintain competitive pricing while protecting margins. 2. **Volume Strategy**: Prioritize market share growth through competitive pricing and wide distribution. 3. **Value Proposition**: Emphasize price-performance ratio in all marketing communications. 4. **Efficiency Gains**: Invest in automation and process improvements to reduce costs. """ else: # Niche Market if quality_level == "High Quality": recommendations += """ ### Premium Niche Strategy 1. **Expert Positioning**: Position as the premium solution for discerning customers in your niche. 2. **Customer Success**: Invest heavily in customer success and support to justify premium pricing. 3. **Community Building**: Create exclusive customer communities and experiences. 4. **Innovation Investment**: Use higher margins to fund continued product innovation. """ else: recommendations += """ ### Niche Market Value Strategy 1. **Specialization**: Focus on specific use cases where you can excel and differentiate. 2. **Service Excellence**: Compete on service quality and customer relationships rather than just price. 3. **Partnership Strategy**: Develop strategic partnerships to expand market reach. 4. **Efficiency Focus**: Optimize operations for your specific niche requirements. """ # Performance-based alerts if profit_margin < 15: recommendations += "\n**Profitability Alert**: Consider cost reduction initiatives or price increases to improve margins.\n" if market_penetration < 10: recommendations += "\n**Market Penetration**: Explore new customer acquisition channels and consider more aggressive pricing.\n" if price_volatility > 0.2: recommendations += "\n**Price Stability**: High price volatility may confuse customers. Consider more stable pricing strategies.\n" # Action items recommendations += """ ## Immediate Action Items 1. **Review Pricing Strategy**: Implement the recommended pricing approach for your market type and quality level. 2. **Monitor Metrics**: Track the key performance indicators identified in this analysis. 3. **Customer Feedback**: Gather more customer feedback to validate quality perceptions and price sensitivity. 4. **Competitive Analysis**: Conduct regular competitive pricing analysis to maintain market position. 5. **Process Optimization**: Identify opportunities to reduce costs while maintaining quality. """ return recommendations except Exception as e: logger.error(f"Error generating recommendations: {e}") return "\n## Recommendations\nError generating recommendations. Please review the analysis manually." @staticmethod def create_enhanced_csv(df: pd.DataFrame, metrics: Dict) -> Optional[str]: """ Create enhanced CSV export. Args: df: Results dataframe metrics: Calculated metrics Returns: Path to CSV file or None on error """ try: # Create enhanced dataframe enhanced_df = df.copy() # Add calculated fields enhanced_df['cumulative_profit'] = enhanced_df['profit'].cumsum() enhanced_df['cumulative_revenue'] = enhanced_df['revenue'].cumsum() enhanced_df['profit_margin_pct'] = ( (enhanced_df['profit'] / enhanced_df['revenue'] * 100).fillna(0) ) enhanced_df['price_change_pct'] = enhanced_df['price'].pct_change() * 100 enhanced_df['demand_change_pct'] = enhanced_df['total_demand'].pct_change() * 100 if 'total_costs' in enhanced_df.columns: enhanced_df['roi_pct'] = ( ((enhanced_df['revenue'] - enhanced_df['total_costs']) / enhanced_df['total_costs'] * 100).fillna(0) ) # Add summary row summary_row = pd.Series({ 'period': 'SUMMARY', 'price': enhanced_df['price'].mean(), 'total_demand': enhanced_df['total_demand'].sum(), 'revenue': enhanced_df['revenue'].sum(), 'profit': enhanced_df['profit'].sum(), 'profit_margin_pct': metrics.get('profit_margin', 0), 'cumulative_profit': enhanced_df['profit'].sum(), 'cumulative_revenue': enhanced_df['revenue'].sum() }) enhanced_df = pd.concat([enhanced_df, pd.DataFrame([summary_row])], ignore_index=True) # Create temp file timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') filename = f'smartprice_analysis_{timestamp}.csv' filepath = os.path.join(tempfile.gettempdir(), filename) enhanced_df.to_csv(filepath, index=False) logger.info(f"CSV export created: {filepath}") return filepath except Exception as e: logger.error(f"Error creating CSV: {e}") return None # ============================== # VALIDATION # ============================== def validate_inputs( product_cost: float, market_size: int, premium_customers_pct: float, premium_willingness: float, regular_willingness: float, simulation_periods: int, quality_uncertainty: float ) -> Tuple[bool, str]: """ Validate all user inputs. Args: All simulation input parameters Returns: Tuple of (is_valid, error_message) """ if product_cost <= 0: return False, "Product cost must be greater than zero." if product_cost > 10000: return False, "Product cost seems unusually high. Please verify your input." if market_size <= 0: return False, "Market size must be greater than zero." if market_size > MAX_MARKET_SIZE: return False, f"Market size cannot exceed {MAX_MARKET_SIZE:,} for performance reasons." if premium_customers_pct < 0 or premium_customers_pct > 100: return False, "Premium customers percentage must be between 0 and 100." if premium_willingness <= 0: return False, "Premium customer budget must be greater than zero." if regular_willingness <= 0: return False, "Regular customer budget must be greater than zero." if premium_willingness < product_cost * 0.5: return False, "Premium customer budget seems too low relative to product cost." if regular_willingness > premium_willingness: return False, "Regular customer budget cannot exceed premium customer budget." if simulation_periods < 3: return False, "Simulation must run for at least 3 periods." if simulation_periods > MAX_SIMULATION_PERIODS: return False, f"Simulation cannot exceed {MAX_SIMULATION_PERIODS} periods." if quality_uncertainty < 0 or quality_uncertainty > 1: return False, "Quality uncertainty must be between 0 and 1." return True, "" # ============================== # MAIN SIMULATION FUNCTION # ============================== def run_comprehensive_pricing_simulation( product_cost: float, market_size: int, quality_level: str, market_type: str, premium_customers_pct: float, premium_willingness: float, regular_willingness: float, simulation_periods: int, quality_uncertainty: float ) -> Tuple[go.Figure, str, Optional[str]]: """ Main simulation orchestration function. Args: product_cost: Production cost per unit market_size: Total addressable market size quality_level: "High Quality" or "Standard Quality" market_type: "Mass Market" or "Niche Market" premium_customers_pct: Percentage of premium customers (0-100) premium_willingness: Premium segment maximum price regular_willingness: Regular segment maximum price simulation_periods: Number of periods to simulate quality_uncertainty: Initial market uncertainty (0-1) Returns: Tuple of (plotly_figure, summary_text, csv_filepath) """ try: # Validate inputs is_valid, error_msg = validate_inputs( product_cost, market_size, premium_customers_pct, premium_willingness, regular_willingness, simulation_periods, quality_uncertainty ) if not is_valid: error_text = f""" # Input Validation Error **Error**: {error_msg} Please correct your inputs and try again. ## Input Guidelines - Product cost must be positive and reasonable - Market size should be between 1 and {MAX_MARKET_SIZE:,} - Premium percentage must be 0-100 - Premium budget must exceed regular budget - Premium budget should be at least 50% of product cost - Simulation periods: 3 to {MAX_SIMULATION_PERIODS} - Quality uncertainty: 0 to 1 """ empty_fig = go.Figure() empty_fig.add_annotation( text="Input validation failed
Please check your parameters", x=0.5, y=0.5, showarrow=False, font=dict(size=16, color=COLORS['vermillion']) ) empty_fig.update_layout( xaxis=dict(visible=False), yaxis=dict(visible=False), paper_bgcolor=COLORS['background'] ) return empty_fig, error_text, None # Convert inputs market_type_enum = MarketType.MASS if market_type == "Mass Market" else MarketType.NICHE quality_level_enum = QualityLevel.HIGH if quality_level == "High Quality" else QualityLevel.LOW true_quality = 0.8 if quality_level == "High Quality" else 0.3 # Set up parameters market_params = MarketParameters( market_type=market_type_enum, total_consumers=int(market_size), cost_per_unit=float(product_cost), max_price=max(product_cost * 4, 100), quality_uncertainty=float(quality_uncertainty), learning_rate=0.12 ) consumer_behavior = ConsumerBehavior( high_wtp_ratio=float(premium_customers_pct) / 100, high_wtp_value=float(premium_willingness), low_wtp_value=float(regular_willingness), price_elasticity=-1.3, quality_sensitivity=0.85 ) # Initialize engine logger.info(f"Starting simulation: {simulation_periods} periods, {market_type}, {quality_level}") engine = DynamicPricingEngine( market_params=market_params, consumer_behavior=consumer_behavior, true_quality=true_quality, quality_level=quality_level_enum ) # Run simulation results = [] for period in range(int(simulation_periods)): result = engine.step() results.append(result) # Create dataframe df = pd.DataFrame(results) logger.info(f"Simulation complete: {len(df)} periods") # Calculate metrics metrics = AdvancedAnalytics.calculate_business_metrics(df, market_params) # Generate recommendations recommendations = ReportGenerator.generate_recommendations( df, metrics, market_type, quality_level ) # Create dashboard fig = AdvancedAnalytics.create_comprehensive_dashboard(df, metrics, true_quality) # Generate summary summary_text = ReportGenerator.generate_executive_summary(df, metrics, recommendations) # Create CSV csv_path = ReportGenerator.create_enhanced_csv(df, metrics) return fig, summary_text, csv_path except Exception as e: logger.error(f"Simulation error: {e}", exc_info=True) error_text = f""" # Simulation Error An unexpected error occurred during the simulation. Please try again with different parameters. **Technical details**: {str(e)} ## Troubleshooting Tips 1. Try reducing the market size or simulation periods 2. Verify that all numeric inputs are reasonable 3. Ensure premium budget exceeds regular budget 4. Check that product cost is positive If the problem persists, please contact support with the error details above. """ empty_fig = go.Figure() empty_fig.add_annotation( text="Simulation error occurred
Please check inputs and try again", x=0.5, y=0.5, showarrow=False, font=dict(size=16, color=COLORS['vermillion']) ) empty_fig.update_layout( title="Simulation Error", xaxis=dict(visible=False), yaxis=dict(visible=False), paper_bgcolor=COLORS['background'] ) return empty_fig, error_text, None # ============================== # GRADIO INTERFACE # ============================== def create_advanced_interface(): """ Create production-grade Gradio interface. Returns: Gradio Blocks interface """ # Custom CSS for accessibility and professional appearance # All colors verified for WCAG AA contrast custom_css = """ .gradio-container { font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', 'Roboto', 'Oxygen', 'Ubuntu', 'Cantarell', 'Fira Sans', 'Droid Sans', 'Helvetica Neue', Arial, sans-serif; max-width: 1400px; margin: 0 auto; } .main-header { text-align: center; background: linear-gradient(135deg, #0F62FE 0%, #0072B2 100%); color: #FFFFFF; padding: 2rem; border-radius: 12px; margin-bottom: 2rem; box-shadow: 0 4px 12px rgba(0, 0, 0, 0.15); } .section-header { background: #F8F9FA; padding: 1rem 1.5rem; border-radius: 8px; border-left: 4px solid #0F62FE; margin: 1rem 0; box-shadow: 0 2px 4px rgba(0, 0, 0, 0.05); } .section-header h3 { margin: 0; color: #0B0B0B; font-size: 1.1rem; font-weight: 600; } .info-box { background: #E3F2FD; border: 1px solid #0072B2; border-radius: 8px; padding: 1rem; margin: 1rem 0; color: #0B0B0B; } .info-box h4 { margin-top: 0; color: #0B0B0B; } /* Ensure focus visibility for keyboard navigation */ button:focus, input:focus, select:focus, textarea:focus { outline: 3px solid #0F62FE; outline-offset: 2px; } /* High contrast for disabled elements */ button:disabled { opacity: 0.6; cursor: not-allowed; } """ with gr.Blocks( title="SmartPrice Pro - Advanced Dynamic Pricing Platform", theme=gr.themes.Soft( primary_hue="blue", secondary_hue="gray", neutral_hue="gray", text_size="md", font=[gr.themes.GoogleFont("Inter"), "Arial", "sans-serif"] ), css=custom_css ) as demo: # Header gr.HTML(""" """) with gr.Row(): # Left column - Inputs with gr.Column(scale=1, min_width=400): # Product Configuration with gr.Group(): gr.HTML('

Product Configuration

') product_cost = gr.Slider( minimum=1, maximum=500, value=25, step=1, label="Production Cost per Unit (USD)", info="The cost to produce or acquire each unit of your product", elem_id="product-cost-slider" ) quality_level = gr.Radio( choices=["High Quality", "Standard Quality"], value="High Quality", label="Product Quality Positioning", info="How does your product quality compare to market alternatives?", elem_id="quality-level-radio" ) market_type = gr.Radio( choices=["Mass Market", "Niche Market"], value="Mass Market", label="Market Strategy Focus", info="Mass Market: broad appeal | Niche Market: specialized segments", elem_id="market-type-radio" ) # Market Intelligence with gr.Group(): gr.HTML('

Market Intelligence

') market_size = gr.Slider( minimum=500, maximum=100000, value=10000, step=500, label="Total Addressable Market (TAM)", info="Total number of potential customers in your target market", elem_id="market-size-slider" ) premium_customers_pct = gr.Slider( minimum=5, maximum=80, value=30, step=5, label="Premium Customer Segment (percent)", info="Percentage of customers willing to pay premium prices", elem_id="premium-pct-slider" ) premium_willingness = gr.Slider( minimum=30, maximum=1000, value=150, step=10, label="Premium Segment Budget (USD)", info="Maximum price premium customers are willing to pay", elem_id="premium-budget-slider" ) regular_willingness = gr.Slider( minimum=10, maximum=300, value=75, step=5, label="Regular Segment Budget (USD)", info="Maximum price regular customers are willing to pay", elem_id="regular-budget-slider" ) # Analysis Configuration with gr.Group(): gr.HTML('

Analysis Settings

') simulation_periods = gr.Slider( minimum=6, maximum=36, value=18, step=1, label="Planning Horizon (months)", info="Number of months to analyze for strategic planning", elem_id="periods-slider" ) quality_uncertainty = gr.Slider( minimum=0.1, maximum=0.9, value=0.4, step=0.1, label="Initial Market Uncertainty", info="How uncertain are customers initially about your product quality?", elem_id="uncertainty-slider" ) # Action button analyze_btn = gr.Button( "Generate Comprehensive Analysis", variant="primary", size="lg", elem_id="analyze-button" ) # Quick examples gr.HTML(""" """) # Right column - Results with gr.Column(scale=2, min_width=800): with gr.Group(): gr.HTML('

Strategic Analysis Dashboard

') results_plot = gr.Plot( label="Interactive Analysis Dashboard", show_label=False, elem_id="results-plot" ) # Footer gr.HTML(""" """) # Event handlers def run_with_error_handling(*args): """Wrapper to handle errors gracefully""" try: return run_comprehensive_pricing_simulation(*args) except Exception as e: logger.error(f"Analysis failed: {e}", exc_info=True) error_fig = go.Figure() error_fig.add_annotation( text=f"Analysis Error
{str(e)[:100]}", x=0.5, y=0.5, showarrow=False, font=dict(size=14, color=COLORS['vermillion']) ) error_fig.update_layout( xaxis=dict(visible=False), yaxis=dict(visible=False), paper_bgcolor=COLORS['background'] ) return error_fig, f"**Error**: {str(e)}", None analyze_btn.click( fn=run_with_error_handling, inputs=[ product_cost, market_size, quality_level, market_type, premium_customers_pct, premium_willingness, regular_willingness, simulation_periods, quality_uncertainty ], outputs=[results_plot, summary_text, csv_download], api_name="analyze" ) # Welcome message on load demo.load( fn=lambda: """ # Welcome to SmartPrice Pro Configure your product parameters on the left and click **Generate Comprehensive Analysis** to begin your pricing strategy optimization. ## Getting Started 1. Enter your product cost and quality positioning 2. Define your target market characteristics 3. Specify customer segment budgets 4. Set analysis parameters 5. Click the analysis button to generate insights The platform will simulate pricing dynamics over your planning horizon and provide strategic recommendations. """, outputs=summary_text ) return demo # ============================== # APPLICATION ENTRY POINT # ============================== if __name__ == "__main__": logger.info("Starting SmartPrice Pro application") logger.info(f"Log level: {LOG_LEVEL}") logger.info(f"PDF generation available: {PDF_AVAILABLE}") if RANDOM_SEED: logger.info(f"Running in deterministic mode with seed: {RANDOM_SEED}") try: demo = create_advanced_interface() # Launch for Hugging Face Spaces # HF manages host and port, so we don't hardcode them demo.launch( show_error=False, # Don't show errors in production UI share=False, favicon_path=None, show_api=False ) except Exception as e: logger.critical(f"Failed to start application: {e}", exc_info=True) print(f"\n{'='*60}") print("CRITICAL ERROR: Failed to start SmartPrice Pro") print(f"{'='*60}") print(f"Error: {e}") print("\nPlease check:") print("1. All dependencies are installed (see requirements.txt)") print("2. Python version is 3.10 or higher") print("3. Sufficient system resources are available") print(f"{'='*60}\n")