Movavi — сессионизация с тремя триггерами разрыва (medium / транзакция / 2 ч)

WITH events_ord AS (
  SELECT
    user_id,
    medium,
    event_name,
    page_location,
    transaction_id,
    to_timestamp,
    LAG(to_timestamp) OVER (PARTITION BY user_id ORDER BY to_timestamp) AS prev_ts,
    LAG(medium)       OVER (PARTITION BY user_id ORDER BY to_timestamp) AS prev_medium,
    LAG(event_name)   OVER (PARTITION BY user_id ORDER BY to_timestamp) AS prev_event
  FROM events
),
flagged AS (
  SELECT
    *,
    -- 1) первое событие юзера = новая сессия
    -- 2) предыдущее = purchase → новая сессия (так как purchase закрывает)
    -- 3) разрыв > 2 часа
    -- 4) сменился medium
    CASE
      WHEN prev_ts IS NULL                                                      THEN 1
      WHEN prev_event = 'purchase'                                              THEN 1
      WHEN EXTRACT(EPOCH FROM (to_timestamp - prev_ts)) > 7200                  THEN 1
      WHEN medium IS DISTINCT FROM prev_medium                                  THEN 1
      ELSE 0
    END AS is_new_session,
 
    -- Тип триггера: нужен, чтобы потом понять, наследовать medium или ставить 'unknown'
    CASE
      WHEN prev_ts IS NULL                                                      THEN 'first'
      WHEN medium IS DISTINCT FROM prev_medium                                  THEN 'medium_change'
      WHEN prev_event = 'purchase'                                              THEN 'after_purchase'
      WHEN EXTRACT(EPOCH FROM (to_timestamp - prev_ts)) > 7200                  THEN 'inactivity'
      ELSE 'continuation'
    END AS trigger_type
  FROM events_ord
),
sessioned AS (
  SELECT
    *,
    SUM(is_new_session) OVER (PARTITION BY user_id ORDER BY to_timestamp
                              ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW) AS session_num
  FROM flagged
),
session_starts AS (
  SELECT
    user_id, session_num,
    MIN(trigger_type) FILTER (WHERE is_new_session = 1) AS start_trigger,
    -- medium на старте сессии: оригинальный medium первого события
    (ARRAY_AGG(medium ORDER BY to_timestamp))[1]        AS first_medium
  FROM sessioned
  GROUP BY user_id, session_num
),
labeled AS (
  SELECT
    s.user_id,
    s.event_name,
    s.page_location,
    s.transaction_id,
    s.to_timestamp,
    -- session_id уникальный
    CONCAT(s.user_id, '_', s.session_num)  AS session_id,
    -- medium сессии: новый только если триггер = смена medium
    CASE
      WHEN ss.start_trigger = 'medium_change' THEN ss.first_medium
      WHEN ss.start_trigger = 'first'         THEN ss.first_medium  -- для первой сессии берём оригинальный
      ELSE 'unknown'
    END AS session_medium
  FROM sessioned s
  JOIN session_starts ss USING (user_id, session_num)
)
SELECT * FROM labeled
ORDER BY user_id, to_timestamp;

import pandas as pd
 
df = pd.read_csv("clickstream.csv", parse_dates=["to_timestamp"])
df = df.sort_values(["user_id", "to_timestamp"]).reset_index(drop=True)
 
g = df.groupby("user_id")
df["prev_ts"]     = g["to_timestamp"].shift(1)
df["prev_medium"] = g["medium"].shift(1)
df["prev_event"]  = g["event_name"].shift(1)
 
gap = (df["to_timestamp"] - df["prev_ts"]).dt.total_seconds()
 
cond_first    = df["prev_ts"].isna()
cond_purchase = df["prev_event"] == "purchase"
cond_inactive = gap > 7200
cond_medium   = df["medium"] != df["prev_medium"]
 
df["trigger_type"] = "continuation"
df.loc[cond_first,    "trigger_type"] = "first"
df.loc[cond_medium & ~cond_first,   "trigger_type"] = "medium_change"
df.loc[cond_purchase & ~cond_first & ~cond_medium, "trigger_type"] = "after_purchase"
df.loc[cond_inactive & ~cond_first & ~cond_medium & ~cond_purchase, "trigger_type"] = "inactivity"
 
df["is_new_session"] = (df["trigger_type"] != "continuation").astype(int)
df["session_num"]    = df.groupby("user_id")["is_new_session"].cumsum()
df["session_id"]     = df["user_id"].astype(str) + "_" + df["session_num"].astype(str)
 
# Эффективный medium сессии
session_meta = (df.groupby(["user_id", "session_num"])
                  .agg(start_trigger=("trigger_type", "first"),
                       first_medium=("medium", "first"))
                  .reset_index())
session_meta["session_medium"] = session_meta.apply(
    lambda r: r["first_medium"]
              if r["start_trigger"] in ("medium_change", "first")
              else "unknown",
    axis=1
)
df = df.merge(session_meta[["user_id", "session_num", "session_medium"]],
              on=["user_id", "session_num"])