Amazon-Specific ML Applications and Business Contexts Questions

import itertools
import heapq
import random
from typing import List, Tuple

class CountMinSketch:
    def __init__(self, width: int, depth: int, seed: int = 0):
        self.width = width
        self.depth = depth
        self.tables = [[0]*width for _ in range(depth)]
        random.seed(seed)
        self.hash_seeds = [random.randrange(1<<30) for _ in range(depth)]
    def _hash(self, x, i):
        return (hash((x, self.hash_seeds[i])) % self.width)
    def add(self, x, cnt=1):
        for i in range(self.depth):
            self.tables[i][self._hash(x,i)] += cnt
    def estimate(self, x):
        return min(self.tables[i][self._hash(x,i)] for i in range(self.depth))

class TopKPairsStream:
    def __init__(self, k:int, cms_width=2000, cms_depth=5, heavy_size=10000, pair_sample_prob=1.0):
        self.k = k
        self.cms = CountMinSketch(cms_width, cms_depth)
        self.heavy = {}  # pair -> estimated count (space-saving approx)
        self.heavy_size = heavy_size
        self.pair_sample_prob = pair_sample_prob
    def _pair_key(self, a, b):
        return (a,b) if a<=b else (b,a)
    def process_basket(self, basket: List[int]):
        # Optionally sample pairs to reduce cost for large baskets
        items = list(set(basket))
        # generate pairs; for large baskets sample to limit O(|B|^2)
        if len(items) > 200 and self.pair_sample_prob < 1.0:
            # sample a subset of pairs
            pairs = []
            for i in range(len(items)):
                for j in range(i+1, len(items)):
                    if random.random() <= self.pair_sample_prob:
                        pairs.append((items[i], items[j]))
        else:
            pairs = itertools.combinations(items, 2)
        for a,b in pairs:
            key = self._pair_key(a,b)
            self.cms.add(key, 1)
            # update heavy hitters table (space-saving style)
            if key in self.heavy:
                self.heavy[key] += 1
            elif len(self.heavy) < self.heavy_size:
                self.heavy[key] = self.cms.estimate(key)
            else:
                # replace min entry conservatively
                min_key = min(self.heavy, key=lambda x: self.heavy[x])
                min_val = self.heavy[min_key]
                # CMS provides a conservative estimate; we set new entry to min_val+1
                if self.cms.estimate(key) > min_val:
                    del self.heavy[min_key]
                    self.heavy[key] = self.cms.estimate(key)
    def get_topk(self) -> List[Tuple[Tuple[int,int], int]]:
        # return top-k by CMS-estimate among heavy candidates
        heap = []
        for key in self.heavy:
            est = self.cms.estimate(key)
            if len(heap) < self.k:
                heapq.heappush(heap, (est, key))
            else:
                if est > heap[0][0]:
                    heapq.heapreplace(heap, (est, key))
        return sorted([(key, est) for est,key in heap], key=lambda x: -x[1])

WITH recent AS (
  SELECT transaction_id, user_id, amount,
         DATE_TRUNC('week', occurred_at) AS week_start
  FROM transactions
  WHERE occurred_at >= DATEADD(day, -90, CURRENT_DATE)  -- push predicate for pruning
)
SELECT
  user_id,
  week_start,
  SUM(amount)        AS total_spend,
  COUNT(*)           AS num_orders,
  SUM(amount)::float / NULLIF(COUNT(*),0) AS avg_order_value
FROM recent
GROUP BY user_id, week_start;