"""Produces the final product-level score, decision and business summary.""" from __future__ import annotations from dataclasses import dataclass, field from typing import Optional from ...constants import constants as C from ...utils.helpers import clamp, saturating_score, weighted_average from .market_analyzer import MarketAnalysis from .product_analyzer import ProductAnalysis from .risk_analyzer import RiskAnalysis from .store_analyzer import StoreAnalysis from .strength_analyzer import StrengthAnalysis from .sku.sku_decision_engine import SkuDecision @dataclass class FinalRecommendation: decision: str confidence: float overall_score: float summary: str market_recommendation: str sku_recommendation: str score_breakdown: dict[str, float] = field(default_factory=dict) confidence_reasons: list[str] = field(default_factory=list) def to_dict(self) -> dict: return { "decision": self.decision, "confidence": round(self.confidence, 2), "overallScore": round(self.overall_score, 2), "summary": self.summary, "marketRecommendation": self.market_recommendation, "skuRecommendation": self.sku_recommendation, "scoreBreakdown": {k: round(v, 2) for k, v in self.score_breakdown.items()}, "confidenceReasons": self.confidence_reasons, } class RecommendationEngine: """Single responsibility: combine every analysis into the final answer.""" def recommend( self, product: ProductAnalysis, store: StoreAnalysis, sku_decisions: list[SkuDecision], market: MarketAnalysis, risk: RiskAnalysis, strength: StrengthAnalysis, ) -> FinalRecommendation: best = self._best_sku(sku_decisions) breakdown = self._score_breakdown(product, store, sku_decisions, best, market, risk, strength) overall = clamp( sum(breakdown[name] * weight for name, weight in C.PRODUCT_SCORE_WEIGHTS.items()) ) # No sellable SKU means the product cannot be sold, period. if best is None or best.result.decision == C.DECISION_DO_NOT_SELL: overall = min(overall, C.PRODUCT_CONSIDER_FALLBACK_SCORE - 1) confidence, confidence_reasons = self._confidence( product, store, sku_decisions, breakdown ) decision = self._decision(overall, risk.score, confidence) return FinalRecommendation( decision=decision, confidence=confidence, overall_score=overall, summary=self._summary(decision, overall, best, store, risk, strength), market_recommendation=self._market_recommendation(market), sku_recommendation=self._sku_recommendation(best, sku_decisions), score_breakdown=breakdown, confidence_reasons=confidence_reasons, ) # ------------------------------------------------------------------ # @staticmethod def _best_sku(sku_decisions: list[SkuDecision]) -> Optional[SkuDecision]: if not sku_decisions: return None return max(sku_decisions, key=lambda d: d.result.score) @staticmethod def _score_breakdown( product: ProductAnalysis, store: StoreAnalysis, sku_decisions: list[SkuDecision], best: Optional[SkuDecision], market: MarketAnalysis, risk: RiskAnalysis, strength: StrengthAnalysis, ) -> dict[str, float]: product_score = (product.quality_score + product.completeness_score) / 2 average_sku = ( sum(d.result.score for d in sku_decisions) / len(sku_decisions) if sku_decisions else 0.0 ) return { "product": product_score, "store": store.score, "best_sku": best.result.score if best else 0.0, "average_sku": average_sku, "market": market.best_score, "risk": 100.0 - risk.score, # low risk contributes positively "strength": strength.score, } @staticmethod def _decision(score: float, risk_score: float, confidence: float) -> str: """SELL requires high score AND low risk AND reliable confidence. CONSIDER covers 60-79, or 50-59 when risk is still moderate. Everything else is DO_NOT_SELL. """ if ( score >= C.PRODUCT_SELL_SCORE and risk_score < C.PRODUCT_SELL_MAX_RISK and confidence >= C.PRODUCT_SELL_MIN_CONFIDENCE ): return C.DECISION_SELL if score >= C.PRODUCT_CONSIDER_SCORE: return C.DECISION_CONSIDER if ( score >= C.PRODUCT_CONSIDER_FALLBACK_SCORE and risk_score <= C.PRODUCT_CONSIDER_MAX_RISK ): return C.DECISION_CONSIDER return C.DECISION_DO_NOT_SELL @staticmethod def _confidence( product: ProductAnalysis, store: StoreAnalysis, sku_decisions: list[SkuDecision], breakdown: dict[str, float], ) -> tuple[float, list[str]]: """Confidence reflects *data quality*, not the score itself: review volume, sales volume, store reputation, listing completeness, shipping-data reliability and consistency across metrics. Hard caps apply when there are very few reviews.""" weights = C.CONFIDENCE_COMPONENT_WEIGHTS reviews_score = saturating_score(product.review_count, C.CONFIDENCE_REVIEWS_FULL) sales_score = saturating_score(product.sales_count, C.CONFIDENCE_SALES_FULL) store_score = store.score if store.score > 0 else None completeness = ( product.completeness_score if product.completeness_score > 0 else None ) shipping_reliability = None if sku_decisions: with_freight = [d for d in sku_decisions if d.freight.has_data] coverage = 100.0 * len(with_freight) / len(sku_decisions) tracked = ( 100.0 * sum(1 for d in with_freight if d.freight.tracking) / len(with_freight) if with_freight else 0.0 ) shipping_reliability = 0.6 * coverage + 0.4 * tracked # Consistency: tight breakdown components -> reliable picture. values = [v for v in breakdown.values() if v is not None] consistency = None if len(values) >= 2: mean = sum(values) / len(values) spread = (sum((v - mean) ** 2 for v in values) / len(values)) ** 0.5 consistency = clamp(100.0 - spread * 2.0) confidence = weighted_average([ (reviews_score, weights["reviews"]), (sales_score, weights["sales"]), (store_score, weights["store_reputation"]), (completeness, weights["listing_completeness"]), (shipping_reliability, weights["shipping_reliability"]), (consistency, weights["metric_consistency"]), ]) reasons: list[str] = [] few, few_cap = C.CONFIDENCE_CAP_FEW_REVIEWS almost_none, none_cap = C.CONFIDENCE_CAP_ALMOST_NO_REVIEWS if product.review_count < almost_none: confidence = min(confidence, none_cap) reasons.append( f"Almost no reviews ({product.review_count}) limits confidence" ) elif product.review_count < few: confidence = min(confidence, few_cap) reasons.append(f"Few reviews ({product.review_count}) limit confidence") if shipping_reliability is not None and shipping_reliability >= 80: reasons.append("Reliable, tracked shipping data") if (store_score or 0) >= 80: reasons.append("Reputable store") if (completeness or 0) >= 90: reasons.append("Complete listing data") if consistency is not None and consistency < 50: reasons.append("Metrics are inconsistent across factors") return clamp(confidence), reasons # ------------------------------------------------------------------ # @staticmethod def _market_recommendation(market: MarketAnalysis) -> str: recommended = [] if market.uk.recommend: recommended.append("United Kingdom") if market.saudi.recommend: recommended.append("Saudi Arabia") if not recommended: return "Not recommended for the UK or Saudi Arabia at this time." return f"Recommended market(s): {', '.join(recommended)}." @staticmethod def _sku_recommendation(best: Optional[SkuDecision], sku_decisions: list[SkuDecision]) -> str: if best is None: return "No sellable SKU was found." sellable = sum( 1 for d in sku_decisions if d.result.decision == C.DECISION_SELL ) label = " / ".join(filter(None, (best.sku.color, best.sku.size))) or best.sku.sku_id return ( f"Focus on SKU {best.sku.sku_id} ({label}) with a score of " f"{best.result.score:.0f}/100; {sellable} of {len(sku_decisions)} " f"SKUs qualify as SELL." ) def _summary( self, decision: str, score: float, best: Optional[SkuDecision], store: StoreAnalysis, risk: RiskAnalysis, strength: StrengthAnalysis, ) -> str: parts: list[str] = [] if decision == C.DECISION_SELL: parts.append( f"This product is a strong candidate (overall score {score:.0f}/100)." ) elif decision == C.DECISION_CONSIDER: parts.append( f"This product is viable but not a clear winner (overall score " f"{score:.0f}/100); validate the highlighted risks before listing." ) else: parts.append( f"This product should not be sold (overall score {score:.0f}/100)." ) if best is not None: parts.append( f"The best SKU nets an estimated {best.pricing.estimated_profit:.2f} " f"profit per sale at a recommended price of " f"{best.pricing.recommended_selling_price:.2f}." ) parts.append(f"The store is graded {store.grade} ({store.score:.0f}/100).") if strength.strengths: parts.append(f"Key strengths: {'; '.join(strength.strengths[:3])}.") if risk.risks: parts.append(f"Key risks: {'; '.join(risk.risks[:3])}.") return " ".join(parts)