Сессии в SQL: как нарезать события по паузе в 30 минут

WITH ev AS (              -- 1. журнал без дублей доставки
  SELECT DISTINCT client_id, event_time, event_name
  FROM app_events
  WHERE client_id = '1f327ac0cb49'
), marked AS (            -- 2. флаг: пауза больше 30 минут или первое событие
  SELECT client_id, event_time,
    CASE WHEN event_time - LAG(event_time) OVER (PARTITION BY client_id ORDER BY event_time)
              <= INTERVAL '30 minutes' THEN 0 ELSE 1 END AS is_new
  FROM ev
), numbered AS (          -- 3. накопительная сумма флагов = номер сессии
  SELECT client_id, event_time,
    SUM(is_new) OVER (PARTITION BY client_id ORDER BY event_time) AS session_no
  FROM marked
)
SELECT session_no, MIN(event_time) AS started, COUNT(*) AS events,
  EXTRACT(EPOCH FROM MAX(event_time) - MIN(event_time)) AS seconds
FROM numbered
GROUP BY client_id, session_no
ORDER BY session_no;

WITH ev AS (
  SELECT DISTINCT client_id, event_time, event_name
  FROM app_events
  WHERE client_id = '1f327ac0cb49'
), gaps AS (
  SELECT client_id, event_time, event_name,
    event_time - LAG(event_time) OVER (PARTITION BY client_id ORDER BY event_time) AS gap
  FROM ev
), marked AS (
  SELECT client_id, event_time, event_name, gap,
    CASE WHEN gap <= INTERVAL '30 minutes' THEN 0 ELSE 1 END AS is_new
  FROM gaps
), numbered AS (
  SELECT event_time, event_name, gap, is_new,
    SUM(is_new) OVER (PARTITION BY client_id ORDER BY event_time) AS session_no
  FROM marked
)
SELECT event_time, event_name,
  ROUND(EXTRACT(EPOCH FROM gap) / 60, 1) AS pause_minutes, is_new, session_no
FROM numbered
WHERE is_new = 1 OR gap > INTERVAL '5 minutes'   -- фильтр после окон
ORDER BY event_time;

WITH ev AS (              -- события с запасом в сутки по краям периода
  SELECT DISTINCT client_id, event_time, event_name
  FROM app_events
  WHERE event_time >= TIMESTAMP '2026-07-19'
    AND event_time <  TIMESTAMP '2026-07-24'
), marked AS (
  SELECT client_id, event_time,
    CASE WHEN event_time - LAG(event_time) OVER (PARTITION BY client_id ORDER BY event_time)
              <= INTERVAL '30 minutes' THEN 0 ELSE 1 END AS is_new
  FROM ev
), numbered AS (
  SELECT client_id, event_time,
    SUM(is_new) OVER (PARTITION BY client_id ORDER BY event_time) AS session_no
  FROM marked
), sess AS (              -- одна строка — одна сессия
  SELECT client_id, session_no, MIN(event_time) AS started, COUNT(*) AS events,
    EXTRACT(EPOCH FROM MAX(event_time) - MIN(event_time)) AS seconds
  FROM numbered
  GROUP BY client_id, session_no
)
SELECT CAST(started AS DATE) AS day, COUNT(*) AS sessions,
  ROUND(AVG(events), 2) AS avg_events, ROUND(AVG(seconds)) AS avg_seconds,
  COUNT(*) FILTER (WHERE events = 1) AS single_event
FROM sess
WHERE started >= TIMESTAMP '2026-07-20'   -- период — по началу сессии
  AND started <  TIMESTAMP '2026-07-23'
GROUP BY CAST(started AS DATE)
ORDER BY day;

