feat(generator): добавлено опоздание заказов в слепках
- Зачем: - менти должен увидеть, как запоздавшая выгрузка меняет открытое окно, не подделывая аудит источника. - Что: - заказам назначена задержка D, D+1 или D+2 последним броском подпотока ORDERS. - назначенный планом заказ приезжает сразу, а Woven передаёт эту гарантию именованным полем. - обновлены документы, проверки границ механики и эталонные хеши слепков. - Проверка: - make lint, make typecheck и make test в generator: 436 тестов прошли. - make config-test: конфигурация прошла проверку. - world_live_day и следующий etl_pipeline перевели заказ 20260608-0001 из awaiting_order в match.
This commit is contained in:
1 parent
58e944aa87
commit
8db1c58c12
11 files changed
+186
-54
No files matched your search
@@ -186,14 +186,18 @@ class _Events:
|
||||
basket: NDArray[np.int64]
|
||||
|
||||
|
||||
# Что торговая половина отдает дню: поток, покупки и их событийные исходы.
|
||||
Woven = tuple[
|
||||
dict[str, NDArray[Any]],
|
||||
NDArray[np.uint8],
|
||||
NDArray[np.int64],
|
||||
Purchases,
|
||||
NDArray[np.int8],
|
||||
]
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class Woven:
|
||||
"""Что торговая половина отдает дню после вплетения событий."""
|
||||
|
||||
columns: dict[str, NDArray[Any]]
|
||||
page: NDArray[np.uint8]
|
||||
product: NDArray[np.int64]
|
||||
purchases: Purchases
|
||||
event_outcome: NDArray[np.int8]
|
||||
# Для каждой покупки: назначен ли её заказ планом.
|
||||
assigned_order: NDArray[np.bool_]
|
||||
|
||||
|
||||
# День без единой корзины: покупок в нём нет, а форма у рядов есть.
|
||||
_NO_PURCHASES = Purchases(
|
||||
@@ -206,6 +210,7 @@ _NO_PURCHASES = Purchases(
|
||||
moment=np.empty(0, dtype="datetime64[s]"),
|
||||
)
|
||||
_NO_EVENT_OUTCOMES = np.empty(0, dtype=np.int8)
|
||||
_NO_ASSIGNED_ORDERS = np.empty(0, dtype=np.bool_)
|
||||
|
||||
|
||||
def weave(
|
||||
@@ -230,7 +235,14 @@ def weave(
|
||||
rng = day_stream(seed, day, Component.COMMERCE)
|
||||
baskets = _baskets(rng, page, product, columns["VisitID"])
|
||||
if not len(baskets):
|
||||
return columns, page, product, _NO_PURCHASES, _NO_EVENT_OUTCOMES
|
||||
return Woven(
|
||||
columns=columns,
|
||||
page=page,
|
||||
product=product,
|
||||
purchases=_NO_PURCHASES,
|
||||
event_outcome=_NO_EVENT_OUTCOMES,
|
||||
assigned_order=_NO_ASSIGNED_ORDERS,
|
||||
)
|
||||
|
||||
goods = catalog.catalog()
|
||||
draws = _draws(rng, baskets)
|
||||
@@ -247,7 +259,14 @@ def weave(
|
||||
columns, page, product, event_outcome = _corrupt_events(
|
||||
seed, day, columns, page, product, purchases, assigned
|
||||
)
|
||||
return columns, page, product, purchases, event_outcome
|
||||
return Woven(
|
||||
columns=columns,
|
||||
page=page,
|
||||
product=product,
|
||||
purchases=purchases,
|
||||
event_outcome=event_outcome,
|
||||
assigned_order=assigned,
|
||||
)
|
||||
|
||||
|
||||
def _draws(rng: np.random.Generator, baskets: _Baskets) -> _Draws:
|
||||
|
||||
@@ -165,7 +165,7 @@ def stream(seed: int, day: int) -> Day:
|
||||
# покупки дня — из них заказная сторона собирает заказы.
|
||||
person = np.repeat(audience.person_id[visits.cookie], visits.pages)
|
||||
assigned_order = np.repeat(visits.ordering, visits.pages)
|
||||
columns, page, product, purchases, purchase_outcome = commerce.weave(
|
||||
woven = commerce.weave(
|
||||
seed,
|
||||
day,
|
||||
{name: value[order] for name, value in columns.items()},
|
||||
@@ -176,11 +176,11 @@ def stream(seed: int, day: int) -> Day:
|
||||
)
|
||||
return Day(
|
||||
day=day,
|
||||
columns=columns,
|
||||
page=page,
|
||||
product=product,
|
||||
orders=orders.of_day(seed, day, purchases),
|
||||
purchase_outcome=purchase_outcome,
|
||||
columns=woven.columns,
|
||||
page=woven.page,
|
||||
product=woven.product,
|
||||
orders=orders.of_day(seed, day, woven.purchases, woven.assigned_order),
|
||||
purchase_outcome=woven.event_outcome,
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -24,11 +24,13 @@
|
||||
одной доли перебрасывала бы весь подпоток после себя
|
||||
(docs/architecture/orders/fate.md).
|
||||
|
||||
**Судьба заказа — исход и его моменты**: из окна изменяемости заказ выходит
|
||||
оплаченным или отменённым, а когда именно это случилось, сказано смещением в
|
||||
секундах от рождения заказа. Момента, которого у исхода нет, нет и в данных:
|
||||
его место занимает −1, а не ноль, — иначе «оплатили в секунду рождения» было
|
||||
бы не отличить от «не оплатили вовсе».
|
||||
**Судьба заказа — исход, его моменты и задержка выгрузки**. Из окна
|
||||
изменяемости заказ выходит оплаченным или отменённым, а когда именно это
|
||||
случилось, сказано смещением в секундах от рождения заказа. Момента, которого
|
||||
у исхода нет, нет и в данных: его место занимает −1, а не ноль, — иначе
|
||||
«оплатили в секунду рождения» было бы не отличить от «не оплатили вовсе».
|
||||
Задержка говорит, сколько ранних слепков заказ пропустит; назначенный планом
|
||||
приезжает сразу.
|
||||
|
||||
**Состояние на границе суток** — чтение готовой судьбы, а не накопление:
|
||||
слепок дня D учитывает моменты не позже границы D|D+1 и по ним называет
|
||||
@@ -62,6 +64,7 @@ _DELIVERY_CUMULATIVE = np.cumsum(
|
||||
)
|
||||
_OUTCOME_CUMULATIVE = np.cumsum(world.ORDER_OUTCOME_WEIGHTS)
|
||||
_MOMENT_HOUR_CUMULATIVE = np.cumsum(world.ORDER_MOMENT_HOUR_WEIGHTS)
|
||||
_SNAPSHOT_DELAY_CUMULATIVE = np.cumsum(world.ORDER_SNAPSHOT_DELAY_WEIGHTS)
|
||||
|
||||
|
||||
class OrderOutcome(IntEnum):
|
||||
@@ -103,12 +106,18 @@ class Orders:
|
||||
outcome: NDArray[np.int64]
|
||||
paid_after: NDArray[np.int64]
|
||||
cancelled_after: NDArray[np.int64]
|
||||
snapshot_delay: NDArray[np.int64]
|
||||
|
||||
def __len__(self) -> int:
|
||||
return self.user_id.size
|
||||
|
||||
|
||||
def of_day(seed: int, day: int, purchases: Purchases) -> Orders:
|
||||
def of_day(
|
||||
seed: int,
|
||||
day: int,
|
||||
purchases: Purchases,
|
||||
assigned_order: NDArray[np.bool_],
|
||||
) -> Orders:
|
||||
"""Заказы дня `day`: его покупки, к которым бэкенд добавил свои деньги."""
|
||||
rng = day_stream(seed, day, Component.ORDERS)
|
||||
delivery = _delivery(rng, len(purchases))
|
||||
@@ -116,6 +125,7 @@ def of_day(seed: int, day: int, purchases: Purchases) -> Orders:
|
||||
product, quantity, items_total = _delta(rng, purchases)
|
||||
created_at = _created_at(rng, purchases)
|
||||
discount = _discount(purchases, items_total)
|
||||
snapshot_delay = _snapshot_delay(rng, assigned_order)
|
||||
return Orders(
|
||||
day=day,
|
||||
order_id=purchases.order_id,
|
||||
@@ -130,6 +140,7 @@ def of_day(seed: int, day: int, purchases: Purchases) -> Orders:
|
||||
outcome=outcome,
|
||||
paid_after=paid_after,
|
||||
cancelled_after=cancelled_after,
|
||||
snapshot_delay=snapshot_delay,
|
||||
)
|
||||
|
||||
|
||||
@@ -166,6 +177,11 @@ def at_boundary(rows: Orders, day: int) -> tuple[list[str], NDArray[np.datetime6
|
||||
return status.tolist(), updated
|
||||
|
||||
|
||||
def arrived(rows: Orders, day: int) -> NDArray[np.bool_]:
|
||||
"""Какие заказы уже попали в выгрузку к слепку дня `day`."""
|
||||
return rows.day + rows.snapshot_delay <= day
|
||||
|
||||
|
||||
def _boundary(day: int) -> np.datetime64:
|
||||
"""Граница суток `day`|`day+1` абсолютной меткой: полночь пояса счётчика.
|
||||
|
||||
@@ -218,6 +234,15 @@ def _created_at(
|
||||
)
|
||||
|
||||
|
||||
def _snapshot_delay(
|
||||
rng: np.random.Generator,
|
||||
assigned_order: NDArray[np.bool_],
|
||||
) -> NDArray[np.int64]:
|
||||
"""Сколько ранних слепков пропустит заказ; назначенный приезжает сразу."""
|
||||
delay = pick(rng, _SNAPSHOT_DELAY_CUMULATIVE, assigned_order.size)
|
||||
return np.where(assigned_order, 0, delay)
|
||||
|
||||
|
||||
def _moment(rng: np.random.Generator, orders: int) -> NDArray[np.int64]:
|
||||
"""Момент внутри окна: час по таблице весов и равномерная секунда в нём.
|
||||
|
||||
|
||||
@@ -45,7 +45,7 @@ from numpy.typing import NDArray
|
||||
|
||||
from clickstream_generator import catalog, schema, world
|
||||
from clickstream_generator.day import Day
|
||||
from clickstream_generator.orders import Orders, at_boundary
|
||||
from clickstream_generator.orders import Orders, arrived, at_boundary
|
||||
|
||||
_ARRAY_PREFIX = "Array("
|
||||
|
||||
@@ -72,9 +72,10 @@ def events(day: Day, limit: int | None = None) -> list[bytes]:
|
||||
def orders(window: Sequence[Orders], day: int) -> list[bytes]:
|
||||
"""Канонические байты слепка дня `day`: по документу JSON на заказ.
|
||||
|
||||
`window` — заказы дней окна, от раннего дня к позднему: слепок несёт их
|
||||
подряд, и порядок строк выходит порядком рождения заказов, он же
|
||||
возрастание `order_id`. Какие это дни, решает `orders.window`.
|
||||
`window` — заказы дней окна, от раннего дня к позднему. Уже приехавшие
|
||||
заказы слепок несёт подряд, поэтому порядок строк остаётся порядком
|
||||
рождения, он же возрастание `order_id`. Какие это дни, решает
|
||||
`orders.window`.
|
||||
|
||||
Деньги уезжают строками с ровно двумя знаками, а не числами: у заказа они
|
||||
станут `Decimal`, и дробь двоичного числа была бы потерей точности до
|
||||
@@ -97,7 +98,8 @@ def orders(window: Sequence[Orders], day: int) -> list[bytes]:
|
||||
delivery = rows.delivery.tolist()
|
||||
total = rows.total.tolist()
|
||||
|
||||
for number, order_id in enumerate(rows.order_id):
|
||||
for number in np.flatnonzero(arrived(rows, day)).tolist():
|
||||
order_id = rows.order_id[number]
|
||||
payloads.append(
|
||||
orjson.dumps(
|
||||
{
|
||||
|
||||
@@ -283,3 +283,8 @@ ORDER_MOMENT_HOUR_WEIGHTS = tuple(
|
||||
# выходит вдвое-втрое ниже броска, около ориентира мастер-спеки в 1–2%. Число
|
||||
# черновое: калибровка — пересборка эталонного мира, этап 7.
|
||||
ORDER_DELTA_PERCENT = 4
|
||||
|
||||
# На сколько первых дневных слепков заказ опоздает: веса δ = 0, 1, 2.
|
||||
# Назначенные планом заказы после броска получают 0, чтобы мост склейки
|
||||
# появился в стартовом мире. Числа черновые: калибровка — этап 7.
|
||||
ORDER_SNAPSHOT_DELAY_WEIGHTS = (90, 5, 5)
|
||||
Reference in new issue
Block a user