dags/dq_check.py

@@ -151,11 +151,17 @@ def build_classes_vs_inventory() -> list[str]:
         raise AirflowException("classes_vs_inventory: в описи нет закрытых дней")
 
     query = (SQL_ROOT / "dq" / "classes_vs_inventory.sql").read_text(encoding="utf-8")
-    actual = {
+    # GROUP BY закономерно не возвращает группу из нуля строк.
-        (data_date, mismatch_class): count
+    actual = {key: 0 for key, count in expected.items() if count == 0}
-        for data_date, mismatch_class, count in _clickhouse_hook().get_records(query)
+    actual.update(
-        if (data_date, mismatch_class) in expected
+        {
-    }
+            (data_date, mismatch_class): count
+            for data_date, mismatch_class, count in _clickhouse_hook().get_records(
+                query
+            )
+            if (data_date, mismatch_class) in expected
+        }
+    )
     summaries = []
     for data_date in sorted({day for day, _ in expected}):
         details = []

data/world-inventory.json

@@ -6,55 +6,57 @@
     {
       "day": 0,
       "date": "2026-06-01",
-      "events": 50626,
+      "events": 50628,
-      "sha256": "81ad7f402bdd4c8b56425f6439a024b9996e5ca95dd207d444290d20a70ed407",
+      "sha256": "d09704ed99e45824e451e2f6bbbbd5c547df382f7a582699955010f3807eb0ee",
       "orders": {
-        "match": 197,
+        "match": 192,
         "cancelled": 20,
+        "lost_event": 3,
+        "duplicate_event": 2,
         "amount_delta": 6
       }
     },
     {
       "day": 1,
       "date": "2026-06-02",
-      "events": 49722,
+      "events": 49727,
-      "sha256": "5f7aa0301832008db612eff05899b1210af958c888ac2e9e89f4b7a552569eff"
+      "sha256": "8005d2934eb353478e35331d7ae7ae6876c7934b2d9032fde92cf6ce86d6460f"
     },
     {
       "day": 2,
       "date": "2026-06-03",
-      "events": 52361,
+      "events": 52363,
-      "sha256": "10ac9ce65ca6dc171689edf78ef5cd785e8c8a80e54d236bf2b9d6dbf6997070"
+      "sha256": "e13f6f3f3977386a94d0e63b6849675cb5d2486a277d559fd415923cfc82e53a"
     },
     {
       "day": 3,
       "date": "2026-06-04",
-      "events": 53157,
+      "events": 53159,
-      "sha256": "7bf4deadd8f69a6c479fceabfe2d1e8ea8db76919b10a8b9e780006708da31c3"
+      "sha256": "91c713add62222adc3f89c04e6cc087d8c44e2385d14d0464e2c016b6952b9fd"
     },
     {
       "day": 4,
       "date": "2026-06-05",
-      "events": 50078,
+      "events": 50087,
-      "sha256": "b9e758fb72f937aa5ad1f2e90dee5ec147b16cfd89e49e51a00582f909458c29"
+      "sha256": "b7fc27d73fc157a48bece0189c7983f6bc672f052e0703cd799b9774da4b904e"
     },
     {
       "day": 5,
       "date": "2026-06-06",
-      "events": 47164,
+      "events": 47169,
-      "sha256": "f08a9f9a439f1da9e4ad8d94b9f65b7f3329256e41446d22d900cc4825f2480c"
+      "sha256": "82cb8017d53d6dee8438d3683b7a901f1ace139f3b8ae836215722bb580fe207"
     },
     {
       "day": 6,
       "date": "2026-06-07",
-      "events": 48683,
+      "events": 48686,
-      "sha256": "7e45d9dd59c902067fa6847d813669b1ab9a4ee88d2f192382577a5ed8d590a7"
+      "sha256": "b50ad0fcc7d1ea0c637cfcba98e5c2b55d65215cffa3693bc77edeaa83e6c783"
     },
     {
       "day": 7,
       "date": "2026-06-08",
       "events": 49394,
-      "sha256": "286ccc2451847ac8de215b5e902c328371021ad5502bd4afca0494b6f313aef7"
+      "sha256": "358d51b36221d822960237702fa6327fb663a57cf49cc4415a859b16e5aa05a2"
     }
   ],
   "snapshots": [

docs/architecture/dm.md

@@ -111,7 +111,7 @@
 
 Зерно — заказ. Покупки из `dds.event_v` сначала сворачиваются по
 `purchase_id[1]`: контракт источника обещает один номер и одну объявленную
-сумму, а отдельный счет событий оставляет будущий класс `duplicate_event`
+сумму, а отдельный счет событий оставляет класс `duplicate_event`
 наблюдаемым. Заказы приходят из `dds.order_v`. Обе стороны записаны
 подзапросами, затем соединены `FULL OUTER GLOBAL JOIN`: голая левая
 Distributed-таблица вернула бы каждый непарный заказ по разу с каждого шарда
@@ -176,8 +176,8 @@ Distributed-таблица вернула бы каждый непарный з
   `dds.session_v`;
 - `classes_vs_inventory`: ключ — `mismatch_class`, эталон — счётчик из
   `orders` в описи мира, объект — число заказов класса в
-  `dm.purchase_vs_orders_v`. Проверяются `match`, `cancelled` и
+  `dm.purchase_vs_orders_v`. Проверяются `match`, `cancelled`, `lost_event`,
-  `amount_delta` только для покрытых описью закрытых дней;
+  `duplicate_event` и `amount_delta` только для покрытых описью закрытых дней;
 - `revenue_daily_vs_dds`: ключ — `product_category` внутри дня. Сравниваются
   `orders`, `units`, `revenue` и `aov` с прямым агрегатом `dds.order_v`;
 - `purchase_vs_orders_vs_dds`: ключ — `order_id` внутри дня. Все деловые

docs/architecture/generator/seeds.md

@@ -30,7 +30,7 @@
         ├── TRAFFIC
         ├── COMMERCE
         ├── ORDERS
-        └── LATECOMERS — место будущей событийной стороны
+        └── BEACONS — потеря и повторная отправка событий
 ```
 
 Отрицательному дню нельзя отдать отрицательную позицию, поэтому предыстория

docs/architecture/orders/README.md

@@ -27,8 +27,8 @@
   не случиться вовсе. На выходе из окна заказ либо `paid`, либо `cancelled` —
   [судьба заказа](fate.md).
 - **Расхождения и опоздания — часть мира, а не грязь**: два подпотока —
-  заказная и событийная стороны; броски независимы, пересечения выходят
+  `ORDERS` для заказной стороны и `BEACONS` для событийной; броски независимы,
-  арифметикой, приоритет классов работает по-настоящему —
+  пересечения выходят арифметикой, приоритет классов работает по-настоящему —
   [классы расхождений и опоздание](fate.md).
 - **Опись хранит только то, чего движение мира не меняет**: хеш байтов
   каждого слепка и счётчики наблюдаемых классов —
@@ -51,10 +51,6 @@
 
 ## Открытые решения
 
-- **Имя подпотока событийной стороны** — вместе с её содержимым на этапе 6;
-  из мёртвых имён никто не бросает, переименование ничего не сдвигает (#72).
-  Заказная сторона уже названа — подпоток `ORDERS` ([судьба
-  заказа](fate.md)).
 - **Конкретные веса и доли** — таблицы исходов, моментов, задержки опоздания,
   доли классов, стоимость доставки — калибровка при реализации; финальная
   фиксация чисел — пересборка эталонного мира, этап 7. При пересборке правки

docs/architecture/orders/fate.md

@@ -16,11 +16,10 @@
 и `LATECOMERS` решения не переживают — они названы по классам витрины, а
 компонент называет часть мира; новые стороны занимают те же позиции.
 Заказная сторона названа при реализации — подпоток `ORDERS` на позиции 2;
-событийная получит имя вместе с содержимым на этапе 6, до тех пор позиция 3
+событийная — `BEACONS` на позиции 3: маячок счетчика либо не долетает, либо
-занята мёртвым именем прежней нарезки. Отклонено: *один компонент на всю
+отправляется повторно. Отклонено: *один компонент на всю судьбу* — правка
-судьбу* — правка событийной механики молча меняла бы байты слепка;
+событийной механики молча меняла бы байты слепка; *компонент на класс* —
-*компонент на класс* — пять имён под ручки калибровки, которые крутятся
+пять имён под ручки калибровки, которые крутятся разом.
-разом.
 
 **Правило формы, без которого разделение не работает: броски заказной стороны
 делаются на полную длину дня, а не на отобранных заказах.** Иначе длина броска
@@ -108,12 +107,13 @@
 покупку. Довод не в связности легенды: точечный дубль ломал бы урок соседнего
 класса — разрыв воронки, на котором держится потеря, сжался бы втрое; при
 сюжетном разрыв снова равен доле потерь. `purchaseID`, суммы и позиции у дубля
-один в один — посчитал наивно, удвоил выручку. Дубль случается только там, где
+один в один — посчитал наивно, удвоил выручку. Дубль случается только там,
-до следующего визита куки остаётся запас сверх таймаута: иначе сборка сессий у
+где пауза до следующего визита куки длиннее таймаута и задержки дубля вместе:
-менти разошлась бы с `VisitID` — сломался бы эталон, ради которого `VisitID`
+иначе сборка сессий у менти разошлась бы с `VisitID` — сломался бы эталон,
-в потоке лежит. Задержка дубля — секунды-минуты, короче таймаута визита,
+ради которого `VisitID` в потоке лежит. Задержка дубля —
-поэтому `VisitID` тот же. Полночь режет дубль парой — просмотр вместе с
+секунды-минуты, короче таймаута визита, поэтому `VisitID` тот же. Полночь
-покупкой, по тому же правилу, что у подтверждения с торговым хвостом.
+режет дубль парой — просмотр вместе с покупкой, по тому же правилу, что у
+подтверждения с торговым хвостом.
 Отклонено: *обе точечные* — дубль затирает урок потери; *обе сюжетные* —
 потеря перестаёт быть видна со стороны трекера.
 

generator/src/clickstream_generator/commerce.py

@@ -53,6 +53,7 @@
 """
 
 from dataclasses import dataclass
+from enum import IntEnum
 from typing import Any
 
 import numpy as np
@@ -160,6 +161,14 @@ class Purchases:
         return self.person_id.size
 
 
+class EventOutcome(IntEnum):
+    """Что случилось с маячком purchase после создания заказа."""
+
+    DELIVERED = 0
+    LOST = 1
+    DUPLICATED = 2
+
+
 @dataclass(frozen=True, slots=True)
 class _Events:
     """Строки одного вида торговых событий, готовые встать в поток.
@@ -177,9 +186,14 @@ class _Events:
     basket: NDArray[np.int64]
 
 
-# Что торговая половина отдаёт дню: поток целиком — колонки, страницы и
+# Что торговая половина отдает дню: поток, покупки и их событийные исходы.
-# товары — и покупки дня вторым выходом.
+Woven = tuple[
-Woven = tuple[dict[str, NDArray[Any]], NDArray[np.uint8], NDArray[np.int64], Purchases]
+    dict[str, NDArray[Any]],
+    NDArray[np.uint8],
+    NDArray[np.int64],
+    Purchases,
+    NDArray[np.int8],
+]
 
 # День без единой корзины: покупок в нём нет, а форма у рядов есть.
 _NO_PURCHASES = Purchases(
@@ -191,6 +205,7 @@ _NO_PURCHASES = Purchases(
     revenue=np.empty(0, dtype=np.int64),
     moment=np.empty(0, dtype="datetime64[s]"),
 )
+_NO_EVENT_OUTCOMES = np.empty(0, dtype=np.int8)
 
 
 def weave(
@@ -200,6 +215,7 @@ def weave(
     page: NDArray[np.uint8],
     product: NDArray[np.int64],
     person: NDArray[np.uint64],
+    assigned_order: NDArray[np.bool_],
 ) -> Woven:
     """Вплетает торговые события в поток и отдаёт поток и покупки дня.
 
@@ -207,14 +223,14 @@ def weave(
     события встают между ними, и поток пересобирается одним порядком. Второй
     выход — покупки дня: то же самое, чем они уехали в события.
 
-    `person` — человек за кукой каждой строки, выровненный по входящему
+    `person` и `assigned_order` выровнены по входящему потоку. Оба
-    потоку: личность приходит от плана состава, а не выводится из трекера
+    признака приходят из плана состава: личность дает заказу `user_id`,
-    (docs/architecture/orders/identity.md).
+    а назначение защищает мост склейки от потери purchase.
     """
     rng = day_stream(seed, day, Component.COMMERCE)
     baskets = _baskets(rng, page, product, columns["VisitID"])
     if not len(baskets):
-        return columns, page, product, _NO_PURCHASES
+        return columns, page, product, _NO_PURCHASES, _NO_EVENT_OUTCOMES
 
     goods = catalog.catalog()
     draws = _draws(rng, baskets)
@@ -225,8 +241,13 @@ def weave(
         _order_events(columns, baskets, goods, draws, revenue, coupons),
     )
     columns, page, product, basket = _stream(rng, columns, page, product, events)
-    purchases = _purchases(columns, basket, baskets, draws, revenue, coupons, person)
+    purchases, assigned = _purchases(
-    return columns, page, product, purchases
+        columns, basket, baskets, draws, revenue, coupons, person, assigned_order
+    )
+    columns, page, product, event_outcome = _corrupt_events(
+        seed, day, columns, page, product, purchases, assigned
+    )
+    return columns, page, product, purchases, event_outcome
 
 
 def _draws(rng: np.random.Generator, baskets: _Baskets) -> _Draws:
@@ -653,7 +674,8 @@ def _purchases(
     revenue: NDArray[np.int64],
     coupons: NDArray[np.object_],
     person: NDArray[np.uint64],
-) -> Purchases:
+    assigned_order: NDArray[np.bool_],
+) -> tuple[Purchases, NDArray[np.bool_]]:
     """Покупки дня — то же, чем они уехали в события, только структурой.
 
     Порядок здесь — порядок строк потока, он же порядок номеров заказа:
@@ -664,7 +686,7 @@ def _purchases(
     here = np.flatnonzero(rows["EventType"] == PURCHASE)
     mine = basket[here]
     bought = [baskets.positions_of(number, draws.kept) for number in mine.tolist()]
-    return Purchases(
+    purchases = Purchases(
         order_id=tuple(rows["purchaseID"][row][0] for row in here.tolist()),
         person_id=person[baskets.confirmation[mine]],
         product=tuple(baskets.product[group] for group in bought),
@@ -673,6 +695,118 @@ def _purchases(
         revenue=revenue[mine],
         moment=rows["UTCEventTime"][here],
     )
+    return purchases, assigned_order[baskets.confirmation[mine]]
+
+
+def _corrupt_events(
+    seed: int,
+    day: int,
+    rows: dict[str, NDArray[Any]],
+    page: NDArray[np.uint8],
+    product: NDArray[np.int64],
+    purchases: Purchases,
+    assigned: NDArray[np.bool_],
+) -> tuple[
+    dict[str, NDArray[Any]], NDArray[np.uint8], NDArray[np.int64], NDArray[np.int8]
+]:
+    """Теряет purchase или повторяет пару подтверждения.
+
+    Заказы уже собраны, а номера покупок уже вписаны в строки. Поэтому
+    порча меняет только событийную сторону. Все броски делаются на полную
+    длину покупок дня, чтобы отбор одного класса не сдвигал другой.
+    """
+    count = len(purchases)
+    if not count:
+        return rows, page, product, _NO_EVENT_OUTCOMES
+
+    rng = day_stream(seed, day, Component.BEACONS)
+    lost = (rng.integers(0, 100, count) < world.LOST_PURCHASE_PERCENT) & ~assigned
+    duplicate_draw = rng.integers(0, 100, count) < world.DUPLICATE_PURCHASE_PERCENT
+    delay = rng.integers(*world.DUPLICATE_DELAY_SECONDS, count)
+    watch = ids.unique_apart_from(rng, 2 * count, rows["WatchID"]).reshape(count, 2)
+
+    purchase_rows = np.flatnonzero(rows["EventType"] == PURCHASE)
+    confirmation_rows = _confirmation_rows(rows, page, purchase_rows)
+    eligible = _duplicate_eligibility(rows, purchase_rows, delay)
+    duplicated = ~lost & duplicate_draw & eligible
+
+    outcome = np.full(count, EventOutcome.DELIVERED, dtype=np.int8)
+    outcome[lost] = EventOutcome.LOST
+    outcome[duplicated] = EventOutcome.DUPLICATED
+
+    keep = np.ones(page.size, dtype=bool)
+    keep[purchase_rows[lost]] = False
+    sources = np.column_stack(
+        (confirmation_rows[duplicated], purchase_rows[duplicated])
+    )
+    selected = np.flatnonzero(duplicated)
+    if selected.size:
+        copies = sources.ravel()
+        shifted = np.repeat(delay[selected], 2).astype("timedelta64[s]")
+        additions = {name: values[copies] for name, values in rows.items()}
+        additions["UTCEventTime"] = additions["UTCEventTime"] + shifted
+        additions["WatchID"] = watch[selected].ravel()
+        # Первая строка пары — обновление confirmation: ее предыдущая
+        # страница — она же, а не checkout из первого просмотра.
+        additions["Referer"] = additions["URL"].copy()
+        rows = {
+            name: np.concatenate((values[keep], additions[name]))
+            for name, values in rows.items()
+        }
+        page = np.concatenate((page[keep], page[copies]))
+        product = np.concatenate((product[keep], product[copies]))
+    else:
+        rows = {name: values[keep] for name, values in rows.items()}
+        page = page[keep]
+        product = product[keep]
+
+    order = np.lexsort((rows["WatchID"], rows["UTCEventTime"]))
+    return (
+        {name: values[order] for name, values in rows.items()},
+        page[order],
+        product[order],
+        outcome,
+    )
+
+
+def _confirmation_rows(
+    rows: dict[str, NDArray[Any]],
+    page: NDArray[np.uint8],
+    purchase_rows: NDArray[np.int64],
+) -> NDArray[np.int64]:
+    """Находит просмотр, с которым purchase образует пару."""
+    confirmation = (rows["EventType"] == "pageview") & (page == Page.CONFIRMATION)
+    by_visit = {
+        int(rows["VisitID"][row]): row for row in np.flatnonzero(confirmation).tolist()
+    }
+    return np.array(
+        [by_visit[int(rows["VisitID"][row])] for row in purchase_rows.tolist()],
+        dtype=np.int64,
+    )
+
+
+def _duplicate_eligibility(
+    rows: dict[str, NDArray[Any]],
+    purchase_rows: NDArray[np.int64],
+    delay: NDArray[np.int64],
+) -> NDArray[np.bool_]:
+    """Проверяет полночь и запас до следующего визита куки."""
+    moment = rows["UTCEventTime"][purchase_rows]
+    shifted = moment + delay.astype("timedelta64[s]")
+    before_midnight = shifted < _midnight(rows) + np.timedelta64(1, "D")
+    eligible = before_midnight.copy()
+    for number, row in enumerate(purchase_rows.tolist()):
+        later_visit = (
+            (rows["ClientID"] == rows["ClientID"][row])
+            & (rows["VisitID"] != rows["VisitID"][row])
+            & (rows["UTCEventTime"] > moment[number])
+        )
+        if np.any(later_visit):
+            next_start = rows["UTCEventTime"][later_visit].min()
+            eligible[number] &= next_start - moment[number] > np.timedelta64(
+                world.VISIT_TIMEOUT_SECONDS + int(delay[number]), "s"
+            )
+    return eligible
 
 
 def _midnight(columns: dict[str, NDArray[Any]]) -> np.datetime64:

generator/src/clickstream_generator/day.py

@@ -100,6 +100,8 @@ class Day:
     page: NDArray[np.uint8]
     product: NDArray[np.int64]
     orders: orders.Orders
+    # Событийный исход каждого заказа, в том же порядке.
+    purchase_outcome: NDArray[np.int8]
 
     def __len__(self) -> int:
         return self.page.size
@@ -162,13 +164,15 @@ def stream(seed: int, day: int) -> Day:
     # целиком: в нём же они и упорядочиваются. Вторым выходом приходят
     # покупки дня — из них заказная сторона собирает заказы.
     person = np.repeat(audience.person_id[visits.cookie], visits.pages)
-    columns, page, product, purchases = commerce.weave(
+    assigned_order = np.repeat(visits.ordering, visits.pages)
+    columns, page, product, purchases, purchase_outcome = commerce.weave(
         seed,
         day,
         {name: value[order] for name, value in columns.items()},
         page[alive][order],
         product[alive][order],
         person[alive][order],
+        assigned_order[alive][order],
     )
     return Day(
         day=day,
@@ -176,6 +180,7 @@ def stream(seed: int, day: int) -> Day:
         page=page,
         product=product,
         orders=orders.of_day(seed, day, purchases),
+        purchase_outcome=purchase_outcome,
     )
 
 

generator/src/clickstream_generator/inventory.py

@@ -119,17 +119,38 @@ def _snapshot(number: int, orders: list[orders_module.Orders]) -> dict[str, Any]
 
 
 def _order_class_counts(today: day_module.Day) -> dict[str, int]:
-    """Заказы дня по итоговому классу: отмена перевешивает дельту суммы."""
+    """Заказы дня по итоговому классу по приоритету спеки."""
     purchase = today.columns["EventType"] == commerce.PURCHASE
-    declared = today.columns["purchaseRevenue"][purchase]
+    declared = {
-    counts = {"match": 0, "cancelled": 0, "amount_delta": 0}
+        purchase_id[0]: round(revenue[0] * commerce.KOPECKS)
+        for purchase_id, revenue in zip(
+            today.columns["purchaseID"][purchase],
+            today.columns["purchaseRevenue"][purchase],
+            strict=True,
+        )
+    }
+    counts = {
+        "match": 0,
+        "cancelled": 0,
+        "lost_event": 0,
+        "duplicate_event": 0,
+        "amount_delta": 0,
+    }
 
-    for outcome, items_total, revenue in zip(
+    for order_id, order_outcome, event_outcome, items_total in zip(
-        today.orders.outcome, today.orders.items_total, declared, strict=True
+        today.orders.order_id,
+        today.orders.outcome,
+        today.purchase_outcome,
+        today.orders.items_total,
+        strict=True,
     ):
-        if outcome != orders_module.OrderOutcome.PAID:
+        if order_outcome != orders_module.OrderOutcome.PAID:
             counts["cancelled"] += 1
-        elif items_total != round(revenue[0] * commerce.KOPECKS):
+        elif event_outcome == commerce.EventOutcome.LOST:
+            counts["lost_event"] += 1
+        elif event_outcome == commerce.EventOutcome.DUPLICATED:
+            counts["duplicate_event"] += 1
+        elif items_total != declared[order_id]:
             counts["amount_delta"] += 1
         else:
             counts["match"] += 1

generator/src/clickstream_generator/seeds.py

@@ -7,7 +7,7 @@
     │   ├── ось          → номер дня: когорта этого дня
     │   └── предыстория  → глубина: когорта дня до D0
     └── дни
-        └── номер дня    → трафик, торговля, заказная сторона, опоздания
+        └── номер дня    → трафик, торговля, заказы, маячки счетчика
 
 Механизм — `numpy.random.SeedSequence`: потомок полностью определяется парой
 (зерно, позиция в дереве), а не порядком вычислений. Сверено через Context7
@@ -40,10 +40,8 @@ class Component(IntEnum):
     # каждого заказа читает из его собственного дня
     # (docs/architecture/orders/fate.md).
     ORDERS = 2
-    # Событийную сторону — потерю и дубль события — заведёт этап 6; до тех
+    # Маячки счетчика: потеря и повторная отправка событий.
-    # пор позиция занята мёртвым именем прежней нарезки, и из него никто не
+    BEACONS = 3
-    # бросает.
-    LATECOMERS = 3
 
 
 class _Branch(IntEnum):

generator/src/clickstream_generator/world.py

@@ -169,6 +169,13 @@ BUYER_CHECKOUT_OF_CART_PERCENT = 60
 # страницу, на которой посетитель его нажал.
 TRADE_DELAY_SECONDS = (2, 8)
 
+# Событийная сторона сверки: потерянные маячки purchase и повторная
+# отправка пары «просмотр подтверждения + purchase». Числа черновые:
+# точная калибровка будет при пересборке эталонного мира, этап 7.
+LOST_PURCHASE_PERCENT = 3
+DUPLICATE_PURCHASE_PERCENT = 2
+DUPLICATE_DELAY_SECONDS = (30, 301)
+
 # Кто и что кладёт в корзину: доля открытых карточек, которая до неё дошла.
 # Оба ряда читаются по уровням спроса товара (`catalog.DEMAND_LEVELS`):
 # магнит, обычный, залёживается.

generator/tests/test_commerce.py

@@ -133,15 +133,15 @@ def opened_cards(events: day.Day) -> set[tuple[int, int]]:
     )
 
 
-def test_a_confirmation_in_the_stream_always_has_its_purchase(weekday: day.Day):
+def test_only_a_lost_purchase_leaves_a_confirmation_without_purchase(
-    """Клиентская сторона честная: подтверждение без покупки — это потеря.
+    weekday: day.Day,
-
+):
-    Потери и дубли стенд заводит намеренно и позже (этапы 4 и 6); здесь их
+    """Потеря выбрасывает purchase, но оставляет просмотр confirmation."""
-    быть не должно ни одной. Обратное тоже верно: покупка без подтверждения
-    в потоке — заказ ниоткуда.
-    """
     bought = set(rows_of(weekday, commerce.PURCHASE)["VisitID"].tolist())
-    assert confirmed_visits(weekday) == bought
+    missing = confirmed_visits(weekday) - bought
+    lost = int((weekday.purchase_outcome == commerce.EventOutcome.LOST).sum())
+    assert len(missing) == lost > 0
+    assert bought <= confirmed_visits(weekday)
 
 
 def test_the_day_boundary_takes_the_confirmation_together_with_its_purchase(
@@ -159,7 +159,10 @@ def test_the_day_boundary_takes_the_confirmation_together_with_its_purchase(
     events = day.stream(CANONICAL_SEED, WEEKDAY)
     bought = set(rows_of(events, commerce.PURCHASE)["VisitID"].tolist())
     assert bought
-    assert confirmed_visits(events) == bought
+    missing = confirmed_visits(events) - bought
+    lost = int((events.purchase_outcome == commerce.EventOutcome.LOST).sum())
+    assert len(missing) == lost
+    assert bought <= confirmed_visits(events)
 
 
 def test_a_purchase_carries_exactly_one_order_number(weekday: day.Day):
@@ -525,13 +528,15 @@ def test_the_goals_repeat_the_trade_events(weekday: day.Day):
 
 def test_the_order_number_reads_as_the_day_and_the_count(weekday: day.Day):
     """Номер — день и порядковый номер покупки в нём; он же `order_id` бэкенда."""
-    orders = rows_of(weekday, commerce.PURCHASE)
+    events = rows_of(weekday, commerce.PURCHASE)
-    numbers = [cell[0] for cell in orders["purchaseID"]]
+    numbers = [cell[0] for cell in events["purchaseID"]]
     date = str(weekday.columns["EventDate"][0]).replace("-", "")
-    assert numbers == [f"{date}-{place:04d}" for place in range(1, len(numbers) + 1)]
+    expected = [f"{date}-{place:04d}" for place in range(1, len(weekday.orders) + 1)]
-    assert len(set(numbers)) == len(numbers)
+    # Порча идет после нумерации: потеря оставляет дыру, а дубль
-    # Нумерация идёт в порядке событий: строки дня уже упорядочены по времени.
+    # повторяет готовый номер и не двигает соседей.
-    assert list(orders["UTCEventTime"]) == sorted(orders["UTCEventTime"])
+    assert list(weekday.orders.order_id) == expected
+    assert set(numbers) <= set(expected)
+    assert list(events["UTCEventTime"]) == sorted(events["UTCEventTime"])
 
 
 def test_the_coupon_comes_from_the_world_table_and_does_not_touch_the_revenue(
@@ -599,7 +604,23 @@ def test_the_trade_flow_draws_from_its_own_named_branch(
 
     monkeypatch.setattr(commerce, "day_stream", spy)
     day.stream(CANONICAL_SEED, WEEKDAY)
-    assert asked == [(CANONICAL_SEED, WEEKDAY, Component.COMMERCE)]
+    assert asked == [
+        (CANONICAL_SEED, WEEKDAY, Component.COMMERCE),
+        (CANONICAL_SEED, WEEKDAY, Component.BEACONS),
+    ]
+
+
+def test_an_assigned_purchase_is_never_lost(monkeypatch: pytest.MonkeyPatch):
+    """Гарантия плана сильнее порчи: мост склейки не теряется."""
+    monkeypatch.setattr(world, "LOST_PURCHASE_PERCENT", 100)
+    monkeypatch.setattr(world, "DUPLICATE_PURCHASE_PERCENT", 0)
+    events = day.stream(CANONICAL_SEED, WEEKDAY)
+    audience = plan.audience(CANONICAL_SEED, WEEKDAY)
+
+    promised = set(audience.client_id[audience.assigned_order].tolist())
+    bought = set(rows_of(events, commerce.PURCHASE)["ClientID"].tolist())
+    assert promised
+    assert bought == promised
 
 
 def test_the_promised_orders_of_a_pair_become_purchases(week: list[day.Day]):
@@ -627,6 +648,78 @@ def test_the_promised_orders_of_a_pair_become_purchases(week: list[day.Day]):
     assert pairs == plan.counters(CANONICAL_SEED, WEEK).pairs
 
 
+def test_a_duplicate_repeats_the_confirmation_story(weekday: day.Day):
+    """Обновление повторяет просмотр и purchase в том же визите."""
+    purchases = rows_of(weekday, commerce.PURCHASE)
+    ids_by_order: dict[str, list[int]] = {}
+    for row, cell in enumerate(purchases["purchaseID"]):
+        ids_by_order.setdefault(str(cell[0]), []).append(row)
+    duplicated = [rows for rows in ids_by_order.values() if len(rows) == 2]
+
+    expected = int((weekday.purchase_outcome == commerce.EventOutcome.DUPLICATED).sum())
+    assert len(duplicated) == expected > 0
+    for pair in duplicated:
+        first, second = sorted(pair, key=lambda row: purchases["UTCEventTime"][row])
+        delay = (
+            (purchases["UTCEventTime"][second] - purchases["UTCEventTime"][first])
+            .astype("timedelta64[s]")
+            .astype(int)
+        )
+        assert world.DUPLICATE_DELAY_SECONDS[0] <= delay
+        assert delay < world.DUPLICATE_DELAY_SECONDS[1]
+        assert delay < world.VISIT_TIMEOUT_SECONDS
+        assert purchases["VisitID"][first] == purchases["VisitID"][second]
+        assert purchases["WatchID"][first] != purchases["WatchID"][second]
+        for name in (
+            "purchaseID",
+            "purchaseRevenue",
+            "productID",
+            "productQuantity",
+            "ecommerce",
+        ):
+            left, right = purchases[name][first], purchases[name][second]
+            if isinstance(left, np.ndarray):
+                assert np.array_equal(left, right), name
+            else:
+                assert left == right, name
+
+        visit = purchases["VisitID"][first]
+        confirmation = (
+            (weekday.columns["EventType"] == "pageview")
+            & (weekday.page == Page.CONFIRMATION)
+            & (weekday.columns["VisitID"] == visit)
+        )
+        views = np.flatnonzero(confirmation)
+        assert views.size == 2
+        assert len(set(weekday.columns["WatchID"][views].tolist())) == 2
+        view_delay = (
+            (
+                weekday.columns["UTCEventTime"][views].max()
+                - weekday.columns["UTCEventTime"][views].min()
+            )
+            .astype("timedelta64[s]")
+            .astype(int)
+        )
+        assert view_delay == delay
+
+        cookie = purchases["ClientID"][first]
+        later_visit = (
+            (weekday.columns["ClientID"] == cookie)
+            & (weekday.columns["VisitID"] != visit)
+            & (weekday.columns["UTCEventTime"] > purchases["UTCEventTime"][second])
+        )
+        if np.any(later_visit):
+            gap = (
+                (
+                    weekday.columns["UTCEventTime"][later_visit].min()
+                    - purchases["UTCEventTime"][second]
+                )
+                .astype("timedelta64[s]")
+                .astype(int)
+            )
+            assert gap > world.VISIT_TIMEOUT_SECONDS
+
+
 def paired_cookies() -> set[int]:
     """Куки двухкуковых пар: их заказы обещаны планом, а не решены днём."""
     cookies: set[int] = set()
@@ -642,21 +735,26 @@ def test_flagged_buyers_buy_again_much_more_often(week: list[day.Day]):
     «Постоянный покупатель» читается только так — как кука, которая ходит
     неделями и покупает не раз: метки в событии нет и не будет.
 
-    Куки двухкуковых пар из счёта исключены: их заказы назначил план, и
+    Куки двухкуковых пар из счета исключены: их заказы назначил план, и
-    вместе с ними лифт мерил бы гарантию, а не два новых рычага. Остаются
+    вместе с ними лифт мерил бы гарантию, а не два новых рычага. Даже после
-    те помеченные, чьи покупки решил день: повторно покупают 4,6% из них
+    событийной порчи оставшиеся помеченные покупают повторно более чем вдвое
-    против 1,5% прочих — втрое чаще при пороге «вдвое».
+    чаще прочих.
     """
-    purchases: dict[int, int] = {}
+    purchases: dict[int, set[str]] = {}
     flagged: set[int] = set()
     for number, events in enumerate(week):
         audience = plan.audience(CANONICAL_SEED, number)
         flagged |= set(audience.client_id[audience.buyer].tolist())
-        for cookie in rows_of(events, commerce.PURCHASE)["ClientID"].tolist():
+        rows = rows_of(events, commerce.PURCHASE)
-            purchases[cookie] = purchases.get(cookie, 0) + 1
+        for cookie, order_id in zip(
+            rows["ClientID"].tolist(),
+            (str(cell[0]) for cell in rows["purchaseID"]),
+            strict=True,
+        ):
+            purchases.setdefault(cookie, set()).add(order_id)
 
     def repeat_share(cookies: set[int]) -> float:
-        again = sum(1 for cookie in cookies if purchases[cookie] > 1)
+        again = sum(1 for cookie in cookies if len(purchases[cookie]) > 1)
         return again / len(cookies)
 
     buyers = set(purchases)

generator/tests/test_day.py

@@ -88,6 +88,7 @@ def snapshot(events: day.Day) -> dict[str, Any]:
         "page": events.page.tolist(),
         "product": events.product.tolist(),
         "orders": orders(events),
+        "purchase_outcome": events.purchase_outcome.tolist(),
     }
 
 
@@ -115,6 +116,7 @@ def test_the_snapshot_notices_everything_the_day_hands_out(weekday: day.Day):
         page=weekday.page,
         product=weekday.product,
         orders=weekday.orders,
+        purchase_outcome=weekday.purchase_outcome,
     )
     assert snapshot(reordered) != original
 

generator/tests/test_inventory.py

@@ -48,7 +48,15 @@ def test_order_class_counters_cover_only_closed_days(inventory: dict):
     rows = [row for row in inventory["days"] if "orders" in row]
 
     assert [row["day"] for row in rows] == [0]
-    assert set(rows[0]["orders"]) == {"match", "cancelled", "amount_delta"}
+    assert set(rows[0]["orders"]) == {
+        "match",
+        "cancelled",
+        "lost_event",
+        "duplicate_event",
+        "amount_delta",
+    }
+    assert rows[0]["orders"]["lost_event"] > 0
+    assert rows[0]["orders"]["duplicate_event"] > 0
     today = day_module.stream(CANONICAL_SEED, rows[0]["day"])
     assert sum(rows[0]["orders"].values()) == len(today.orders)
 

generator/tests/test_orders.py

@@ -58,17 +58,33 @@ def user_id_by_order(events: day.Day) -> dict[str, int]:
     )
 
 
-def test_every_order_of_the_day_is_a_purchase_of_the_day(weekday: day.Day):
+def purchase_row_by_order(
-    """Заказ — вторая проекция покупки, а не второе порождение.
+    events: day.Day,
+) -> tuple[dict[str, NDArray[Any]], dict[str, int]]:
+    """Первая строка purchase каждого доехавшего заказа."""
+    rows = purchases_of(events)
+    index: dict[str, int] = {}
+    for row, cell in enumerate(rows["purchaseID"]):
+        index.setdefault(str(cell[0]), row)
+    return rows, index
 
-    У всякого заказа ровно одно событие `purchase`, и номер у них один:
+
-    сойтись двум источникам больше негде — сверка стенда соединяется
+def test_every_order_of_the_day_is_a_purchase_of_the_day(weekday: day.Day):
-    именно по нему.
+    """Заказ рождается до потери или дубля его событийной проекции."""
-    """
     events = purchases_of(weekday)
     numbers = [cell[0] for cell in events["purchaseID"]]
     assert len(numbers) > 100
-    assert list(weekday.orders.order_id) == numbers
+    lost = {
+        order_id
+        for order_id, outcome in zip(
+            weekday.orders.order_id, weekday.purchase_outcome, strict=True
+        )
+        if outcome == commerce.EventOutcome.LOST
+    }
+    assert set(numbers) == set(weekday.orders.order_id) - lost
+    assert len(numbers) - len(set(numbers)) == int(
+        (weekday.purchase_outcome == commerce.EventOutcome.DUPLICATED).sum()
+    )
 
 
 def dropped_line(client: list[tuple[Any, int]], order: list[tuple[Any, int]]):
@@ -88,16 +104,18 @@ def test_the_order_repeats_the_purchase_up_to_one_dropped_line(weekday: day.Day)
     """
     goods = catalog.catalog()
     price = dict(zip(goods.sku.tolist(), goods.price.tolist(), strict=True))
-    events = purchases_of(weekday)
+    events, row_by_order = purchase_row_by_order(weekday)
     money = weekday.orders
     deltas = 0
 
     for number, order in enumerate(weekday.orders.order_id):
-        assert order == events["purchaseID"][number][0]
+        if weekday.purchase_outcome[number] == commerce.EventOutcome.LOST:
+            continue
+        row = row_by_order[order]
         client = list(
             zip(
-                events["productID"][number].tolist(),
+                events["productID"][row].tolist(),
-                events["productQuantity"][number].tolist(),
+                events["productQuantity"][row].tolist(),
                 strict=True,
             )
         )
@@ -110,7 +128,7 @@ def test_the_order_repeats_the_purchase_up_to_one_dropped_line(weekday: day.Day)
         )
         # Выручка клиента и `items_total` бэкенда — одно число: в событии
         # оно дробное, у заказа целое в копейках.
-        revenue = round(events["purchaseRevenue"][number][0] * commerce.KOPECKS)
+        revenue = round(events["purchaseRevenue"][row][0] * commerce.KOPECKS)
         if mine == client:
             assert money.items_total[number] == revenue
             continue
@@ -136,15 +154,18 @@ def test_the_discount_comes_from_the_coupon_of_the_event(
 
     for today in start_world:
         money = today.orders
-        events = purchases_of(today)
+        events, row_by_order = purchase_row_by_order(today)
-        codes = [cell[0] for cell in events["purchaseCoupon"]]
+        for number, order_id in enumerate(money.order_id):
-        for number, code in enumerate(codes):
+            if today.purchase_outcome[number] == commerce.EventOutcome.LOST:
+                continue
+            row = row_by_order[order_id]
+            code = events["purchaseCoupon"][row][0]
             items_total = money.items_total[number]
             expected = items_total * percent[code] // 100 if code else 0
             assert money.discount[number] == expected, money.order_id[number]
             assert money.discount[number] < items_total
 
-            revenue = round(events["purchaseRevenue"][number][0] * commerce.KOPECKS)
+            revenue = round(events["purchaseRevenue"][row][0] * commerce.KOPECKS)
             discounted_delta += int(bool(code) and items_total != revenue)
 
     # Иначе проверка не отличила бы новую базу скидки от прежней.
@@ -255,8 +276,13 @@ def test_the_user_id_is_the_person_behind_the_cookie(weekday: day.Day):
         zip(audience.client_id.tolist(), audience.person_id.tolist(), strict=True)
     )
     events = purchases_of(weekday)
-    for number, cookie in enumerate(events["ClientID"].tolist()):
+    by_order = user_id_by_order(weekday)
-        assert weekday.orders.user_id[number] == person[cookie]
+    for order_id, cookie in zip(
+        (cell[0] for cell in events["purchaseID"]),
+        events["ClientID"].tolist(),
+        strict=True,
+    ):
+        assert by_order[order_id] == person[cookie]
     assert np.all(weekday.orders.user_id > 0)
     assert weekday.orders.user_id.max() < 2**53
 

generator/tests/test_seeds.py

@@ -78,7 +78,7 @@ def test_day_components_do_not_share_randomness():
         Component.TRAFFIC,
         Component.COMMERCE,
         Component.ORDERS,
-        Component.LATECOMERS,
+        Component.BEACONS,
     )
     assert len({int(component) for component in components}) == 4
     draws = {

generator/tests/test_snapshot.py

@@ -128,7 +128,9 @@ def test_the_record_matches_the_wire_contract(days: list[day_module.Day]):
         )
 
         if _born(record) == _date(SHORT_WINDOW_DAY):
-            assert record["created_at"][:19] + "Z" == bought[record["order_id"]]
+            # Потерянного purchase в трекере нет, но заказ остается.
+            if record["order_id"] in bought:
+                assert record["created_at"][:19] + "Z" == bought[record["order_id"]]
 
 
 def test_negative_money_does_not_leave_the_source(days: list[day_module.Day]):
@@ -325,7 +327,10 @@ def _purchase_times(today: day_module.Day) -> dict[str, str]:
     times = np.datetime_as_string(
         today.columns["UTCEventTime"][here], unit="s", timezone="UTC"
     )
-    return dict(zip(numbers, times.tolist(), strict=True))
+    result: dict[str, str] = {}
+    for number, moment in zip(numbers, times.tolist(), strict=True):
+        result.setdefault(number, moment)
+    return result
 
 
 def _born(record: dict) -> date:

generator/tests/test_world.py

@@ -122,6 +122,11 @@ def test_the_trade_event_never_overtakes_the_next_page():
     assert max(world.TRADE_DELAY_SECONDS) <= min(world.PAGE_PAUSE_SECONDS)
 
 
+def test_the_duplicate_delay_stays_inside_a_visit_timeout():
+    assert 0 < min(world.DUPLICATE_DELAY_SECONDS)
+    assert max(world.DUPLICATE_DELAY_SECONDS) <= world.VISIT_TIMEOUT_SECONDS
+
+
 def conversion(cart: int, checkout_of_cart: int) -> float:
     """Конверсия визита в заказ, проценты: произведение трёх шагов воронки."""
     return cart * checkout_of_cart * world.CONFIRMATION_OF_CHECKOUT_PERCENT / 100**2

sql/dq/classes_vs_inventory.sql

@@ -2,13 +2,19 @@
 --
 -- Эталонные счётчики читает Airflow из world-inventory.json: у настоящего
 -- магазина такого оракула нет. Запрос возвращает только наблюдаемую сторону;
--- Python-задача сопоставляет ее с тремя счетчиками закрытого дня из описи.
+-- Python-задача сопоставляет ее с пятью счетчиками закрытого дня из описи.
 SELECT
     order_day AS data_date,
     mismatch_class,
     count() AS orders
 FROM dm.purchase_vs_orders_v
-WHERE mismatch_class IN ('match', 'cancelled', 'amount_delta')
+WHERE mismatch_class IN (
+    'match',
+    'cancelled',
+    'lost_event',
+    'duplicate_event',
+    'amount_delta'
+)
 GROUP BY
     data_date,
     mismatch_class