File size: 2,602 Bytes
d13981b | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 | from __future__ import annotations
from opsd_utils.dynamic_trigger_monitor import DynamicTriggerConfig, DynamicTriggerMonitor
def _config(**overrides: float | int) -> DynamicTriggerConfig:
values = {
"ema_alpha": 1.0,
"min_progress": 0.2,
"patience_steps": 2,
"sampling_mixed_max": 0.2,
"sampling_zero_loss_min": 0.7,
"rl_mixed_min": 0.3,
"rl_zero_loss_max": 0.3,
}
values.update(overrides)
return DynamicTriggerConfig(**values)
def test_sampling_needed_latches_after_patience() -> None:
monitor = DynamicTriggerMonitor(_config())
first = monitor.update(mixed_rate=0.1, zero_loss_rate=0.8, progress=0.25)
second = monitor.update(mixed_rate=0.1, zero_loss_rate=0.8, progress=0.30)
assert first["dynamic_sampling_needed_now"] == 1.0
assert first["dynamic_sampling_needed_streak"] == 1.0
assert first["dynamic_sampling_would_trigger"] == 0.0
assert second["dynamic_sampling_would_trigger"] == 1.0
assert second["dynamic_sampling_trigger_progress"] == 0.30
assert second["dynamic_rl_would_trigger"] == 0.0
def test_rl_ready_latches_independently() -> None:
monitor = DynamicTriggerMonitor(_config())
monitor.update(mixed_rate=0.4, zero_loss_rate=0.2, progress=0.40)
metrics = monitor.update(mixed_rate=0.4, zero_loss_rate=0.2, progress=0.45)
assert metrics["dynamic_rl_ready_now"] == 1.0
assert metrics["dynamic_rl_ready_streak"] == 2.0
assert metrics["dynamic_rl_would_trigger"] == 1.0
assert metrics["dynamic_rl_trigger_progress"] == 0.45
assert metrics["dynamic_sampling_would_trigger"] == 0.0
def test_conditions_do_not_accumulate_before_min_progress() -> None:
monitor = DynamicTriggerMonitor(_config())
metrics = monitor.update(mixed_rate=0.1, zero_loss_rate=0.9, progress=0.10)
assert metrics["dynamic_sampling_needed_now"] == 0.0
assert metrics["dynamic_sampling_needed_streak"] == 0.0
def test_ema_is_updated_and_trigger_progress_stays_latched() -> None:
monitor = DynamicTriggerMonitor(_config(ema_alpha=0.5, patience_steps=1))
first = monitor.update(mixed_rate=0.1, zero_loss_rate=0.9, progress=0.25)
second = monitor.update(mixed_rate=0.5, zero_loss_rate=0.1, progress=0.50)
assert first["dynamic_mixed_rate_ema"] == 0.1
assert first["dynamic_zero_loss_rate_ema"] == 0.9
assert second["dynamic_mixed_rate_ema"] == 0.3
assert second["dynamic_zero_loss_rate_ema"] == 0.5
assert second["dynamic_sampling_would_trigger"] == 1.0
assert second["dynamic_sampling_trigger_progress"] == 0.25
|