WITH ev AS (
  SELECT DISTINCT client_id, event_time, event_name
  FROM app_events
  WHERE event_time >= TIMESTAMP '2026-07-19'
    AND event_time <  TIMESTAMP '2026-07-24'
), gaps AS (              -- пауза считается один раз
  SELECT client_id, event_time,
    event_time - LAG(event_time) OVER (PARTITION BY client_id ORDER BY event_time) AS gap
  FROM ev
), thresholds AS (
  SELECT 15 AS threshold_min UNION ALL SELECT 30 UNION ALL SELECT 60
), marked AS (            -- каждая строка — в трёх копиях, по одной на порог
  SELECT t.threshold_min, g.client_id, g.event_time,
    CASE WHEN g.gap <= t.threshold_min * INTERVAL '1 minute' THEN 0 ELSE 1 END AS is_new
  FROM gaps AS g
  CROSS JOIN thresholds AS t
), numbered AS (
  SELECT threshold_min, client_id, event_time,
    SUM(is_new) OVER (PARTITION BY threshold_min, client_id ORDER BY event_time) AS session_no
  FROM marked
), sess AS (
  SELECT threshold_min, client_id, session_no, MIN(event_time) AS started,
    EXTRACT(EPOCH FROM MAX(event_time) - MIN(event_time)) AS seconds
  FROM numbered
  GROUP BY threshold_min, client_id, session_no
)
SELECT threshold_min, COUNT(*) AS sessions, ROUND(AVG(seconds)) AS avg_seconds,
  percentile_cont(0.5) WITHIN GROUP (ORDER BY seconds) AS median_seconds
FROM sess
WHERE started >= TIMESTAMP '2026-07-20'
  AND started <  TIMESTAMP '2026-07-23'
GROUP BY threshold_min
ORDER BY threshold_min;

WITH ev AS (
  SELECT DISTINCT client_id, event_time, event_name
  FROM app_events
  WHERE event_time >= TIMESTAMP '2026-07-19'
    AND event_time <  TIMESTAMP '2026-07-24'
), gaps AS (
  SELECT event_time, event_name,
    event_time - LAG(event_time) OVER (PARTITION BY client_id ORDER BY event_time) AS gap
  FROM ev
)
SELECT
  CASE WHEN gap <= INTERVAL '1 minute'   THEN '1. до 1 минуты'
       WHEN gap <= INTERVAL '5 minutes'  THEN '2. 1–5 минут'
       WHEN gap <= INTERVAL '15 minutes' THEN '3. 5–15 минут'
       WHEN gap <= INTERVAL '30 minutes' THEN '4. 15–30 минут'
       WHEN gap <= INTERVAL '60 minutes' THEN '5. 30–60 минут'
       ELSE '6. больше часа или нет предыдущего' END AS pause_bucket,
  COUNT(*) AS events,
  COUNT(*) FILTER (WHERE event_name = 'select_fare') AS before_select_fare,
  COUNT(*) FILTER (WHERE event_name IN ('app_open', 'push_open')) AS before_open
FROM gaps
WHERE event_time >= TIMESTAMP '2026-07-20'
  AND event_time <  TIMESTAMP '2026-07-23'
GROUP BY 1
ORDER BY 1;

WITH src AS (
  SELECT 'как в журнале' AS variant, client_id, event_time
  FROM app_events
  WHERE event_time >= TIMESTAMP '2026-07-19'
    AND event_time <  TIMESTAMP '2026-07-24'
  UNION ALL
  SELECT 'после DISTINCT', client_id, event_time
  FROM (
    SELECT DISTINCT client_id, event_time, event_name
    FROM app_events
    WHERE event_time >= TIMESTAMP '2026-07-19'
      AND event_time <  TIMESTAMP '2026-07-24'
  ) AS d
), marked AS (
  SELECT variant, client_id, event_time,
    CASE WHEN event_time - LAG(event_time) OVER (PARTITION BY variant, client_id ORDER BY event_time)
              <= INTERVAL '30 minutes' THEN 0 ELSE 1 END AS is_new
  FROM src
), numbered AS (
  SELECT variant, client_id, event_time,
    SUM(is_new) OVER (PARTITION BY variant, client_id ORDER BY event_time) AS session_no
  FROM marked
), sess AS (
  SELECT variant, client_id, session_no, MIN(event_time) AS started, COUNT(*) AS events
  FROM numbered
  GROUP BY variant, client_id, session_no
)
SELECT variant, COUNT(*) AS sessions, SUM(events) AS events,
  COUNT(*) FILTER (WHERE events = 1) AS single_event
FROM sess
WHERE started >= TIMESTAMP '2026-07-20'
  AND started <  TIMESTAMP '2026-07-23'
GROUP BY variant
ORDER BY variant;

