feat(generator): добавлена модель связанного визита

- Зачем:
  - генератор должен создавать учебно полезный визит с общим click_id, правдоподобным путём и честным тиковым бюджетом событий.
- Что:
  - добавлена марковская цепочка страниц, запланированные метки времени и потолок GEN_MAX_SESSION_EVENTS.
  - добавлен набор тикового батча из нескольких визитов до рассчитанного бюджета событий.
  - обновлены тесты, README, KNOWN_ISSUES и статусы задач 01/02.
- Проверка:
  - uv run --with pytest --with-requirements generator/requirements.txt pytest generator/tests -q

generator/KNOWN_ISSUES.md

@@ -5,6 +5,13 @@
 > Витрины и Superset-дашборд строятся на **статическом сиде** (`data/*.jsonl`).
 > Причина — ниже. Это не «сырой код по мелочи», а концептуальный дефект
 > генеративной модели, который надо осознанно чинить перед использованием.
+>
+> **Обновление (2026-06-11):** первый срез модели визита реализован в
+> `generate_batch()`: один публичный вызов строит один `click_id` с несколькими
+> событиями, марковским путём по страницам и запланированными строго растущими
+> метками времени. Остальные пункты ниже остаются полезным историческим
+> контекстом и списком следующих шагов: популяция возвращающихся пользователей,
+> межсессионные паузы и полноценное состояние активных визитов ещё не закрыты.
 
 Заметка написана при дизайне Superset-дашборда (ветка
 `docs/advanced-clickstream-course`): разбирались, почему на дашборде
@@ -16,12 +23,13 @@
 
 - **Модель интенсивности потока (сколько событий и когда) — нормальная.**
   Poisson по тикам + дневной коэффициент + jitter. Её можно оставить.
-- **Генеративная модель сущностей (кто, какая сессия, какое событие) — неверная.**
-  На каждое событие штампуется свежий `click_id`, а атрибуты копируются из
-  случайной сид-строки. Это уничтожает понятие визита/сессии.
-- **Вывод:** прежде чем использовать генератор как источник, переделать именно
-  модель сущностей (иерархия пользователь → сессия → событие). Математику
-  интенсивности трогать не обязательно.
+- **Генеративная модель сущностей исправляется по шагам.** Срез одного визита
+  уже не штампует свежий `click_id` на каждое событие, но полная иерархия
+  пользователь → несколько визитов → события ещё требует популяции
+  возвращающихся пользователей.
+- **Вывод:** прежде чем использовать генератор как полноценный источник,
+  доделать оставшиеся уровни модели сущностей. Математику интенсивности
+  трогать не обязательно.
 
 ## Доменная модель (как задумано в DDL)
 
@@ -50,10 +58,10 @@ user_domain_id (постоянный пользователь, cookie)
 
 Это адекватная модель *интенсивности во времени*. Претензий к ней нет.
 
-## Корневой дефект: модель сущностей в `generate_batch()`
+## Исторический корневой дефект: модель сущностей в `generate_batch()`
 
