From c28f724706fa00bb330e3c13acd2e2d5881558a8 Mon Sep 17 00:00:00 2001 From: Dmitry Dementev Date: Fri, 28 Aug 2026 13:13:47 -0400 Subject: [PATCH] =?UTF-8?q?feat(generator):=20=D0=BF=D0=BE=D1=82=D0=B5?= =?UTF-8?q?=D1=80=D1=8F=20=D0=B8=20=D0=B4=D1=83=D0=B1=D0=BB=D1=8C=20purcha?= =?UTF-8?q?se=20(B,=20D)=20=D0=B8=20=D1=81=D1=87=D0=B5=D1=82=D1=87=D0=B8?= =?UTF-8?q?=D0=BA=D0=B8=20=D0=BE=D0=BF=D0=B8=D1=81=D0=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Зачем: - трекер стенда был полным, и правило «деньги — по бэкенду» держалось на слове; классы B и D делают недосчет и пересчет трекера наблюдаемыми (этап 6, #184). - Что: - подпоток BEACONS: точечная потеря purchase (~3%) и сюжетный дубль (~2%) швом после _purchases, с гарантией моста идентичностей и условием допуска дубля по паузе до следующего визита; - опись считает пять классов по приоритету спеки; фильтр classes_vs_inventory расширен, нулевой класс читается нулем; - fate.md, seeds.md, dm.md и orders/README.md приведены к имени BEACONS и строгому правилу допуска. - Проверка: - make lint, make typecheck, make test (433 passed) и make inventory из generator/ — хеши слепков не изменились; - на стенде: make smoke, make check-clickhouse, make check-services зеленые; в dq_summary classes_vs_inventory и sessions_vs_reference pass, витрина сверки показывает lost_event и duplicate_event. Co-Authored-By: Claude Fable 5 --- dags/dq_check.py | 16 +- data/world-inventory.json | 34 ++-- docs/architecture/dm.md | 6 +- docs/architecture/generator/seeds.md | 2 +- docs/architecture/orders/README.md | 8 +- docs/architecture/orders/fate.md | 22 +-- .../src/clickstream_generator/commerce.py | 156 ++++++++++++++++-- generator/src/clickstream_generator/day.py | 7 +- .../src/clickstream_generator/inventory.py | 35 +++- generator/src/clickstream_generator/seeds.py | 8 +- generator/src/clickstream_generator/world.py | 7 + generator/tests/test_commerce.py | 146 +++++++++++++--- generator/tests/test_day.py | 2 + generator/tests/test_inventory.py | 10 +- generator/tests/test_orders.py | 62 +++++-- generator/tests/test_seeds.py | 2 +- generator/tests/test_snapshot.py | 9 +- generator/tests/test_world.py | 5 + sql/dq/classes_vs_inventory.sql | 10 +- 19 files changed, 433 insertions(+), 114 deletions(-) diff --git a/dags/dq_check.py b/dags/dq_check.py index accefcf..c2323ea 100644 --- a/dags/dq_check.py +++ b/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 = { - (data_date, mismatch_class): count - for data_date, mismatch_class, count in _clickhouse_hook().get_records(query) - if (data_date, mismatch_class) in expected - } + # GROUP BY закономерно не возвращает группу из нуля строк. + actual = {key: 0 for key, count in expected.items() if count == 0} + actual.update( + { + (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 = [] diff --git a/data/world-inventory.json b/data/world-inventory.json index 8c610a6..38d08dd 100644 --- a/data/world-inventory.json +++ b/data/world-inventory.json @@ -6,55 +6,57 @@ { "day": 0, "date": "2026-06-01", - "events": 50626, - "sha256": "81ad7f402bdd4c8b56425f6439a024b9996e5ca95dd207d444290d20a70ed407", + "events": 50628, + "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, - "sha256": "5f7aa0301832008db612eff05899b1210af958c888ac2e9e89f4b7a552569eff" + "events": 49727, + "sha256": "8005d2934eb353478e35331d7ae7ae6876c7934b2d9032fde92cf6ce86d6460f" }, { "day": 2, "date": "2026-06-03", - "events": 52361, - "sha256": "10ac9ce65ca6dc171689edf78ef5cd785e8c8a80e54d236bf2b9d6dbf6997070" + "events": 52363, + "sha256": "e13f6f3f3977386a94d0e63b6849675cb5d2486a277d559fd415923cfc82e53a" }, { "day": 3, "date": "2026-06-04", - "events": 53157, - "sha256": "7bf4deadd8f69a6c479fceabfe2d1e8ea8db76919b10a8b9e780006708da31c3" + "events": 53159, + "sha256": "91c713add62222adc3f89c04e6cc087d8c44e2385d14d0464e2c016b6952b9fd" }, { "day": 4, "date": "2026-06-05", - "events": 50078, - "sha256": "b9e758fb72f937aa5ad1f2e90dee5ec147b16cfd89e49e51a00582f909458c29" + "events": 50087, + "sha256": "b7fc27d73fc157a48bece0189c7983f6bc672f052e0703cd799b9774da4b904e" }, { "day": 5, "date": "2026-06-06", - "events": 47164, - "sha256": "f08a9f9a439f1da9e4ad8d94b9f65b7f3329256e41446d22d900cc4825f2480c" + "events": 47169, + "sha256": "82cb8017d53d6dee8438d3683b7a901f1ace139f3b8ae836215722bb580fe207" }, { "day": 6, "date": "2026-06-07", - "events": 48683, - "sha256": "7e45d9dd59c902067fa6847d813669b1ab9a4ee88d2f192382577a5ed8d590a7" + "events": 48686, + "sha256": "b50ad0fcc7d1ea0c637cfcba98e5c2b55d65215cffa3693bc77edeaa83e6c783" }, { "day": 7, "date": "2026-06-08", "events": 49394, - "sha256": "286ccc2451847ac8de215b5e902c328371021ad5502bd4afca0494b6f313aef7" + "sha256": "358d51b36221d822960237702fa6327fb663a57cf49cc4415a859b16e5aa05a2" } ], "snapshots": [ diff --git a/docs/architecture/dm.md b/docs/architecture/dm.md index a3e5bf4..b2aed93 100644 --- a/docs/architecture/dm.md +++ b/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` и - `amount_delta` только для покрытых описью закрытых дней; + `dm.purchase_vs_orders_v`. Проверяются `match`, `cancelled`, `lost_event`, + `duplicate_event` и `amount_delta` только для покрытых описью закрытых дней; - `revenue_daily_vs_dds`: ключ — `product_category` внутри дня. Сравниваются `orders`, `units`, `revenue` и `aov` с прямым агрегатом `dds.order_v`; - `purchase_vs_orders_vs_dds`: ключ — `order_id` внутри дня. Все деловые diff --git a/docs/architecture/generator/seeds.md b/docs/architecture/generator/seeds.md index 46eaccb..3fe3f43 100644 --- a/docs/architecture/generator/seeds.md +++ b/docs/architecture/generator/seeds.md @@ -30,7 +30,7 @@ ├── TRAFFIC ├── COMMERCE ├── ORDERS - └── LATECOMERS — место будущей событийной стороны + └── BEACONS — потеря и повторная отправка событий ``` Отрицательному дню нельзя отдать отрицательную позицию, поэтому предыстория diff --git a/docs/architecture/orders/README.md b/docs/architecture/orders/README.md index c1d3f19..32e3336 100644 --- a/docs/architecture/orders/README.md +++ b/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. При пересборке правки diff --git a/docs/architecture/orders/fate.md b/docs/architecture/orders/fate.md index 117aee7..419f891 100644 --- a/docs/architecture/orders/fate.md +++ b/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` тот же. Полночь +режет дубль парой — просмотр вместе с покупкой, по тому же правилу, что у +подтверждения с торговым хвостом. Отклонено: *обе точечные* — дубль затирает урок потери; *обе сюжетные* — потеря перестаёт быть видна со стороны трекера. diff --git a/generator/src/clickstream_generator/commerce.py b/generator/src/clickstream_generator/commerce.py index 6eee879..59ef50f 100644 --- a/generator/src/clickstream_generator/commerce.py +++ b/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[dict[str, NDArray[Any]], NDArray[np.uint8], NDArray[np.int64], Purchases] +# Что торговая половина отдает дню: поток, покупки и их событийные исходы. +Woven = tuple[ + 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` — человек за кукой каждой строки, выровненный по входящему - потоку: личность приходит от плана состава, а не выводится из трекера - (docs/architecture/orders/identity.md). + `person` и `assigned_order` выровнены по входящему потоку. Оба + признака приходят из плана состава: личность дает заказу `user_id`, + а назначение защищает мост склейки от потери 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) - return columns, page, product, purchases + purchases, assigned = _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: diff --git a/generator/src/clickstream_generator/day.py b/generator/src/clickstream_generator/day.py index 0a186cf..754a064 100644 --- a/generator/src/clickstream_generator/day.py +++ b/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, ) diff --git a/generator/src/clickstream_generator/inventory.py b/generator/src/clickstream_generator/inventory.py index 51b1f08..7378e3a 100644 --- a/generator/src/clickstream_generator/inventory.py +++ b/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] - counts = {"match": 0, "cancelled": 0, "amount_delta": 0} + declared = { + 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( - today.orders.outcome, today.orders.items_total, declared, strict=True + for order_id, order_outcome, event_outcome, items_total in zip( + 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 diff --git a/generator/src/clickstream_generator/seeds.py b/generator/src/clickstream_generator/seeds.py index b0375e3..7398069 100644 --- a/generator/src/clickstream_generator/seeds.py +++ b/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; до тех - # пор позиция занята мёртвым именем прежней нарезки, и из него никто не - # бросает. - LATECOMERS = 3 + # Маячки счетчика: потеря и повторная отправка событий. + BEACONS = 3 class _Branch(IntEnum): diff --git a/generator/src/clickstream_generator/world.py b/generator/src/clickstream_generator/world.py index 9a0b305..2b5b664 100644 --- a/generator/src/clickstream_generator/world.py +++ b/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`): # магнит, обычный, залёживается. diff --git a/generator/tests/test_commerce.py b/generator/tests/test_commerce.py index 0a2f6fb..80637f8 100644 --- a/generator/tests/test_commerce.py +++ b/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): - """Клиентская сторона честная: подтверждение без покупки — это потеря. - - Потери и дубли стенд заводит намеренно и позже (этапы 4 и 6); здесь их - быть не должно ни одной. Обратное тоже верно: покупка без подтверждения - в потоке — заказ ниоткуда. - """ +def test_only_a_lost_purchase_leaves_a_confirmation_without_purchase( + weekday: day.Day, +): + """Потеря выбрасывает 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) - numbers = [cell[0] for cell in orders["purchaseID"]] + events = rows_of(weekday, commerce.PURCHASE) + 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)] - assert len(set(numbers)) == len(numbers) - # Нумерация идёт в порядке событий: строки дня уже упорядочены по времени. - assert list(orders["UTCEventTime"]) == sorted(orders["UTCEventTime"]) + expected = [f"{date}-{place:04d}" for place in range(1, len(weekday.orders) + 1)] + # Порча идет после нумерации: потеря оставляет дыру, а дубль + # повторяет готовый номер и не двигает соседей. + 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(): - purchases[cookie] = purchases.get(cookie, 0) + 1 + rows = rows_of(events, commerce.PURCHASE) + 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) diff --git a/generator/tests/test_day.py b/generator/tests/test_day.py index 3818235..58f23e8 100644 --- a/generator/tests/test_day.py +++ b/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 diff --git a/generator/tests/test_inventory.py b/generator/tests/test_inventory.py index fbe303a..174d225 100644 --- a/generator/tests/test_inventory.py +++ b/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) diff --git a/generator/tests/test_orders.py b/generator/tests/test_orders.py index c9fec14..1eae821 100644 --- a/generator/tests/test_orders.py +++ b/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["productQuantity"][number].tolist(), + events["productID"][row].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) - codes = [cell[0] for cell in events["purchaseCoupon"]] - for number, code in enumerate(codes): + events, row_by_order = purchase_row_by_order(today) + for number, order_id in enumerate(money.order_id): + 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()): - assert weekday.orders.user_id[number] == person[cookie] + by_order = user_id_by_order(weekday) + 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 diff --git a/generator/tests/test_seeds.py b/generator/tests/test_seeds.py index 4a87a65..3bc6747 100644 --- a/generator/tests/test_seeds.py +++ b/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 = { diff --git a/generator/tests/test_snapshot.py b/generator/tests/test_snapshot.py index a189ec3..53acdfe 100644 --- a/generator/tests/test_snapshot.py +++ b/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: diff --git a/generator/tests/test_world.py b/generator/tests/test_world.py index 872b16f..c19510d 100644 --- a/generator/tests/test_world.py +++ b/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 diff --git a/sql/dq/classes_vs_inventory.sql b/sql/dq/classes_vs_inventory.sql index d686290..a53217a 100644 --- a/sql/dq/classes_vs_inventory.sql +++ b/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