WITH ev AS (              -- события ровно за период: начала сессий отрезаны
  SELECT DISTINCT client_id, event_time, event_name
  FROM app_events
  WHERE event_time >= TIMESTAMP '2026-07-20'
    AND event_time <  TIMESTAMP '2026-07-23'
), marked AS (
  SELECT event_name,
    CASE WHEN event_time - LAG(event_time) OVER (PARTITION BY client_id ORDER BY event_time)
              <= INTERVAL '30 minutes' THEN 0 ELSE 1 END AS is_new
  FROM ev
)
SELECT COUNT(*) AS sessions,   -- число сессий — число флагов начала
  COUNT(*) FILTER (WHERE event_name NOT IN ('app_open', 'push_open')) AS not_from_open
FROM marked
WHERE is_new = 1;

WITH ev AS (
  SELECT DISTINCT client_id, event_time, event_name
  FROM app_events
  WHERE event_time >= TIMESTAMP '2026-07-19'
    AND event_time <  TIMESTAMP '2026-07-24'
), gaps AS (
  SELECT event_time,
    event_time - LAG(event_time) OVER (PARTITION BY client_id ORDER BY event_time) AS gap
  FROM ev
)
SELECT
  COUNT(*) FILTER (WHERE gap IS NULL OR gap > INTERVAL '30 minutes') AS correct,
  COUNT(*) FILTER (WHERE gap > INTERVAL '30 minutes') AS without_first,
  COUNT(*) FILTER (WHERE gap IS NULL OR EXTRACT(MINUTE FROM gap) > 30) AS minute_part
FROM gaps
WHERE event_time >= TIMESTAMP '2026-07-20'
  AND event_time <  TIMESTAMP '2026-07-23';

SELECT
  date_diff('minute', TIMESTAMP '2026-07-10 07:37:55', TIMESTAMP '2026-07-10 08:15:12') AS boundaries,
  date_sub('minute', TIMESTAMP '2026-07-10 07:37:55', TIMESTAMP '2026-07-10 08:15:12') AS full_minutes;

import pandas as pd

ev = pd.read_parquet('app_events.parquet', columns=['client_id', 'event_time', 'event_name'])
ev['event_time'] = pd.to_datetime(ev['event_time'])
# запас в сутки с каждой стороны периода 20–22 июля
ev = ev.loc[ev.event_time.ge('2026-07-19') & ev.event_time.lt('2026-07-24')]
ev = ev.drop_duplicates().sort_values(['client_id', 'event_time'])

gap = ev.groupby('client_id')['event_time'].diff()
ev['is_new'] = (gap.isna() | (gap > pd.Timedelta('30min'))).astype(int)
ev['session_no'] = ev.groupby('client_id')['is_new'].cumsum()

sess = ev.groupby(['client_id', 'session_no'])['event_time'].agg(started='min', ended='max', events='count')
sess = sess.loc[sess.started.ge('2026-07-20') & sess.started.lt('2026-07-23')]
seconds = (sess.ended - sess.started).dt.total_seconds()
print(len(sess), round(sess.events.mean(), 2), round(seconds.mean()))

# без isna(): сравнение с NaT даёт False, первое событие клиента не открывает сессию
in_period = ev.event_time.ge('2026-07-20') & ev.event_time.lt('2026-07-23')
print(int((gap > pd.Timedelta('30min'))[in_period].sum()))

WITH ev AS (
  SELECT DISTINCT client_id, event_time, event_name,
    CASE WHEN event_name = 'app_open' THEN properties::json ->> 'source' END AS source
  FROM app_events
  WHERE client_id = '1f327ac0cb49'
), marked AS (
  SELECT client_id, event_time, event_name, source,
    CASE WHEN event_time - LAG(event_time) OVER (PARTITION BY client_id ORDER BY event_time)
              <= INTERVAL '30 minutes' THEN 0 ELSE 1 END AS is_new
  FROM ev
), numbered AS (
  SELECT client_id, event_time, event_name, source,
    SUM(is_new) OVER (PARTITION BY client_id ORDER BY event_time) AS session_no
  FROM marked
)
SELECT session_no, MIN(event_time) AS started, MAX(source) AS source,
  MAX(CASE WHEN event_name = 'purchase' THEN 1 ELSE 0 END) AS has_purchase
FROM numbered
GROUP BY client_id, session_no
ORDER BY session_no;