Spaces:
Running
Running
| import logging | |
| from typing import Any, Dict | |
| from sqlalchemy.orm import Session | |
| from core.models import BusinessProductService | |
| logger = logging.getLogger(__name__) | |
| class DynamicPricingService: | |
| def __init__(self, db: Session): | |
| self.db = db | |
| def get_adjusted_price(self, product_id: str) -> float: | |
| """ | |
| Calculates a dynamic price based on inventory, demand, and margin floors. | |
| """ | |
| product = self.db.query(BusinessProductService).filter(BusinessProductService.id == product_id).first() | |
| if not product: | |
| return 0.0 | |
| base_price = product.base_price | |
| stock = getattr(product, "stock_quantity", 50) # Default to 50 if field missing in some versions | |
| unit_cost = getattr(product, "unit_cost", 0.0) | |
| # Scarcity Multiplier (Low Stock) | |
| if stock > 0 and stock < 10: | |
| adjusted_price = base_price * 1.15 | |
| logger.info(f"Dynamic Pricing: Scarcity adjustment (+15%) for {product_id}. New price: {adjusted_price}") | |
| # Liquidation Multiplier (High Stock) | |
| elif stock > 100: | |
| adjusted_price = base_price * 0.90 | |
| logger.info(f"Dynamic Pricing: Liquidation adjustment (-10%) for {product_id}. New price: {adjusted_price}") | |
| else: | |
| adjusted_price = base_price | |
| # Competitor Matching (if available in metadata) | |
| metadata = product.metadata_json or {} | |
| competitor_price = metadata.get("competitor_price") | |
| if competitor_price: | |
| target_price = competitor_price * 0.98 # Undercut by 2% | |
| if target_price < adjusted_price: | |
| # Only undercut if it doesn't violate margin floor (10% above cost) | |
| margin_floor = unit_cost * 1.10 | |
| if target_price >= margin_floor: | |
| adjusted_price = target_price | |
| logger.info(f"Dynamic Pricing: Competitor undercut (-2% vs {competitor_price}) for {product_id}. New price: {adjusted_price}") | |
| return round(adjusted_price, 2) | |