"""Synthetic user events data generator.""" import uuid import random from datetime import datetime, timedelta, date import numpy as np import pandas as pd from faker import Faker from .schema import ( EVENT_TYPES, DEVICE_TYPES, OS_VALUES, BROWSER_VALUES, COUNTRY_CODES, DEVICE_OS_COMPAT, OS_BROWSER_COMPAT, ) fake = Faker() def generate(n_rows: int, seed: int = 42) -> pd.DataFrame: """Generate n_rows of clean synthetic user events.""" rng = np.random.default_rng(seed) random.seed(seed) Faker.seed(seed) # Pre-generate a pool of users and sessions n_users = max(100, n_rows // 50) n_sessions = max(500, n_rows // 10) user_ids = [str(uuid.uuid4()) for _ in range(n_users)] session_ids = [str(uuid.uuid4()) for _ in range(n_sessions)] session_user = {s: random.choice(user_ids) for s in session_ids} # Base timestamp range: last 30 days now = datetime(2024, 3, 1, 12, 0, 0) start = now - timedelta(days=30) rows = [] for _ in range(n_rows): session_id = random.choice(session_ids) user_id = session_user[session_id] device_type = random.choice(DEVICE_TYPES) os = random.choice(DEVICE_OS_COMPAT[device_type]) browser = random.choice(OS_BROWSER_COMPAT[os]) country_code = random.choice(COUNTRY_CODES) event_type = random.choice(EVENT_TYPES) ts = start + timedelta(seconds=int(rng.integers(0, int((now - start).total_seconds())))) is_purchase = event_type == "purchase" is_search = event_type == "search" is_cart = event_type in ("add_to_cart", "purchase", "checkout") row = { "event_id": str(uuid.uuid4()), "user_id": user_id, "session_id": session_id, "event_type": event_type, "timestamp": ts, "page_url": fake.url() + fake.uri_path(), "referrer_url": fake.url() if random.random() > 0.3 else None, "ip_address": fake.ipv4_public(), "device_type": device_type, "os": os, "browser": browser, "country_code": country_code, "city": fake.city(), "duration_seconds": int(rng.integers(1, 3600)), "revenue": round(float(rng.uniform(0.99, 999.99)), 2) if is_purchase else 0.0, "product_id": str(uuid.uuid4()) if is_cart else None, "quantity": int(rng.integers(1, 10)) if is_purchase else 0, "search_query": fake.sentence(nb_words=3) if is_search else None, "user_agent": fake.user_agent(), "is_authenticated": bool(random.random() > 0.2), "user_age": int(rng.integers(18, 80)), "email": fake.email(), "user_registration_date": (ts - timedelta(days=int(rng.integers(1, 1000)))).date(), } rows.append(row) return pd.DataFrame(rows)