Spaces:
Runtime error
Runtime error
| """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) | |