fix(find_cheapest_schedule): retry valid window start after time gaps

_find_cheapest_window_in_pool() scans for contiguous available blocks
when scheduling multiple tasks. When a block-in-progress hit a
temporal gap (e.g. from price-level filtering removing intervals from
the middle of the search range, or missing API data), the scanner
jumped to i = j + 1 instead of i = j, silently skipping the interval
right after the gap as a valid — sometimes cheaper — window start.

Distinguish 'unavailable slot' (correctly skipped via j + 1) from
'temporal gap' (must retry at j, since that slot was never actually
tested as a window start).

Impact: find_cheapest_schedule can now find the true cheapest window
for a task when the search range contains time gaps, instead of
occasionally picking a more expensive window right after such a gap.
This commit is contained in:
Julian Pawlowski 2026-07-04 18:18:42 +00:00
parent b8e40bfa3b
commit 36e9cdf4b9
2 changed files with 89 additions and 2 deletions

View file

@ -229,6 +229,7 @@ def _find_cheapest_window_in_pool(
# and all intervals are contiguous in time (no gaps) # and all intervals are contiguous in time (no gaps)
block: list[dict[str, Any]] = [] block: list[dict[str, Any]] = []
j = i j = i
stopped_at_gap = False
while j < n and len(block) < duration_intervals: while j < n and len(block) < duration_intervals:
if not available[j]: if not available[j]:
break break
@ -239,7 +240,10 @@ def _find_cheapest_window_in_pool(
prev_dt = datetime.fromisoformat(prev_start) if isinstance(prev_start, str) else prev_start prev_dt = datetime.fromisoformat(prev_start) if isinstance(prev_start, str) else prev_start
curr_dt = datetime.fromisoformat(curr_start) if isinstance(curr_start, str) else curr_start curr_dt = datetime.fromisoformat(curr_start) if isinstance(curr_start, str) else curr_start
if curr_dt - prev_dt != timedelta(minutes=INTERVAL_MINUTES): if curr_dt - prev_dt != timedelta(minutes=INTERVAL_MINUTES):
# Gap in time — can't extend this block, skip to j+1 # Gap in time — can't extend this block, but slot j itself
# (already confirmed available above) is a fresh, untried
# candidate window start and must not be skipped.
stopped_at_gap = True
break break
block.append(pool[j]) block.append(pool[j])
j += 1 j += 1
@ -253,8 +257,15 @@ def _find_cheapest_window_in_pool(
best_sum = window_sum best_sum = window_sum
best_start = i best_start = i
i += 1 i += 1
elif stopped_at_gap:
# Retry starting exactly at the gap position (index j), not j+1 —
# j is available and untested as a window start. Regression guard:
# previously this jumped to j+1, silently skipping a valid
# (sometimes cheaper) candidate window right after a time gap
# caused by price-level filtering or missing data.
i = j
else: else:
# Skip past the blocking unavailable/non-contiguous slot # Skip past the blocking unavailable slot (or end of pool)
i = j + 1 i = j + 1
if best_start == -1: if best_start == -1:

View file

@ -0,0 +1,76 @@
"""Tests for pure internal helpers in find_cheapest_schedule.py."""
from __future__ import annotations
from datetime import UTC, datetime, timedelta
from custom_components.tibber_prices.services.find_cheapest_schedule import _find_cheapest_window_in_pool
def _build_pool(minutes_and_prices: list[tuple[int, float]], base: datetime | None = None) -> list[dict]:
"""Build a pool of interval dicts from (minute_offset, price) tuples."""
base = base or datetime(2026, 1, 1, 0, 0, tzinfo=UTC)
return [
{"startsAt": (base + timedelta(minutes=minute)).isoformat(), "total": price}
for minute, price in minutes_and_prices
]
class TestFindCheapestWindowInPoolTemporalGap:
"""Regression coverage: a temporal gap must not hide the cheapest window.
`_find_cheapest_window_in_pool` scans for contiguous available blocks. When
a block-in-progress hits a time gap (e.g. because price-level filtering
removed an interval from the middle of the range), the position where the
gap was detected is itself a valid, untested candidate window start and
must be retried not skipped.
"""
def test_cheapest_window_right_after_gap_is_found(self) -> None:
"""The cheapest 2-interval window sits right after a 30min time gap."""
# Minutes: 0, 15, [gap - 30 missing], 45, 60, 75, 90
# Cheapest contiguous pair by price sum is (45, 60) = 1 + 1 = 2.
pool = _build_pool([(0, 10.0), (15, 10.0), (45, 1.0), (60, 1.0), (75, 10.0), (90, 10.0)])
available = [True] * len(pool)
result = _find_cheapest_window_in_pool(pool, 2, available)
assert result is not None
start, end = result
assert (start, end) == (2, 4)
chosen = pool[start:end]
assert [iv["startsAt"] for iv in chosen] == [pool[2]["startsAt"], pool[3]["startsAt"]]
assert sum(iv["total"] for iv in chosen) == 2.0
def test_cheapest_window_immediately_before_gap_still_found(self) -> None:
"""A candidate window ending right before a gap must also still work."""
# Minutes: 0, 15, 30 [gap], 60, 75
# Cheapest contiguous pair is (0, 15) = 1 + 1 = 2, right before the gap.
pool = _build_pool([(0, 1.0), (15, 1.0), (30, 10.0), (60, 10.0), (75, 10.0)])
available = [True] * len(pool)
result = _find_cheapest_window_in_pool(pool, 2, available)
assert result == (0, 2)
def test_multiple_gaps_all_candidate_starts_considered(self) -> None:
"""Multiple gaps in a row must not compound-skip valid starts."""
# Minutes: 0, [gap], 30, [gap], 60, [gap], 90 — no 2 contiguous intervals
# exist at all here, so no window should be found despite many gaps.
pool = _build_pool([(0, 1.0), (30, 1.0), (60, 1.0), (90, 1.0)])
available = [True] * len(pool)
result = _find_cheapest_window_in_pool(pool, 2, available)
assert result is None
def test_unavailable_slot_is_still_correctly_skipped(self) -> None:
"""An unavailable (already-claimed) slot must still be skipped entirely."""
pool = _build_pool([(0, 1.0), (15, 1.0), (30, 1.0), (45, 1.0)])
# Claim index 1 (minute 15) as unavailable (e.g. assigned to another task).
available = [True, False, True, True]
result = _find_cheapest_window_in_pool(pool, 2, available)
# Only (30, 45) is a valid contiguous available pair.
assert result == (2, 4)