import collections from datetime import datetime, timedelta from typing import Dict, Any, List, Optional from data_fetcher import DataFetcher from core.logger_config import get_logger from .constants import CONCENTRATION_WINDOW_DAYS from .helpers import ( entropy_from_counts, hhi_from_counts, weighted_recency, math_confidence, ) logger = get_logger(__name__) class StatisticalAnalyzer: """ Computes mathematical indicators of user behavior. Strictly deterministic and descriptive (Entropy, HHI, Variance, Ratios). """ def __init__(self, db_session=None): self.db_session = db_session self.fetcher = DataFetcher(db_session) if db_session else None def analyze(self, profile_json: Dict[str, Any]) -> Dict[str, Any]: signals = profile_json.get("signals", {}) order_signals = signals.get("order", {}) activity_signals = signals.get("activity", {}) impression_signals = signals.get("impression", {}) metadata = profile_json.get("metadata", {}) user_id = metadata.get("user_id") logger.info(f"Running Statistical Analysis for User {user_id}") cart_signals = signals.get("cart", {}) add_to_cart_stats = cart_signals.get("add_to_cart_stats") or {} concentration = self._analyze_concentration(order_signals, user_id) stats = { "temporal": self._analyze_temporal_rhythm(activity_signals), "concentration": concentration, "stability": self._analyze_stability(order_signals), "engagement": self._analyze_engagement(order_signals, activity_signals, impression_signals), "recency": weighted_recency(order_signals.get("velocity", {}).get("last_order_days_ago")), "confidence": self._calculate_math_confidence(profile_json, concentration), "cart": { "add_to_cart_events_per_item_ordered": add_to_cart_stats.get("add_to_cart_events_per_item_ordered", 0.0), "count_distinct_items_added_but_never_ordered": add_to_cart_stats.get("count_distinct_items_added_but_never_ordered", 0), "count_distinct_items_added_then_removed_from_cart": add_to_cart_stats.get("count_distinct_items_added_then_removed_from_cart", 0), "total_add_to_cart_events_in_window": add_to_cart_stats.get("total_add_to_cart_events_in_window", 0), "total_items_ordered_in_window": add_to_cart_stats.get("total_items_ordered_in_window", 0), }, } logger.info(f"Statistical Analysis Complete. Confidence: {stats['confidence']}") return stats def _analyze_temporal_rhythm(self, activity: Dict[str, Any]) -> Dict[str, Any]: hour_dist = activity.get("temporal_distribution", {}).get("hour", {}) total_events = activity.get("total_events", 0) if total_events == 0 or not hour_dist: return {"dominant_hour_share": 0, "entropy": 0} max_hour_val = max(hour_dist.values()) dominant_share = round(max_hour_val / total_events, 2) ent = entropy_from_counts(hour_dist, total_events) result = {"dominant_hour_share": dominant_share, "entropy": ent} logger.info(f"Temporal Analysis: {result}") return result def _analyze_concentration(self, orders: Dict[str, Any], user_id: Optional[int] = None) -> Dict[str, Any]: rest_hhi = 0.0 item_hhi = 0.0 if self.fetcher and user_id: full_data = self._fetch_full_concentration_data(user_id) total_orders = full_data["total_orders"] if total_orders > 0: rest_hhi = hhi_from_counts(full_data["restaurant_counts"], total_orders) total_items = full_data["total_items"] if total_items > 0: item_hhi = hhi_from_counts(full_data["item_counts"], total_items) else: top_rests = orders.get("loyalty", {}).get("top_restaurants", []) total_orders = orders.get("metrics", {}).get("total_orders", 0) if total_orders > 0 and top_rests: rest_counts = {r[0]: r[1] for r in top_rests} rest_hhi = hhi_from_counts(rest_counts, total_orders) top_items = orders.get("inventory", {}).get("top_items", []) total_items = orders.get("inventory", {}).get("total_items_ordered", 0) if total_items > 0 and top_items: item_counts = {i[0]: i[1] for i in top_items} item_hhi = hhi_from_counts(item_counts, total_items) result = {"restaurant_hhi": rest_hhi, "item_hhi": item_hhi} logger.info(f"Concentration Analysis (HHI): {result}") return result def _fetch_full_concentration_data(self, user_id: int) -> Dict[str, Any]: end_date = datetime.now() start_date = end_date - timedelta(days=CONCENTRATION_WINDOW_DAYS) str_start = start_date.strftime("%Y-%m-%d %H:%M:%S") str_end = end_date.strftime("%Y-%m-%d %H:%M:%S") orders = self.fetcher.fetch_orders(user_id, str_start, str_end) order_ids = [o.get("id") for o in orders if o.get("id")] items = self.fetcher.fetch_order_items(order_ids) rest_counts = collections.Counter([o.get("restaurant_name") for o in orders if o.get("restaurant_name")]) item_counts = collections.Counter( [i.get("menu_name") or i.get("item_name") for i in items if (i.get("menu_name") or i.get("item_name"))] ) return { "restaurant_counts": dict(rest_counts), "item_counts": dict(item_counts), "total_orders": len(orders), "total_items": len(items), } def _analyze_stability(self, orders: Dict[str, Any]) -> Dict[str, Any]: velocity = orders.get("velocity", {}) result = { "avg_order_gap": velocity.get("avg_days_between_orders"), "order_gap_std_dev": velocity.get("std_dev_days_between_orders"), } logger.info(f"Stability Analysis: {result}") return result def _analyze_engagement( self, orders: Dict[str, Any], activity: Dict[str, Any], impressions: Dict[str, Any], ) -> Dict[str, Any]: total_orders = orders.get("metrics", {}).get("total_orders", 0) total_impr = impressions.get("total_impressions", 0) searches = activity.get("top_searches", []) total_searches = sum(s[1] for s in searches) result = { "browse_to_order_ratio": round(total_orders / total_impr, 4) if total_impr > 0 else 0, "search_to_order_ratio": round(total_searches / total_orders, 2) if total_orders > 0 else 0, } logger.info(f"Engagement Analysis: {result}") return result def _calculate_math_confidence( self, profile: Dict[str, Any], concentration: Dict[str, Any] ) -> float: total_orders = profile["signals"]["order"]["metrics"].get("total_orders", 0) if total_orders == 0: return 0.0 return math_confidence(total_orders, concentration["restaurant_hhi"])