End-to-End ML System Design Questions

-- Assume we have a table "cutoffs(user_id, cutoff_date)" listing cutoffs to label for.
-- If you only have a single cutoff, replace cutoffs with a VALUES(...) or a literal date.
WITH purchases_after_cutoff AS (
  SELECT
    c.user_id,
    c.cutoff_date,
    MIN(p.purchase_time) AS first_purchase_after_cutoff
  FROM cutoffs c
  LEFT JOIN purchases p
    ON p.user_id = c.user_id
    AND p.purchase_time > c.cutoff_date
    AND p.purchase_time <= c.cutoff_date + INTERVAL '7' DAY
  GROUP BY c.user_id, c.cutoff_date
)
SELECT
  c.user_id,
  c.cutoff_date,
  CASE WHEN p.first_purchase_after_cutoff IS NOT NULL THEN 1 ELSE 0 END AS label,
  p.first_purchase_after_cutoff
FROM cutoffs c
LEFT JOIN purchases_after_cutoff p
  USING (user_id, cutoff_date);

import time
import math
from collections import defaultdict
from typing import Dict, Callable
import numpy as np
from scipy import stats

class RolloutManager:
    def __init__(self, router_update: Callable[[Dict[str,float]], None],
                 metric_fetch: Callable[[str], Dict[str, list]],
                 slo_thresholds: Dict[str, float],
                 min_samples=100):
        """
        router_update: function to set traffic weights {'green':0.1,'blue':0.9}
        metric_fetch: function that returns recent metric lists for a variant {'latency':[...], 'error_rate':[...]}
        slo_thresholds: allowed upper limits for metrics (e.g., {'error_rate':0.02})
        """
        self.router_update = router_update
        self.metric_fetch = metric_fetch
        self.slo = slo_thresholds
        self.min_samples = min_samples
        self.state = 'idle'
        self.weights = {}

    def ramp(self, from_w: Dict[str,float], to_w: Dict[str,float], duration_s:int, steps:int=10):
        for i in range(1, steps+1):
            frac = i/steps
            w = {k: from_w.get(k,0)*(1-frac) + to_w.get(k,0)*frac for k in set(from_w)|set(to_w)}
            self.weights = w
            self.router_update(w)
            time.sleep(duration_s/steps)

    def aggregate_metrics(self, variant):
        # returns aggregated metrics (mean, std, n)
        m = self.metric_fetch(variant)
        agg = {}
        for k, vals in m.items():
            arr = np.array(vals)
            agg[k] = {'mean': arr.mean() if len(arr) else float('nan'),
                      'std': arr.std(ddof=1) if len(arr)>1 else float('nan'),
                      'n': len(arr)}
        return agg

    def significance_test(self, metric, a_vals, b_vals, alpha=0.05):
        # use Welch's t-test if approx normal, else permutation test (small sample)
        if len(a_vals)>=30 and len(b_vals)>=30:
            t, p = stats.ttest_ind(a_vals, b_vals, equal_var=False, nan_policy='omit')
            return p < alpha, p
        # fallback permutation test
        concat = np.concatenate([a_vals,b_vals])
        obs = abs(np.mean(a_vals)-np.mean(b_vals))
        count=0
        iters=1000
        for _ in range(iters):
            np.random.shuffle(concat)
            if abs(concat[:len(a_vals)].mean()-concat[len(a_vals):].mean())>=obs:
                count+=1
        p = (count+1)/(iters+1)
        return p<alpha, p

    def evaluate(self, baseline='blue', candidate='green'):
        a = self.metric_fetch(baseline)
        b = self.metric_fetch(candidate)
        # early exit if not enough samples
        if sum(len(v) for v in b.values()) < self.min_samples:
            return 'insufficient', {}
        violations = []
        details = {}
        for metric in self.slo:
            a_vals = np.array(a.get(metric,[]))
            b_vals = np.array(b.get(metric,[]))
            better, p = self.significance_test(metric, a_vals, b_vals)
            details[metric] = {'baseline_mean': a_vals.mean() if len(a_vals) else None,
                               'cand_mean': b_vals.mean() if len(b_vals) else None,
                               'p': p}
            # assume lower-is-better metrics (latency, error_rate)
            if (b_vals.mean() > self.slo[metric]) or (b_vals.mean() > a_vals.mean() and p<0.05):
                violations.append(metric)
        return ('promote' if not violations else 'rollback'), details

    def canary_deploy(self, baseline_w, candidate_w, ramp_duration=300):
        # start ramp-up
        self.ramp(baseline_w, candidate_w, duration_s=ramp_duration)
        decision, details = self.evaluate()
        if decision=='promote':
            # move to full candidate
            self.ramp(self.weights, {k: (1.0 if k=='green' else 0.0) for k in self.weights}, duration_s=60)
            return 'promoted', details
        elif decision=='rollback':
            # rollback to baseline
            self.ramp(self.weights, {k: (1.0 if k=='blue' else 0.0) for k in self.weights}, duration_s=60)
            return 'rolled_back', details
        else:
            return 'wait_for_more_data', details