Customer Focus and Impact Questions

-- find days with zero or anomalously low rows in events table
SELECT date(event_time) AS day,
       COUNT(*) AS rows
FROM events
WHERE event_time BETWEEN '2025-11-01' AND '2025-11-30'
GROUP BY day
ORDER BY day;

-- join against raw event store (staging_raw) to find mismatch
SELECT d.day, COALESCE(r.raw_count,0) raw_count, COALESCE(p.prod_count,0) prod_count
FROM (
  SELECT generate_series('2025-11-01'::date,'2025-11-30'::date,'1 day') AS day
) d
LEFT JOIN (
  SELECT date(event_time) day, count(*) raw_count FROM staging_raw
  WHERE event_time BETWEEN '2025-11-01' AND '2025-11-30' GROUP BY day
) r USING(day)
LEFT JOIN (
  SELECT date(event_time) day, count(*) prod_count FROM events
  WHERE event_time BETWEEN '2025-11-01' AND '2025-11-30' GROUP BY day
) p USING(day)
WHERE prod_count IS NULL OR prod_count < raw_count * 0.9
ORDER BY day;

import time
import threading
from collections import defaultdict

class TokenBucket:
    def __init__(self, capacity, refill_rate):
        self.capacity = float(capacity)
        self.refill_rate = float(refill_rate)  # tokens per second
        self.tokens = float(capacity)
        self.last = time.monotonic()
        self.lock = threading.Lock()

    def allow_request(self, tokens=1.0):
        with self.lock:
            now = time.monotonic()
            elapsed = now - self.last
            # refill
            self.tokens = min(self.capacity, self.tokens + elapsed * self.refill_rate)
            self.last = now
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

class RateLimiter:
    def __init__(self):
        # map customer_id -> TokenBucket
        self.buckets = {}
        self.map_lock = threading.Lock()

    def configure_customer(self, customer_id, capacity, refill_rate):
        with self.map_lock:
            self.buckets[customer_id] = TokenBucket(capacity, refill_rate)

    def allow(self, customer_id, tokens=1.0):
        bucket = self.buckets.get(customer_id)
        if not bucket:
            # default behavior: deny or create default bucket
            return False
        return bucket.allow_request(tokens)

from datetime import datetime, timedelta
from collections import defaultdict

def month_start(dt):
    return datetime(dt.year, dt.month, 1)

def next_month(dt):
    # return first day of next month
    y, m = dt.year, dt.month
    if m == 12:
        return datetime(y+1, 1, 1)
    return datetime(y, m+1, 1)

def parse_time(t):
    # accept datetime or ISO string
    if isinstance(t, datetime):
        return t
    return datetime.fromisoformat(t)

def compute_monthly_churn(subscriptions):
    """
    subscriptions: list of (user_id, event_type, event_time)
    event_type: 'start' or 'cancel'
    event_time: datetime or ISO string
    Returns: dict mapping YYYY-MM (e.g. '2023-07') -> churn_rate (float)
    """
    # normalize and sort events
    events = [(uid, etype, parse_time(t)) for uid, etype, t in subscriptions]
    if not events:
        return {}
    events.sort(key=lambda x: x[2])
    
    # calendar range
    first_month = month_start(events[0][2])
    last_month = month_start(events[-1][2])
    
    # pointers and state
    i = 0
    n = len(events)
    active = set()
    result = {}
    cur_month = first_month
    while cur_month <= last_month:
        start = cur_month
        end = next_month(cur_month)
        # apply events strictly before month start to build active-at-start
        while i < n and events[i][2] < start:
            uid, etype, _ = events[i]
            if etype == 'start':
                active.add(uid)
            elif etype == 'cancel':
                active.discard(uid)
            i += 1
        active_at_start = len(active)
        # count cancellations during the month; also apply events in-month for next months
        cancels = 0
        j = i
        while j < n and events[j][2] < end:
            uid, etype, _ = events[j]
            if etype == 'cancel':
                cancels += 1
                active.discard(uid)
            elif etype == 'start':
                active.add(uid)
            j += 1
        # churn definition: cancellations during month / active users at start of that month
        churn = 0.0
        if active_at_start > 0:
            churn = cancels / active_at_start
        # store with YYYY-MM key
        key = f"{start.year:04d}-{start.month:02d}"
        result[key] = churn
        # advance pointer and month
        i = j
        cur_month = end
    return result

import heapq
import itertools
from typing import Iterable, Iterator, Dict, Any

def merge_event_streams(stream_a: Iterable[Dict[str, Any]],
                        stream_b: Iterable[Dict[str, Any]]
                       ) -> Iterator[Dict[str, Any]]:
    """
    Merge two iterables of events sorted by 'timestamp' into a single stream.
    Each event is a dict with at least 'user_id' and 'timestamp'.
    Preserves stable ordering when timestamps are equal using a sequence counter.
    """
    iterators = [iter(stream_a), iter(stream_b)]
    heap = []
    seq = itertools.count()  # tie-breaker to preserve arrival order

    # initialize heap with first item from each stream (if any)
    for idx, it in enumerate(iterators):
        try:
            ev = next(it)
            heapq.heappush(heap, (ev['timestamp'], next(seq), idx, ev))
        except StopIteration:
            continue

    while heap:
        timestamp, _, idx, ev = heapq.heappop(heap)
        yield ev
        try:
            nxt = next(iterators[idx])
            heapq.heappush(heap, (nxt['timestamp'], next(seq), idx, nxt))
        except StopIteration:
            continue