ddmitry/clickstream-ch-kafka-superset-demo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat(generator): добавлено ускорение модельного времени

Browse Source 
- Зачем:
  - ускоренный стенд должен проходить больше модельного времени и давать соответствующий событийный бюджет.
- Что:
  - расчёт интенсивности переведён на модельную длительность тика.
  - добавлена устойчивая выборка бюджета при больших λ.
  - усилены тесты ×K, дневного коэффициента и независимости от настенного часа.
- Проверка:
  - make generator-test.
  - review gate после issue 03 пройден после исправления underflow Poisson.
This commit is contained in:
ddadmin
2026-06-14 17:48:51 +03:00
parent efb07b0283
commit 589e2321b3
7 changed files with 237 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
generator/tests/test_generation.py
+96 -18
View File
@@ -6,7 +6,7 @@ import json
import random
import uuid
from dataclasses import replace
from datetime import datetime, timedelta
from datetime import datetime, timedelta, timezone
import pytest
from generator import (
@@ -16,6 +16,7 @@ from generator import (
TickStreamGenerator,
calculate_events_count,
generate_tick_batch,
hour_factor,
)
@@ -1030,14 +1031,19 @@ class TestEventGeneration:
class TestPoissonDistribution:
"""Тесты статистической модели."""
def test_event_budget_mean_follows_lambda_and_hour_factor(
self, base_config, monkeypatch
):
def test_hour_factor_uses_model_timezone(self):
"""Дневной коэффициент считается по заданному часовому поясу модели."""
assert hour_factor(
datetime(2026, 1, 1, 2, 30, tzinfo=timezone.utc),
"Europe/Moscow",
) == 0.7
assert hour_factor(
datetime(2026, 1, 1, 6, 30, tzinfo=timezone.utc),
"Europe/Moscow",
) == 1.2
def test_event_budget_mean_follows_lambda_and_hour_factor(self, base_config):
"""Средний событийный бюджет следует λ и часовому коэффициенту."""
monkeypatch.setattr(
"clickstream_generator.intensity.hour_factor",
lambda: 1.2,
)
config = replace(
base_config,
tick_seconds=60,
@@ -1045,6 +1051,8 @@ class TestPoissonDistribution:
jitter_pct=0,
min_events_per_tick=1,
max_events_per_tick=100,
model_t0=datetime(2026, 1, 1, 10, 0, tzinfo=timezone.utc),
model_timezone="UTC",
)
rng = random.Random(config.seed)
@@ -1053,20 +1061,16 @@ class TestPoissonDistribution:
assert mean_budget == pytest.approx(30 * 1.2, rel=0.15)
def test_default_tick_budget_floor_does_not_outgrow_target_lambda(
self, base_config, monkeypatch
):
def test_default_tick_budget_floor_does_not_outgrow_target_lambda(self, base_config):
"""Дефолтная нижняя граница бюджета не разгоняет lambda=30 на тике 5 секунд."""
monkeypatch.setattr(
"clickstream_generator.intensity.hour_factor",
lambda: 1.0,
)
config = replace(
base_config,
tick_seconds=5,
lambda_base_per_min=30,
jitter_pct=0,
max_events_per_tick=50,
model_t0=datetime(2026, 1, 1, 7, 0, tzinfo=timezone.utc),
model_timezone="UTC",
)
rng = random.Random(config.seed)
@@ -1075,6 +1079,69 @@ class TestPoissonDistribution:
assert events_per_minute == pytest.approx(config.lambda_base_per_min, rel=0.20)
def test_event_budget_uses_model_tick_duration(self, base_config):
"""При ускорении событийный бюджет растёт по модельной длительности тика."""
base = replace(
base_config,
tick_seconds=60,
lambda_base_per_min=30,
jitter_pct=0,
min_events_per_tick=1,
max_events_per_tick=10_000,
model_time_speed=1,
)
accelerated = replace(base, model_time_speed=10)
model_tick_at = datetime(2026, 1, 1, 7, 0)
normal_rng = random.Random(base.seed)
accelerated_rng = random.Random(accelerated.seed)
normal_samples = [
calculate_events_count(base, normal_rng, now=model_tick_at)
for _ in range(300)
]
accelerated_samples = [
calculate_events_count(accelerated, accelerated_rng, now=model_tick_at)
for _ in range(300)
]
normal_mean = sum(normal_samples) / len(normal_samples)
accelerated_mean = sum(accelerated_samples) / len(accelerated_samples)
assert accelerated_mean / normal_mean == pytest.approx(10, rel=0.15)
def test_large_event_budget_does_not_stick_on_knuth_underflow(self, base_config):
"""Для λ > 1000 средний бюджет растёт вместе с целевой интенсивностью."""
config = replace(
base_config,
tick_seconds=60,
lambda_base_per_min=1200,
jitter_pct=0,
min_events_per_tick=0,
max_events_per_tick=10_000,
model_t0=datetime(2026, 1, 1, 7, 0, tzinfo=timezone.utc),
model_timezone="UTC",
)
rng = random.Random(config.seed)
samples = [calculate_events_count(config, rng) for _ in range(500)]
mean_budget = sum(samples) / len(samples)
assert mean_budget == pytest.approx(1200, rel=0.05)
assert mean_budget > 1000
def test_event_generator_hour_factor_defaults_to_model_t0(
self, event_dictionary, base_config
):
"""Wrapper без аргумента берёт модельную точку, а не настенный час."""
config = replace(
base_config,
model_t0=datetime(2026, 1, 1, 6, 30, tzinfo=timezone.utc),
model_timezone="Europe/Moscow",
)
generator = EventGenerator(event_dictionary, config)
assert generator._hour_factor() == 1.2
def test_calculate_events_respects_bounds(self, event_dictionary, base_config):
"""Расчет количества событий уважает границы."""
generator = EventGenerator(event_dictionary, base_config)
@@ -1084,10 +1151,21 @@ class TestPoissonDistribution:
assert all(s >= base_config.min_events_per_tick for s in samples)
assert all(s <= base_config.max_events_per_tick for s in samples)
def test_jitter_increases_variance(self, event_dictionary, base_config, config_no_jitter):
def test_jitter_increases_variance(self, event_dictionary, base_config):
"""Jitter увеличивает дисперсию."""
gen_with = EventGenerator(event_dictionary, base_config)
gen_without = EventGenerator(event_dictionary, config_no_jitter)
config_with_jitter = replace(
base_config,
tick_seconds=60,
lambda_base_per_min=30,
jitter_pct=50,
min_events_per_tick=1,
max_events_per_tick=100,
model_t0=datetime(2026, 1, 1, 7, 0, tzinfo=timezone.utc),
model_timezone="UTC",
)
config_without_jitter = replace(config_with_jitter, jitter_pct=0)
gen_with = EventGenerator(event_dictionary, config_with_jitter)
gen_without = EventGenerator(event_dictionary, config_without_jitter)
samples_with = [gen_with._calculate_events_count() for _ in range(200)]
samples_without = [gen_without._calculate_events_count() for _ in range(200)]
generator/tests/test_service.py
+50 -1
View File
@@ -135,6 +135,11 @@ class TestGeneratorServiceSteadyStream:
)
assert len(night_wall_run["browser_events"]) == len(day_wall_run["browser_events"])
assert night_wall_run["budget_model_times"] == [
datetime(2026, 1, 1, 10, 0, tzinfo=timezone.utc),
datetime(2026, 1, 1, 10, 1, tzinfo=timezone.utc),
]
assert day_wall_run["budget_model_times"] == night_wall_run["budget_model_times"]
assert night_wall_run["browser_events"]
timestamps = {
event["event_timestamp"]
@@ -145,6 +150,38 @@ class TestGeneratorServiceSteadyStream:
"2026-01-01 10:01:00.000000",
}
def test_service_live_tick_advances_event_timestamps_by_model_speed(self, base_config):
"""При ×K сервис сдвигает события на ускоренный модельный шаг."""
model_t0 = datetime(2026, 1, 1, 10, 0, tzinfo=timezone.utc)
config = replace(
base_config,
tick_seconds=60,
lambda_base_per_min=600,
jitter_pct=0,
min_events_per_tick=1,
max_events_per_tick=1000,
max_session_events=1,
max_active_sessions=250,
population_max=251,
model_t0=model_t0,
model_time_speed=10,
)
published = self._run_service_ticks(
config,
wall_now=datetime(2026, 6, 14, 11, 0, tzinfo=timezone.utc),
ticks_count=2,
)
timestamps = {
event["event_timestamp"]
for event in published["browser_events"]
}
assert timestamps == {
"2026-01-01 10:00:00.000000",
"2026-01-01 10:10:00.000000",
}
def _run_service_ticks(self, config, wall_now: datetime, ticks_count: int):
service = GeneratorService(config)
service.publisher = MagicMock()
@@ -154,6 +191,8 @@ class TestGeneratorServiceSteadyStream:
service.history = MagicMock()
service._running = True
sleep_calls = 0
budget_model_times = []
original_calculate_events_count = service.generator._calculate_events_count
class FrozenDateTime(datetime):
@classmethod
@@ -168,7 +207,16 @@ class TestGeneratorServiceSteadyStream:
if sleep_calls >= ticks_count:
service._running = False
with patch("clickstream_generator.intensity.datetime", FrozenDateTime), \
def calculate_events_count(now=None):
budget_model_times.append(now)
return original_calculate_events_count(now=now)
with patch("clickstream_generator.service.datetime", FrozenDateTime), \
patch.object(
service.generator,
"_calculate_events_count",
side_effect=calculate_events_count,
), \
patch("clickstream_generator.service.time.sleep", side_effect=stop_after_tick):
service._main_loop()
@@ -176,6 +224,7 @@ class TestGeneratorServiceSteadyStream:
for call in service.publisher.publish.call_args_list:
topic, events = call.args
published.setdefault(topic, []).extend(events)
published["budget_model_times"] = budget_model_times
return published
def test_service_ticks_publish_connected_multi_event_visit(self, base_config):