User Behavior and Analytics Questions

import numpy as np
import pandas as pd
from multiprocessing import Pool
from functools import partial

def retention_curve_for_cohort(user_first_day, activity_days, cohort_day, max_day=30):
    """
    Build a binary matrix: rows=users in cohort, cols=0..max_day indicating activity on day t (t days since first_day).
    Inputs:
    - user_first_day: Series indexed by user_id -> first_day (datetime or int)
    - activity_days: DataFrame with columns ['user_id', 'activity_day'] (datetime or int)
    - cohort_day: value matching first_day to select cohort
    Returns:
    - users: np.array of user_ids in cohort
    - activity_matrix: boolean ndarray shape (n_users, max_day+1)
    """
    users = user_first_day[user_first_day == cohort_day].index.values
    if len(users) == 0:
        return users, np.zeros((0, max_day+1), dtype=bool)
    # compute days_since_first per activity
    act = activity_days[activity_days['user_id'].isin(users)].copy()
    # assume activity_day and user_first_day in same scale (e.g., integer day index)
    act = act.merge(user_first_day.rename('first_day'), left_on='user_id', right_index=True)
    act['day_since'] = act['activity_day'] - act['first_day']
    act = act[(act['day_since'] >= 0) & (act['day_since'] <= max_day)]
    # pivot to boolean matrix
    idx_map = {u:i for i,u in enumerate(users)}
    mat = np.zeros((len(users), max_day+1), dtype=bool)
    for _, row in act[['user_id','day_since']].drop_duplicates().iterrows():
        mat[idx_map[row['user_id']], int(row['day_since'])] = True
    return users, mat

def bootstrap_retention(mat, B=1000, seed=0):
    """
    mat: boolean ndarray (n_users, days)
    returns: point_estimate (days,), ci_lower (days,), ci_upper (days,)
    """
    n, d = mat.shape
    if n == 0:
        return np.zeros(d), np.zeros(d), np.zeros(d)
    rng = np.random.default_rng(seed)
    boot = np.empty((B, d), dtype=float)
    for b in range(B):
        # sample user indices with replacement
        idx = rng.integers(0, n, n)
        sample = mat[idx].any(axis=1) if False else mat[idx]  # nothing special; mat[idx] shapes (n,d)
        # retention per day = fraction of sampled users active that day
        boot[b] = sample.sum(axis=0) / n
    point = mat.sum(axis=0) / n
    lower = np.percentile(boot, 2.5, axis=0)
    upper = np.percentile(boot, 97.5, axis=0)
    return point, lower, upper

def process_cohort(cohort_day, user_first_day, activity_days, max_day=30, B=1000, seed_base=0):
    users, mat = retention_curve_for_cohort(user_first_day, activity_days, cohort_day, max_day)
    return cohort_day, bootstrap_retention(mat, B=B, seed=seed_base + hash(cohort_day) % (2**31))

# Example orchestration for many cohorts with multiprocessing
def compute_all_cohorts(user_first_day, activity_days, cohorts, max_day=30, B=1000, n_jobs=8):
    func = partial(process_cohort, user_first_day=user_first_day, activity_days=activity_days,
                   max_day=max_day, B=B, seed_base=42)
    with Pool(n_jobs) as p:
        results = p.map(func, cohorts)
    # results: list of (cohort_day, (point, lower, upper))
    return {c:res for c,res in results}

-- count potential duplicates per user/event/timestamp bucket
SELECT user_id, event_name, event_ts, COUNT(*) AS cnt
FROM events
GROUP BY user_id, event_name, event_ts
HAVING COUNT(*) > 1;

SELECT payload_hash, COUNT(*) FROM events GROUP BY payload_hash HAVING COUNT(*)>1;

SELECT COUNT(*) FROM events WHERE event_ts IS NULL OR event_ts = 0;

SELECT COUNT(*) FROM events WHERE event_ts > CURRENT_TIMESTAMP OR event_ts < '2000-01-01';

WITH client AS (
  SELECT date_trunc('hour', event_ts) hr, COUNT(*) c FROM client_events GROUP BY 1
),
server AS (
  SELECT date_trunc('hour', event_ts) hr, COUNT(*) s FROM server_events GROUP BY 1
)
SELECT c.hr, c.c, s.s, (c.c - COALESCE(s.s,0)) diff
FROM client FULL OUTER JOIN server USING (hr)
ORDER BY hr;

SELECT hr, diff, ROUND(100.0*ABS(diff)/GREATEST(c,1),2) pct_diff FROM ... WHERE ABS(diff)/GREATEST(c,1) > 0.05;

WITH events AS (
  SELECT session_id, user_id, event, step, timestamp
  FROM tracking_events
  WHERE event IN ('step_started','step_completed')
)
SELECT
  step,
  COUNT(DISTINCT CASE WHEN event='step_started' THEN session_id||'_'||user_id END) as started,
  COUNT(DISTINCT CASE WHEN event='step_completed' THEN session_id||'_'||user_id END) as completed,
  1.0 * COUNT(DISTINCT CASE WHEN event='step_completed' THEN session_id||'_'||user_id END) /
    NULLIF(COUNT(DISTINCT CASE WHEN event='step_started' THEN session_id||'_'||user_id END),0) as conversion
FROM events
GROUP BY step
ORDER BY step;

import numpy as np

def bootstrap_ab(control, variant, n_boot=20000, seed=0):
    """
    control, variant: 1D arrays of 0/1 outcomes
    n_boot: number of bootstrap resamples
    Returns: dict with obs_diff, ci_lower, ci_upper, p_value
    """
    rng = np.random.default_rng(seed)
    control = np.asarray(control)
    variant = np.asarray(variant)
    n_c, n_v = len(control), len(variant)

    obs_diff = variant.mean() - control.mean()

    # bootstrap samples of difference in conversion rates
    diffs = np.empty(n_boot)
    for i in range(n_boot):
        c_sample = rng.choice(control, size=n_c, replace=True)
        v_sample = rng.choice(variant, size=n_v, replace=True)
        diffs[i] = v_sample.mean() - c_sample.mean()

    # 95% percentile CI
    ci_lower, ci_upper = np.percentile(diffs, [2.5, 97.5])

    # p-value: two-sided using bootstrap distribution centered at 0 (null of no difference)
    # shift diffs so null is zero by subtracting observed difference, then compute extreme fraction
    diffs_centered = diffs - obs_diff
    p_value = np.mean(np.abs(diffs_centered) >= np.abs(obs_diff))

    return {
        "obs_diff": obs_diff,
        "ci": (ci_lower, ci_upper),
        "p_value": p_value,
        "boot_diffs": diffs  # optional for diagnostics
    }