-Файл `generator/generator.py`, `generate_batch()` (стр. ~262–330). На каждое
-событие в батче:
+До среза от 2026-06-11 файл `generator/generator.py`, `generate_batch()` на
+каждое событие в батче делал примерно следующее:
 
 ```python
 base_browser = self.rng.choice(self.dictionary.browser_events)  # случайная сид-строка

generator/README.md

@@ -1,9 +1,10 @@
 # Генератор событий (MVP rev5)
 
-> ⚠️ **Перед использованием как источник витрин — прочитать
-> [KNOWN_ISSUES.md](./KNOWN_ISSUES.md).** Генеративная модель сущностей неверна
-> (свежий `click_id` на каждое событие ломает семантику визита/сессии); ветка
-> не влита в `main` именно поэтому. Математику интенсивности это не затрагивает.
+> Перед использованием как источник витрин прочитать
+> [KNOWN_ISSUES.md](./KNOWN_ISSUES.md): часть старого дефекта уже исправлена
+> (один `click_id` на визит, путь по страницам, монотонное время), но популяция
+> возвращающихся пользователей и полноценное состояние активных визитов ещё
+> остаются следующими шагами.
 
 Автономный генератор событий для Kafka с режимом `steady-stream`.
 
@@ -19,13 +20,17 @@ generator-service -> Kafka topics -> (потребители отдельно)
 
 - Публикуем постепенно, **короткими тиками** (по умолчанию каждые 5 секунд)
 - На каждом тике отправляем небольшую порцию сообщений
-- Держим целевую интенсивность `events/min` без крупных минутных batch
+- Держим целевую интенсивность `events/min` без крупных минутных batch: тик
+  набирается из одного или нескольких визитов до рассчитанного бюджета событий
 - Распределяем события по 4 топикам:
   - `browser_events`
   - `location_events`
   - `device_events`
   - `geo_events`
-- Сохраняем связи `event_id <-> location`, `click_id <-> device/geo`
+- Публичный вызов генеративного ядра строит один визит: общий `click_id`,
+  разные `event_id`, общий device/geo-контекст, путь по страницам воронки и
+  строго растущие запланированные `event_timestamp`.
+- Сохраняем связи `event_id <-> location`, `click_id <-> device/geo`.
 
 ## Конфигурация (env)
 
@@ -37,6 +42,7 @@ generator-service -> Kafka topics -> (потребители отдельно)
 | `GEN_JITTER_PCT` | Процент вариативности | `20` |
 | `GEN_MIN_EVENTS_PER_TICK` | Минимум событий за тик | `5` |
 | `GEN_MAX_EVENTS_PER_TICK` | Максимум событий за тик | `50` |
+| `GEN_MAX_SESSION_EVENTS` | Потолок длины одного визита, защита от петель | `30` |
 | `GEN_DATA_DIR` | Путь к JSONL файлам | `/data` |
 | `GEN_SEED` | Сид для воспроизводимости | — |
 | `GEN_ENABLED` | Включить генерацию | `true` |

generator/generator.py

@@ -15,7 +15,7 @@ import sys
 import time
 import uuid
 from dataclasses import dataclass, field
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from pathlib import Path
 from typing import Any
 
@@ -71,6 +71,62 @@ METRICS_LAST_SUCCESS = Gauge(
 )
 
 
+PAGE_START_DISTRIBUTION = [
+    ("/home", 0.59),
+    ("/product_a", 0.20),
+    ("/product_b", 0.14),
+    ("/cart", 0.04),
+    ("/payment", 0.02),
+    ("/confirmation", 0.01),
+]
+
+PAGE_TRANSITIONS = {
+    "/home": [
+        ("/home", 0.40),
+        ("/product_a", 0.28),
+        ("/product_b", 0.18),
+        ("/cart", 0.04),
+        (None, 0.10),
+    ],
+    "/product_a": [
+        ("/home", 0.30),
+        ("/product_a", 0.16),
+        ("/product_b", 0.18),
+        ("/cart", 0.27),
+        (None, 0.09),
+    ],
+    "/product_b": [
+        ("/home", 0.32),
+        ("/product_a", 0.15),
+        ("/product_b", 0.18),
+        ("/cart", 0.27),
+        (None, 0.08),
+    ],
+    "/cart": [
+        ("/home", 0.20),
+        ("/product_a", 0.12),
+        ("/product_b", 0.10),
+        ("/cart", 0.10),
+        ("/payment", 0.42),
+        (None, 0.06),
+    ],
+    "/payment": [
+        ("/home", 0.12),
+        ("/cart", 0.24),
+        ("/payment", 0.12),
+        ("/confirmation", 0.38),
+        (None, 0.14),
+    ],
+    "/confirmation": [
+        ("/home", 0.35),
+        ("/product_a", 0.15),
+        ("/product_b", 0.10),
+        ("/confirmation", 0.05),
+        (None, 0.35),
+    ],
+}
+
+
 # ---------------------------------------------------------------------------
 # Конфигурация через env
 # ---------------------------------------------------------------------------
@@ -99,6 +155,9 @@ class Config:
     max_events_per_tick: int = field(
         default_factory=lambda: int(os.getenv("GEN_MAX_EVENTS_PER_TICK", "50"))
     )
+    max_session_events: int = field(
+        default_factory=lambda: int(os.getenv("GEN_MAX_SESSION_EVENTS", "30"))
+    )
 
     # Пути к данным
     data_dir: Path = field(
@@ -136,6 +195,8 @@ class Config:
             raise ValueError("GEN_TICK_SECONDS must be >= 1")
         if self.lambda_base_per_min < 1:
             raise ValueError("GEN_LAMBDA_BASE_PER_MIN must be >= 1")
+        if self.max_session_events < 1:
+            raise ValueError("GEN_MAX_SESSION_EVENTS must be >= 1")
         if not self.data_dir.exists():
             raise ValueError(f"Data directory does not exist: {self.data_dir}")
 
@@ -220,6 +281,44 @@ class EventGenerator:
         """Возвращает текущую метку времени в формате JSONL."""
         return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S.%f")
 
+    def _format_timestamp(self, timestamp: datetime) -> str:
+        """Форматирует запланированную метку времени для JSONL."""
+        return timestamp.strftime("%Y-%m-%d %H:%M:%S.%f")
+
+    def _weighted_choice(self, choices: list[tuple[Any, float]]) -> Any:
+        """Разыгрывает значение по списку весов."""
+        point = self.rng.random()
+        cumulative = 0.0
+        for value, weight in choices:
+            cumulative += weight
+            if point < cumulative:
+                return value
+        return choices[-1][0]
+
+    def _generate_visit_path(self, max_events: int, min_events: int = 1) -> list[str]:
+        """Генерирует путь визита по страницам с защитой от бесконечных петель."""
+        page = self._weighted_choice(PAGE_START_DISTRIBUTION)
+        path = []
+
+        while page is not None and len(path) < max_events:
+            path.append(page)
+            page = self._weighted_choice(PAGE_TRANSITIONS[page])
+
+        while len(path) < min_events and len(path) < max_events:
+            transitions = [
+                (next_page, weight)
+                for next_page, weight in PAGE_TRANSITIONS[path[-1]]
+                if next_page is not None
+            ]
+            path.append(self._weighted_choice(transitions))
+
+        return path
+
+    def _visit_pause_seconds(self) -> float:
+        """Разыгрывает паузу между событиями визита."""
+        pause = self.rng.lognormvariate(math.log(45.0), 1.0)
+        return max(1.0, min(pause, 29 * 60.0))
+
     def _hour_factor(self) -> float:
         """Возвращает коэффициент интенсивности в зависимости от часа дня."""
         hour = datetime.now(timezone.utc).hour
@@ -264,7 +363,7 @@ class EventGenerator:
 
     def generate_batch(self, batch_size: int) -> dict[str, list[dict]]:
         """
-        Генерирует батч событий с сохранением связей.
+        Генерирует один визит с сохранением связей.
 
         Возвращает словарь {topic: [events]}
         """
@@ -287,38 +386,44 @@ class EventGenerator:
         if batch_size <= 0:
             return batch
 
+        max_visit_events = min(batch_size, self.config.max_session_events)
+        min_visit_events = min(2, max_visit_events)
+        visit_path = self._generate_visit_path(max_visit_events, min_visit_events)
+
         visit_candidates = [
             click_id for click_id, browser_events in self.dictionary.browser_by_click_id.items()
             if (
-                len(browser_events) >= batch_size
+                len(browser_events) >= len(visit_path)
                 and click_id in self.dictionary.device_by_click_id
                 and click_id in self.dictionary.geo_by_click_id
                 and all(
                     event["event_id"] in self.dictionary.location_by_event_id
-                    for event in browser_events[:batch_size]
+                    for event in browser_events[:len(visit_path)]
                 )
             )
         ]
         if visit_candidates:
             base_click_id = self.rng.choice(visit_candidates)
-            base_browser_events = self.dictionary.browser_by_click_id[base_click_id][:batch_size]
+            base_browser_events = self.dictionary.browser_by_click_id[base_click_id][:len(visit_path)]
         else:
             # Крайний случай для очень малого сида: сохраняем форму визита,
             # даже если приходится брать события с повторением.
             base_browser = self.rng.choice(self.dictionary.browser_events)
             base_click_id = base_browser["click_id"]
-            base_browser_events = [base_browser for _ in range(batch_size)]
+            base_browser_events = [base_browser for _ in range(len(visit_path))]
 
         base_device = self.dictionary.device_by_click_id.get(base_click_id)
         base_geo = self.dictionary.geo_by_click_id.get(base_click_id)
         new_click_id = self._new_uuid()
 
-        for base_browser in base_browser_events:
+        planned_timestamp = datetime.now(timezone.utc)
+
+        for base_browser, page_url_path in zip(base_browser_events, visit_path):
             base_location = self.dictionary.location_by_event_id.get(base_browser["event_id"])
 
             # Генерируем новые ID
             new_event_id = self._new_uuid()
-            new_timestamp = self._current_timestamp()
+            new_timestamp = self._format_timestamp(planned_timestamp)
 
             # Создаём новое браузерное событие
             browser_event = {
@@ -334,6 +439,8 @@ class EventGenerator:
                 location_event = {
                     **base_location,
                     "event_id": new_event_id,
+                    "page_url": f"http://www.dummywebsite.com{page_url_path}",
+                    "page_url_path": page_url_path,
                 }
                 batch["location_events"].append(location_event)
 
@@ -353,8 +460,38 @@ class EventGenerator:
                 }
                 batch["geo_events"].append(geo_event)
 
+            planned_timestamp += timedelta(seconds=self._visit_pause_seconds())
+
         return batch
 
+    def generate_tick_batch(self, event_budget: int) -> dict[str, list[dict]]:
+        """
+        Генерирует батч тика из одного или нескольких визитов.
+
+        `generate_batch()` остаётся публичным срезом одного визита. Для тика
+        нужно набрать рассчитанный бюджет событий, поэтому здесь несколько
+        визитов объединяются в один набор записей для публикации.
+        """
+        tick_batch = {
+            "browser_events": [],
+            "location_events": [],
+            "device_events": [],
+            "geo_events": [],
+        }
+
+        remaining_events = event_budget
+        while remaining_events > 0:
+            visit_batch = self.generate_batch(remaining_events)
+            generated_events = len(visit_batch["browser_events"])
+            if generated_events == 0:
+                break
+
+            for topic, events in visit_batch.items():
+                tick_batch[topic].extend(events)
+            remaining_events -= generated_events
+
+        return tick_batch
+
 
 # ---------------------------------------------------------------------------
 # Batch record для истории
@@ -923,9 +1060,9 @@ class GeneratorService:
                     events_count = self.generator._calculate_events_count()
                     logger.info(f"Generating ~{events_count} base events")
 
-                    # Генерируем батч
+                    # Генерируем тиковый батч из одного или нескольких визитов
                     gen_start = time.time()
-                    batch = self.generator.generate_batch(events_count)
+                    batch = self.generator.generate_tick_batch(events_count)
                     gen_duration = time.time() - gen_start
 
                     # Публикуем в Kafka

generator/tests/test_config.py

@@ -23,6 +23,12 @@ class TestConfigValidation:
         with pytest.raises(ValueError, match="GEN_LAMBDA_BASE_PER_MIN"):
             replace(base_config, lambda_base_per_min=0)
 
+    def test_max_session_events_must_be_positive(self, base_config):
+        """max_session_events должен быть >= 1."""
+        from dataclasses import replace
+        with pytest.raises(ValueError, match="GEN_MAX_SESSION_EVENTS"):
+            replace(base_config, max_session_events=0)
+
     def test_data_dir_must_exist(self, base_config):
         """data_dir должен существовать."""
         from dataclasses import replace

generator/tests/test_generation.py

@@ -3,12 +3,40 @@
 """
 
 import uuid
+from dataclasses import replace
 from datetime import datetime
 
 import pytest
 from generator import EventGenerator, EventDictionary
 
 
+ALLOWED_PAGE_PATHS = {
+    "/home",
+    "/product_a",
+    "/product_b",
+    "/cart",
+    "/payment",
+    "/confirmation",
+}
+
+
+def _parse_event_timestamps(batch):
+    return [
+        datetime.fromisoformat(event["event_timestamp"].replace(" ", "T"))
+        for event in batch["browser_events"]
+    ]
+
+
+def _page_path_visits(generator, visits_count: int):
+    return [
+        [
+            event["page_url_path"]
+            for event in generator.generate_batch(30)["location_events"]
+        ]
+        for _ in range(visits_count)
+    ]
+
+
 class TestEventGeneration:
     """Тесты генерации событий."""
 
@@ -51,15 +79,28 @@ class TestEventGeneration:
         assert "device_events" in batch
         assert "geo_events" in batch
 
-    def test_generate_batch_size(self, event_dictionary, base_config):
-        """Батч содержит правильное количество событий."""
+    def test_generate_batch_respects_requested_visit_budget(self, event_dictionary, base_config):
+        """Визит не превышает запрошенный бюджет событий."""
         generator = EventGenerator(event_dictionary, base_config)
         batch = generator.generate_batch(10)
 
-        assert len(batch["browser_events"]) == 10
-        assert len(batch["location_events"]) == 10
-        assert len(batch["device_events"]) == 10
-        assert len(batch["geo_events"]) == 10
+        browser_count = len(batch["browser_events"])
+
+        assert 1 <= browser_count <= 10
+        assert len(batch["location_events"]) == browser_count
+        assert len(batch["device_events"]) == browser_count
+        assert len(batch["geo_events"]) == browser_count
+
+    def test_generate_tick_batch_fills_requested_event_budget(self, event_dictionary, base_config):
+        """Тиковый батч набирает целевой бюджет из одного или нескольких визитов."""
+        generator = EventGenerator(event_dictionary, base_config)
+
+        batch = generator.generate_tick_batch(20)
+
+        assert len(batch["browser_events"]) == 20
+        assert len(batch["location_events"]) == 20
+        assert len(batch["device_events"]) == 20
+        assert len(batch["geo_events"]) == 20
 
     def test_generate_batch_creates_one_connected_visit(self, event_dictionary, base_config):
         """Публичный вызов генератора создаёт один связанный визит."""
@@ -98,6 +139,17 @@ class TestEventGeneration:
         assert all(context == device_context[0] for context in device_context)
         assert all(context == geo_context[0] for context in geo_context)
 
+    def test_generate_batch_creates_multi_event_visit_when_budget_allows(
+        self, event_dictionary, base_config
+    ):
+        """Минимальный связанный визит не схлопывается в одно событие."""
+        config = replace(base_config, seed=2)
+        generator = EventGenerator(event_dictionary, config)
+
+        batch = generator.generate_batch(3)
+
+        assert len(batch["browser_events"]) >= 2
+
     def test_event_ids_are_new_uuids(self, event_dictionary, base_config):
         """event_id и click_id — новые UUID, не из оригинальных данных."""
         generator = EventGenerator(event_dictionary, base_config)
@@ -126,6 +178,106 @@ class TestEventGeneration:
         dt = datetime.fromisoformat(timestamp.replace(" ", "T"))
         assert dt.year >= 2024
 
+    def test_visit_event_timestamps_strictly_increase(self, event_dictionary, base_config):
+        """Время событий внутри одного визита строго возрастает."""
+        generator = EventGenerator(event_dictionary, base_config)
+        batch = generator.generate_batch(8)
+
+        timestamps = _parse_event_timestamps(batch)
+
+        assert all(
+            previous < current
+            for previous, current in zip(timestamps, timestamps[1:])
+        )
+
+    def test_visit_event_timestamps_use_planned_user_pauses(self, event_dictionary, base_config):
+        """Метки времени визита разделены пользовательскими паузами, а не временем цикла."""
+        generator = EventGenerator(event_dictionary, base_config)
+        batch = generator.generate_batch(8)
+
+        timestamps = _parse_event_timestamps(batch)
+        pauses_seconds = [
+            (current - previous).total_seconds()
+            for previous, current in zip(timestamps, timestamps[1:])
+        ]
+
+        assert min(pauses_seconds) >= 1.0
+
+    def test_visit_path_uses_known_funnel_pages(self, event_dictionary, base_config):
+        """Путь визита состоит из страниц воронки."""
+        generator = EventGenerator(event_dictionary, base_config)
+        batch = generator.generate_batch(30)
+
+        page_paths = [event["page_url_path"] for event in batch["location_events"]]
+
+        assert page_paths
+        assert set(page_paths) <= ALLOWED_PAGE_PATHS
+
+    def test_visit_starts_from_calibrated_start_distribution(self, event_dictionary, base_config):
+        """Визиты стартуют не только с /home, а по стартовому распределению."""
+        generator = EventGenerator(event_dictionary, base_config)
+
+        visits = _page_path_visits(generator, 1000)
+        first_pages = [visit[0] for visit in visits]
+        home_share = first_pages.count("/home") / len(first_pages)
+        product_entry_share = (
+            first_pages.count("/product_a") + first_pages.count("/product_b")
+        ) / len(first_pages)
+
+        assert 0.54 <= home_share <= 0.64
+        assert product_entry_share >= 0.25
+
+    def test_visit_length_is_capped_by_session_limit(self, event_dictionary, base_config):
+        """Длина визита ограничена потолком, который защищает от петель."""
+        config = replace(base_config, max_session_events=4)
+        generator = EventGenerator(event_dictionary, config)
+
+        visits = _page_path_visits(generator, 200)
+
+        assert max(len(visit) for visit in visits) <= 4
+        assert any(len(visit) == 4 for visit in visits)
+
+    def test_visit_pauses_stay_below_session_timeout_scale(self, event_dictionary, base_config):
+        """Паузы внутри визита остаются меньше 30 минут, p95 — единицы минут."""
+        generator = EventGenerator(event_dictionary, base_config)
+        pauses_seconds = []
+
+        for _ in range(500):
+            timestamps = _parse_event_timestamps(generator.generate_batch(30))
+            pauses_seconds.extend(
+                (current - previous).total_seconds()
+                for previous, current in zip(timestamps, timestamps[1:])
+            )
+
+        pauses_seconds.sort()
+        p95 = pauses_seconds[int(len(pauses_seconds) * 0.95)]
+
+        assert max(pauses_seconds) < 30 * 60
+        assert p95 < 5 * 60
+
+    def test_confirmation_share_matches_seed_scale(self, event_dictionary, base_config):
+        """Около четверти визитов доходят до /confirmation."""
+        generator = EventGenerator(event_dictionary, base_config)
+
+        visits = _page_path_visits(generator, 1000)
+        confirmation_share = (
+            sum("/confirmation" in visit for visit in visits) / len(visits)
+        )
+
+        assert 0.20 <= confirmation_share <= 0.30
+
+    def test_visit_can_continue_after_confirmation(self, event_dictionary, base_config):
+        """/confirmation не обязан быть последним событием визита."""
+        generator = EventGenerator(event_dictionary, base_config)
+
+        visits = _page_path_visits(generator, 1000)
+
+        assert any(
+            page_path == "/confirmation" and index < len(visit) - 1
+            for visit in visits
+            for index, page_path in enumerate(visit)
+        )
+
     def test_links_consistency(self, event_dictionary, base_config):
         """Связи между событиями сохраняются."""
         generator = EventGenerator(event_dictionary, base_config)