Analytics Platforms and Dashboards Questions

WITH gsc AS (
  SELECT
    PARSE_DATE('%Y-%m-%d', date) AS date,
    REGEXP_REPLACE(LOWER(REGEXP_REPLACE(page, r'https?://[^/]+', '')), r'\?.*$', '') AS path,
    SUM(clicks) AS clicks,
    SUM(impressions) AS impressions
  FROM `proj.gsc_table`
  GROUP BY date, path
),
sessions AS (
  SELECT
    DATE(event_time) AS date,
    REGEXP_REPLACE(LOWER(REGEXP_REPLACE(page_location, r'https?://[^/]+', '')), r'\?.*$', '') AS path,
    COUNT(DISTINCT session_id) AS sessions,
    SUM(IF(event_name='purchase', value, 0)) AS revenue,
    SUM(IF(event_name='conversion',1,0)) AS conversions
  FROM `proj.events_table`
  GROUP BY date, path
)
SELECT
  g.date, g.path, g.impressions, g.clicks,
  s.sessions, s.conversions, s.revenue,
  SAFE_DIVIDE(s.conversions, s.sessions) AS conv_rate,
  SAFE_DIVIDE(g.clicks, g.impressions) AS search_ctr,
  SAFE_DIVIDE(s.sessions, g.clicks) AS session_per_click
FROM gsc g
LEFT JOIN sessions s
  ON g.date = s.date AND g.path = s.path
ORDER BY g.impressions DESC

import numpy as np
from collections import deque
from statsmodels.tsa.holtwinters import ExponentialSmoothing

class OnlineSeries:
    def __init__(self, seasonality=24):
        self.buf = deque(maxlen=7*seasonality)
        self.seasonality = seasonality
        self.model = None
        self.last_fit = None

    def update(self, point):
        self.buf.append(point)
        if len(self.buf) >= 2*self.seasonality:
            self._fit()
        return self.predict_next()

    def _fit(self):
        try:
            self.model = ExponentialSmoothing(
                np.array(self.buf), seasonal='add', seasonal_periods=self.seasonality
            ).fit(optimized=True)
            self.last_fit = self.model
        except Exception:
            self.model = None

    def predict_next(self):
        if self.model:
            return self.model.forecast(1)[0]
        return np.mean(self.buf) if self.buf else 0

def score_observation(series_obj, observed):
    pred = series_obj.update(observed)
    residual = observed - pred
    # robust sigma using recent MAD
    window = np.array(series_obj.buf)
    mad = np.median(np.abs(window - np.median(window))) or 1e-6
    z = residual / (1.4826 * mad)
    return z, residual, pred

def enrich_and_rank(segments_scores):
    # segments_scores: dict segment->(z,residual,pred)
    # compute contribution = residual * traffic_weight
    contributions = {s: v[1]*v[2] for s,v in segments_scores.items()}  # simplistic
    ranked = sorted(contributions.items(), key=lambda x: -abs(x[1]))
    return ranked

# Example usage in streaming loop:
# for window in stream.windows('1m'):
#   compute conv_rate per segment; call score_observation; aggregate alerts when |z|>3 and multiple segments correlated