P8 Sessionize events · P9 Consecutive streaks

import json
from collections import defaultdict
from datetime import datetime, timedelta

GAP = timedelta(minutes=30)
by_user = defaultdict(list)
for line in open("events.jsonl"):
    e = json.loads(line)
    ts = datetime.fromisoformat(e["ts"])
    by_user[e["user_id"]].append((ts, e["event"]))

for user, events in sorted(by_user.items()):
    events.sort()                         # never trust input order
    sessions, prev = [], None
    for ts, ev in events:
        if prev is None or ts - prev > GAP:
            sessions.append([])           # start a new session
        sessions[-1].append(ev)
        prev = ts
    print(user, sessions)

1 [['view', 'cart', 'checkout', 'purchase'], ['view']]
2 [['view', 'view'], ['cart']]
3 [['view', 'cart', 'checkout']]
4 [['view']]

import csv
from collections import defaultdict
from datetime import date, timedelta

ONE = timedelta(days=1)
days = defaultdict(set)
for r in csv.DictReader(open("logins.csv")):
    days[int(r["user_id"])].add(date.fromisoformat(r["login_date"]))

for user, ds in sorted(days.items()):
    best = (0, None)
    for d in ds:
        if d - ONE not in ds:             # d starts a streak
            n = 1
            while d + n * ONE in ds:
                n += 1
            best = max(best, (n, d))
    print(user, "longest:", best[0], "days from", best[1])

1 longest: 3 days from 2025-04-01
2 longest: 2 days from 2025-04-01
3 longest: 4 days from 2025-04-03