from __future__ import annotations import csv from collections import defaultdict, deque from dataclasses import dataclass from datetime import UTC, datetime, timedelta from pathlib import Path import pandas as pd import requests from app.cache import RedisCache from app.config import Settings from app.feature_engineering import KAGGLE_RETAIL_CATEGORICAL_FEATURES, KAGGLE_RETAIL_NUMERIC_FEATURES from app.feature_engineering import add_derived_features from app.modeling import load_model_bundle from app.schemas import ( KagglePricingRequest, KagglePricingResponse, OrderEvent, PricingRequest, PricingResponse, ) @dataclass class PricingRecommendation: response: PricingResponse ml_price: float blended_price: float @dataclass class KagglePricingRecommendation: response: KagglePricingResponse ml_price: float class FlashSaleTracker: def __init__(self, threshold: int, lookback_minutes: int): self.threshold = threshold self.lookback = timedelta(minutes=lookback_minutes) self.events: dict[str, deque[datetime]] = defaultdict(deque) def register(self, event: OrderEvent) -> bool: queue = self.events[event.sku_id] queue.append(event.event_time) self._trim(event.sku_id, event.event_time) return len(queue) >= self.threshold def is_flash_sale(self, sku_id: str, now: datetime | None = None) -> bool: reference_time = now or datetime.now(UTC) self._trim(sku_id, reference_time) return len(self.events[sku_id]) >= self.threshold def recent_event_count(self) -> int: now = datetime.now(UTC) total = 0 for sku_id in list(self.events.keys()): self._trim(sku_id, now) total += len(self.events[sku_id]) return total def flash_sale_skus(self) -> list[str]: now = datetime.now(UTC) active = [] for sku_id in list(self.events.keys()): self._trim(sku_id, now) if len(self.events[sku_id]) >= self.threshold: active.append(sku_id) return active def _trim(self, sku_id: str, reference_time: datetime) -> None: queue = self.events[sku_id] while queue and reference_time - queue[0] > self.lookback: queue.popleft() class CompetitorPriceClient: def __init__(self, settings: Settings): self.settings = settings def get_price(self, sku_id: str, fallback_price: float) -> float: if not self.settings.competitor_api_url: return fallback_price try: response = requests.get( self.settings.competitor_api_url, params={"sku_id": sku_id}, timeout=0.7, ) response.raise_for_status() payload = response.json() return float(payload.get("competitor_price", fallback_price)) except Exception: return fallback_price class PricingEngine: def __init__(self, settings: Settings): if not settings.model_path.exists(): raise FileNotFoundError( f"Model artifact not found at {settings.model_path}. Train the model first." ) self.settings = settings self.bundle = load_model_bundle(settings.model_path) self.dataset_profile = str(self.bundle.get("dataset_profile", "synthetic")) self.pipeline = self.bundle["pipeline"] self.flash_sale_tracker = FlashSaleTracker( threshold=settings.flash_sale_order_threshold, lookback_minutes=settings.flash_sale_lookback_minutes, ) self.competitor_client = CompetitorPriceClient(settings) self.cache = RedisCache(settings.redis_url) def recommend_price(self, request: PricingRequest) -> PricingRecommendation: if self.dataset_profile != "synthetic": raise ValueError( "The loaded model is not compatible with the synthetic pricing request schema." ) frame = pd.DataFrame([request.model_dump()]) live_competitor_price = self.competitor_client.get_price( request.sku_id, request.competitor_price ) frame["competitor_price"] = live_competitor_price enriched = add_derived_features(frame) ml_price = float(self.pipeline.predict(enriched)[0]) blended_price = ( ml_price * self.settings.model_weight + live_competitor_price * self.settings.competitor_weight ) inventory_adjustment = self._inventory_adjustment( inventory_level=request.inventory_level, inventory_days_cover=request.inventory_days_cover, ) demand_adjustment = self._demand_adjustment( units_sold_last_5m=request.units_sold_last_5m, units_sold_last_1h=request.units_sold_last_1h, conversion_rate=request.conversion_rate, is_festival=request.is_festival, ) detected_flash_sale = self.flash_sale_tracker.is_flash_sale(request.sku_id) flash_sale_multiplier = 1.12 if detected_flash_sale else 1.0 candidate_price = blended_price * inventory_adjustment * demand_adjustment candidate_price *= flash_sale_multiplier guardrailed_price = self._apply_guardrails( candidate_price=candidate_price, base_cost=request.base_cost, current_price=request.current_price, ) confidence = self._confidence_score(request, live_competitor_price) reason = self._build_reason( inventory_adjustment=inventory_adjustment, demand_adjustment=demand_adjustment, detected_flash_sale=detected_flash_sale, ) response = PricingResponse( sku_id=request.sku_id, recommended_price=round(guardrailed_price, 2), ml_price=round(ml_price, 2), blended_price=round(blended_price, 2), inventory_adjustment=round(inventory_adjustment, 3), demand_adjustment=round(demand_adjustment, 3), flash_sale_multiplier=round(flash_sale_multiplier, 3), confidence=round(confidence, 3), detected_flash_sale=detected_flash_sale, reason=reason, generated_at=datetime.now(UTC), ) self._append_price_history(response, request.base_cost) return PricingRecommendation( response=response, ml_price=ml_price, blended_price=blended_price, ) def recommend_kaggle_price( self, request: KagglePricingRequest ) -> KagglePricingRecommendation: if self.dataset_profile != "kaggle_retail": raise ValueError( "The loaded model is not compatible with the Kaggle retail pricing request schema." ) payload = request.model_dump() current_price = payload.pop("current_price") feature_frame = pd.DataFrame([payload])[ KAGGLE_RETAIL_NUMERIC_FEATURES + KAGGLE_RETAIL_CATEGORICAL_FEATURES ] ml_price = float(self.pipeline.predict(feature_frame)[0]) competitor_values = [ payload["comp_1"] or 0.0, payload["comp_2"] or 0.0, payload["comp_3"] or 0.0, ] non_zero_competitors = [value for value in competitor_values if value > 0] competitor_anchor = ( sum(non_zero_competitors) / len(non_zero_competitors) if non_zero_competitors else None ) confidence = self._kaggle_confidence_score(request, competitor_anchor) reason = self._build_kaggle_reason(current_price, ml_price, competitor_anchor) response = KagglePricingResponse( product_id=request.product_id, product_category_name=request.product_category_name, recommended_price=round(ml_price, 2), current_price=round(current_price, 2), gap_to_current_price=round(ml_price - current_price, 2), competitor_anchor_price=( round(competitor_anchor, 2) if competitor_anchor is not None else None ), confidence=round(confidence, 3), reason=reason, generated_at=datetime.now(UTC), ) return KagglePricingRecommendation(response=response, ml_price=ml_price) def register_order_event(self, event: OrderEvent) -> bool: if event.event_time.tzinfo is None: event = OrderEvent( sku_id=event.sku_id, quantity=event.quantity, event_time=event.event_time.replace(tzinfo=UTC), ) return self.flash_sale_tracker.register(event) def get_cached_recommendation(self, sku_id: str) -> dict[str, object] | None: return self.cache.get_json(f"price:{sku_id}") def _inventory_adjustment(self, inventory_level: int, inventory_days_cover: float) -> float: if inventory_level <= 15 or inventory_days_cover < 3: return 1.10 if inventory_level >= 120 or inventory_days_cover > 21: return 0.93 return 1.0 def _demand_adjustment( self, units_sold_last_5m: int, units_sold_last_1h: int, conversion_rate: float, is_festival: int, ) -> float: rapid_demand = units_sold_last_5m >= 8 or units_sold_last_1h >= 40 strong_conversion = conversion_rate >= 0.045 if rapid_demand and strong_conversion: return 1.08 + (0.03 if is_festival else 0.0) if units_sold_last_1h <= 8 and conversion_rate < 0.02: return 0.94 return 1.0 def _apply_guardrails(self, candidate_price: float, base_cost: float, current_price: float) -> float: min_price = base_cost * (1.0 + self.settings.min_margin) max_price = current_price * self.settings.max_price_multiplier return max(min(candidate_price, max_price), min_price) def _confidence_score(self, request: PricingRequest, competitor_price: float) -> float: signal_score = min(request.conversion_rate * 12 + request.click_through_rate * 4, 0.45) inventory_score = 0.25 if request.inventory_level > 10 else 0.12 competitor_score = 0.20 if competitor_price > 0 else 0.08 recency_score = 0.10 if request.units_sold_last_1h > 0 else 0.04 return min(signal_score + inventory_score + competitor_score + recency_score, 0.98) def _build_reason( self, inventory_adjustment: float, demand_adjustment: float, detected_flash_sale: bool, ) -> str: reasons = ["ML baseline with competitor blending"] if inventory_adjustment > 1.0: reasons.append("low inventory pressure") elif inventory_adjustment < 1.0: reasons.append("overstock discount") if demand_adjustment > 1.0: reasons.append("strong short-term demand") elif demand_adjustment < 1.0: reasons.append("soft demand correction") if detected_flash_sale: reasons.append("flash sale multiplier active") return ", ".join(reasons) def _kaggle_confidence_score( self, request: KagglePricingRequest, competitor_anchor: float | None, ) -> float: demand_score = min((request.qty / 40) + (request.customers / 80), 0.40) rating_score = min(request.product_score / 10, 0.20) competitor_score = 0.20 if competitor_anchor is not None else 0.08 recency_score = 0.10 if request.lag_price > 0 else 0.04 seasonal_score = 0.10 if request.holiday or request.weekend else 0.05 return min( demand_score + rating_score + competitor_score + recency_score + seasonal_score, 0.97, ) def _build_kaggle_reason( self, current_price: float, ml_price: float, competitor_anchor: float | None, ) -> str: reasons = ["Kaggle retail model baseline"] if competitor_anchor is not None: if ml_price > competitor_anchor: reasons.append("positioned above competitor average") elif ml_price < competitor_anchor: reasons.append("positioned below competitor average") else: reasons.append("aligned with competitor average") if ml_price > current_price: reasons.append("upside versus current price") elif ml_price < current_price: reasons.append("defensive move versus current price") else: reasons.append("flat versus current price") return ", ".join(reasons) def _append_price_history(self, response: PricingResponse, base_cost: float) -> None: path = self.settings.price_history_path path.parent.mkdir(parents=True, exist_ok=True) file_exists = path.exists() with path.open("a", newline="", encoding="utf-8") as handle: writer = csv.writer(handle) if not file_exists: writer.writerow( [ "generated_at", "sku_id", "recommended_price", "ml_price", "blended_price", "confidence", "detected_flash_sale", "base_cost", ] ) writer.writerow( [ response.generated_at.isoformat(), response.sku_id, response.recommended_price, response.ml_price, response.blended_price, response.confidence, int(response.detected_flash_sale), round(base_cost, 2), ] ) self.cache.set_json(f"price:{response.sku_id}", response.model_dump(mode="json"))