Model Monitoring and Observability Questions

import random
import math

class StreamingKSTest:
    def __init__(self, capacity=1000):
        # capacity per reservoir
        self.capacity = capacity
        self.reservoirs = {
            'baseline': {'items': [], 'count': 0},
            'current':  {'items': [], 'count': 0}
        }

    def ingest(self, window_label, value):
        assert window_label in self.reservoirs
        r = self.reservoirs[window_label]
        r['count'] += 1
        n = r['count']
        if len(r['items']) < self.capacity:
            r['items'].append(value)
        else:
            # reservoir sampling: replace with probability capacity/n
            s = random.randint(1, n)
            if s <= self.capacity:
                r['items'][s-1] = value

    def ks_statistic(self):
        a = sorted(self.reservoirs['baseline']['items'])
        b = sorted(self.reservoirs['current']['items'])
        if not a or not b:
            return None  # insufficient data
        i = j = 0
        na, nb = len(a), len(b)
        cdf_a = cdf_b = 0.0
        max_diff = 0.0
        while i < na and j < nb:
            if a[i] <= b[j]:
                cdf_a = (i+1)/na
                i += 1
            else:
                cdf_b = (j+1)/nb
                j += 1
            max_diff = max(max_diff, abs(cdf_a - cdf_b))
        # consume remaining
        while i < na:
            cdf_a = (i+1)/na; i += 1
            max_diff = max(max_diff, abs(cdf_a - cdf_b))
        while j < nb:
            cdf_b = (j+1)/nb; j += 1
            max_diff = max(max_diff, abs(cdf_a - cdf_b))
        return max_diff

WITH
-- replace window bounds as needed
baseline AS (
  SELECT feature_value, COUNT(*)::float AS cnt
  FROM predictions
  WHERE event_time >= '2025-10-01' AND event_time < '2025-11-01'
  GROUP BY 1
),
current AS (
  SELECT feature_value, COUNT(*)::float AS cnt
  FROM predictions
  WHERE event_time >= '2025-11-01' AND event_time < '2025-12-01'
  GROUP BY 1
),
totals AS (
  SELECT
    COALESCE(b.feature_value, c.feature_value) AS feature_value,
    COALESCE(b.cnt, 0) AS baseline_cnt,
    COALESCE(c.cnt, 0) AS current_cnt
  FROM baseline b
  FULL OUTER JOIN current c USING (feature_value)
),
props AS (
  SELECT
    feature_value,
    baseline_cnt,
    current_cnt,
    -- safe proportions with small floor epsilon
    CASE WHEN SUM(baseline_cnt) OVER() = 0 THEN 0
         ELSE baseline_cnt / NULLIF(SUM(baseline_cnt) OVER(),0) END AS p_raw,
    CASE WHEN SUM(current_cnt) OVER() = 0 THEN 0
         ELSE current_cnt / NULLIF(SUM(current_cnt) OVER(),0) END AS q_raw
  FROM totals
),
eps AS (
  -- choose small epsilon, e.g., 1e-6
  SELECT
    feature_value,
    baseline_cnt, current_cnt,
    GREATEST(p_raw, 1e-6) AS p,
    GREATEST(q_raw, 1e-6) AS q
  FROM props
)
SELECT
  feature_value,
  baseline_cnt, current_cnt,
  p AS baseline_prop,
  q AS current_prop,
  (p - q) * LN(p / q) AS psi_contribution
FROM eps
UNION ALL
SELECT
  'TOTAL' AS feature_value, NULL, NULL, NULL, NULL, SUM((p - q) * LN(p / q)) 
FROM eps;

# inputs: X_base (n x d), X_curr (m x d), alpha=0.05
# step 1: optional dim reduction (PCA or random proj)
def preprocess(X, method='pca', k=128):
    if method=='pca':
        return PCA(n_components=k).fit_transform(X)
    if method=='random':
        R = normal(size=(X.shape[1], k)) / sqrt(k)
        return X.dot(R)

# step 2: approximate MMD via Random Fourier Features (RFF)
def rff_features(X, D=1024, gamma=1.0):
    W = normal(scale=sqrt(2*gamma), size=(X.shape[1], D))
    b = uniform(0, 2*pi, size=D)
    Z = sqrt(2.0/D) * cos(X.dot(W) + b)
    return Z

def mmd_rff(Z1, Z2):
    mu1 = Z1.mean(axis=0)
    mu2 = Z2.mean(axis=0)
    return np.sum((mu1 - mu2)**2)

# main
X_base_p = preprocess(X_base, method='pca', k=128)
X_curr_p = preprocess(X_curr, method='pca', k=128)
Zb = rff_features(X_base_p, D=1024, gamma=gamma)
Zc = rff_features(X_curr_p, D=1024, gamma=gamma)
stat = mmd_rff(Zb, Zc)
# compute threshold via permutation (or precompute null distribution)
p_value = permutation_test_mmd(Zb, Zc, n_perm=200)
if p_value < alpha: alert_drift()