From b8e40bfa3b7358a5df90a793d3f26acb91de7846 Mon Sep 17 00:00:00 2001 From: Julian Pawlowski Date: Sat, 4 Jul 2026 17:43:04 +0000 Subject: [PATCH] fix(services): prevent plan_charging overcharge from segment-constraint bridging apply_segment_constraints could add far more grid energy than requested when max_cycles_per_day or min_charge_duration_minutes forced bridging across expensive gaps between cheap intervals. With max_cycles_per_day=1, isolated cheap intervals were merged into one continuous segment by filling every gap in between, without ever trimming the surplus back down, resulting in achieved_soc_percent far above 100%. Add a post-bridging trim step that removes segment-edge intervals (highest price first) until total grid energy matches the requested target again, while still respecting max_cycles_per_day/min_charge_duration_minutes and never dropping below the target itself (fixed-power mode's expected last-interval rounding overshoot is preserved). A related edge case is also fixed: trimming could previously remove an interval required to satisfy a must_reach_by deadline, silently flipping deadline_met to False even though the overall energy target was still reached. Deadline-critical intervals are now passed through as protected_starts and are never removed during trimming. Fixes #167 Impact: plan_charging no longer overcharges the battery/EV past the requested target SoC when max_cycles_per_day or min_charge_duration_minutes is set, and must_reach_by deadlines are honored even when those constraints require bridging across expensive price gaps. --- .../services/charging/power_scheduler.py | 372 +++++++++++++----- .../tibber_prices/services/plan_charging.py | 11 + tests/services/test_plan_charging.py | 61 ++- tests/services/test_power_scheduler.py | 84 +++- 4 files changed, 430 insertions(+), 98 deletions(-) diff --git a/custom_components/tibber_prices/services/charging/power_scheduler.py b/custom_components/tibber_prices/services/charging/power_scheduler.py index 933f460..d2604eb 100644 --- a/custom_components/tibber_prices/services/charging/power_scheduler.py +++ b/custom_components/tibber_prices/services/charging/power_scheduler.py @@ -218,6 +218,245 @@ def _add_interval_if_available( return True +def _constraints_satisfied( + intervals: list[dict[str, Any]], + *, + max_cycles_per_day: int | None, + min_charge_duration_minutes: int | None, + interval_minutes: int, +) -> bool: + """Check whether current interval selection satisfies active constraints.""" + grouped_segments = group_intervals_into_segments(intervals) + + if max_cycles_per_day and len(grouped_segments) > max_cycles_per_day: + return False + + if min_charge_duration_minutes: + required_intervals = max(1, math.ceil(min_charge_duration_minutes / interval_minutes)) + if any(segment["interval_count"] < required_intervals for segment in grouped_segments): + return False + + return True + + +def _extend_for_min_duration( + selected_map: dict[str, dict[str, Any]], + *, + candidate_map: dict[str, dict[str, Any]], + candidates_sorted: list[dict[str, Any]], + candidate_index: dict[str, int], + minimum_power_w: int, + charging_efficiency: float, + interval_minutes: int, + min_charge_duration_minutes: int, + warnings: list[str], +) -> None: + """Extend short segments by adding contiguous neighbor intervals.""" + required_intervals = max(1, math.ceil(min_charge_duration_minutes / interval_minutes)) + progress = True + + while progress: + progress = False + selected_intervals = sorted(selected_map.values(), key=_interval_start) + segments = group_intervals_into_segments(selected_intervals) + + for segment in segments: + if segment["interval_count"] >= required_intervals: + continue + + while segment["interval_count"] < required_intervals: + first = segment["intervals"][0]["startsAt"] + last = segment["intervals"][-1]["startsAt"] + first_index = candidate_index[first] + last_index = candidate_index[last] + + prev_interval = candidates_sorted[first_index - 1] if first_index > 0 else None + next_interval = candidates_sorted[last_index + 1] if last_index + 1 < len(candidates_sorted) else None + + options: list[dict[str, Any]] = [] + if ( + prev_interval is not None + and _interval_start(candidate_map[first]) - _interval_start(prev_interval) + == timedelta(minutes=interval_minutes) + and prev_interval["startsAt"] not in selected_map + ): + options.append(prev_interval) + + if ( + next_interval is not None + and _interval_start(next_interval) - _interval_start(candidate_map[last]) + == timedelta(minutes=interval_minutes) + and next_interval["startsAt"] not in selected_map + ): + options.append(next_interval) + + if not options: + warnings.append("min_charge_duration_unreachable") + break + + cheapest = min(options, key=lambda interval: (_sort_price(interval), _interval_start(interval))) + added = _add_interval_if_available( + selected_map, + candidate_map, + cheapest["startsAt"], + power_w=minimum_power_w, + charging_efficiency=charging_efficiency, + interval_minutes=interval_minutes, + ) + if not added: + break + + progress = True + selected_intervals = sorted(selected_map.values(), key=_interval_start) + segment = next( + seg + for seg in group_intervals_into_segments(selected_intervals) + if first in {iv["startsAt"] for iv in seg["intervals"]} + ) + + +def _merge_for_max_cycles( + selected_map: dict[str, dict[str, Any]], + *, + candidate_map: dict[str, dict[str, Any]], + candidates_sorted: list[dict[str, Any]], + candidate_index: dict[str, int], + minimum_power_w: int, + charging_efficiency: float, + interval_minutes: int, + max_cycles_per_day: int, + warnings: list[str], +) -> None: + """Bridge cheapest gaps until the cycle limit is satisfied.""" + while True: + selected_intervals = sorted(selected_map.values(), key=_interval_start) + segments = group_intervals_into_segments(selected_intervals) + if len(segments) <= max_cycles_per_day: + break + + best_gap: tuple[float, list[dict[str, Any]]] | None = None + for left, right in pairwise(segments): + left_end_index = candidate_index[left["intervals"][-1]["startsAt"]] + right_start_index = candidate_index[right["intervals"][0]["startsAt"]] + gap = candidates_sorted[left_end_index + 1 : right_start_index] + + if not gap: + continue + if any(interval["startsAt"] in selected_map for interval in gap): + continue + if any( + _interval_start(gap[index + 1]) - _interval_start(gap[index]) != timedelta(minutes=interval_minutes) + for index in range(len(gap) - 1) + ): + continue + + penalty = sum(_sort_price(interval) for interval in gap) + if best_gap is None or penalty < best_gap[0]: + best_gap = (penalty, gap) + + if best_gap is None: + warnings.append("max_cycles_unreachable") + break + + for interval in best_gap[1]: + _add_interval_if_available( + selected_map, + candidate_map, + interval["startsAt"], + power_w=minimum_power_w, + charging_efficiency=charging_efficiency, + interval_minutes=interval_minutes, + ) + + +def _collect_removable_edge_indices( + selected_intervals: list[dict[str, Any]], + *, + total_grid_energy: float, + target_grid_energy_kwh: float, + max_cycles_per_day: int | None, + min_charge_duration_minutes: int | None, + interval_minutes: int, + protected_starts: frozenset[str] | None, +) -> list[int]: + """Return edge interval indices that can be removed while keeping constraints valid.""" + removable_indices: list[int] = [] + segments = group_intervals_into_segments(selected_intervals) + + for segment in segments: + first_start = segment["intervals"][0]["startsAt"] + last_start = segment["intervals"][-1]["startsAt"] + + for edge_start in (first_start, last_start): + if protected_starts is not None and edge_start in protected_starts: + continue + + edge_index = next( + (index for index, interval in enumerate(selected_intervals) if interval["startsAt"] == edge_start), + None, + ) + if edge_index is None or edge_index in removable_indices: + continue + + candidate = selected_intervals[edge_index] + new_total = total_grid_energy - float(candidate["grid_energy_kwh"]) + if new_total + _INTERVAL_TOLERANCE < target_grid_energy_kwh: + continue + + new_selection = selected_intervals[:edge_index] + selected_intervals[edge_index + 1 :] + if not new_selection: + continue + if not _constraints_satisfied( + new_selection, + max_cycles_per_day=max_cycles_per_day, + min_charge_duration_minutes=min_charge_duration_minutes, + interval_minutes=interval_minutes, + ): + continue + + removable_indices.append(edge_index) + + return removable_indices + + +def _trim_to_target_energy( + selected_map: dict[str, dict[str, Any]], + *, + target_grid_energy_kwh: float, + max_cycles_per_day: int | None, + min_charge_duration_minutes: int | None, + interval_minutes: int, + protected_starts: frozenset[str] | None = None, +) -> dict[str, dict[str, Any]]: + """Trim excess energy from selection by removing expensive edge intervals first. + + Intervals whose ``startsAt`` is listed in ``protected_starts`` (for example, intervals + required to satisfy a ``must_reach_by`` deadline) are never removed, even if that means + the target energy cannot be fully reached through trimming alone. + """ + selected_intervals = sorted(selected_map.values(), key=_interval_start) + total_grid_energy = sum(float(interval["grid_energy_kwh"]) for interval in selected_intervals) + + while selected_intervals and total_grid_energy > target_grid_energy_kwh + _INTERVAL_TOLERANCE: + removable_indices = _collect_removable_edge_indices( + selected_intervals, + total_grid_energy=total_grid_energy, + target_grid_energy_kwh=target_grid_energy_kwh, + max_cycles_per_day=max_cycles_per_day, + min_charge_duration_minutes=min_charge_duration_minutes, + interval_minutes=interval_minutes, + protected_starts=protected_starts, + ) + if not removable_indices: + break + + best_index = max(removable_indices, key=lambda index: _sort_price(selected_intervals[index])) + total_grid_energy -= float(selected_intervals[best_index]["grid_energy_kwh"]) + del selected_intervals[best_index] + + return {interval["startsAt"]: interval for interval in selected_intervals} + + def apply_segment_constraints( schedule: dict[str, Any], candidate_intervals: list[dict[str, Any]], @@ -225,9 +464,15 @@ def apply_segment_constraints( charging_efficiency: float, min_charge_duration_minutes: int | None = None, max_cycles_per_day: int | None = None, + target_grid_energy_kwh: float | None = None, + protected_starts: frozenset[str] | None = None, interval_minutes: int = 15, ) -> tuple[dict[str, Any], list[str]]: - """Extend/bridge selected intervals to satisfy segment duration and cycle constraints.""" + """Extend/bridge selected intervals to satisfy segment duration and cycle constraints. + + ``protected_starts`` marks intervals (by ``startsAt``) that must never be removed while + trimming to ``target_grid_energy_kwh``, e.g. intervals required to meet a deadline. + """ warnings: list[str] = [] selected_map = {interval["startsAt"]: dict(interval) for interval in schedule["intervals"]} candidate_map = {interval["startsAt"]: interval for interval in candidate_intervals} @@ -236,103 +481,40 @@ def apply_segment_constraints( minimum_power_w = int(schedule["minimum_power_w"]) if min_charge_duration_minutes: - required_intervals = max(1, math.ceil(min_charge_duration_minutes / interval_minutes)) - progress = True - while progress: - progress = False - selected_intervals = sorted(selected_map.values(), key=_interval_start) - segments = group_intervals_into_segments(selected_intervals) - for segment in segments: - if segment["interval_count"] >= required_intervals: - continue - while segment["interval_count"] < required_intervals: - first = segment["intervals"][0]["startsAt"] - last = segment["intervals"][-1]["startsAt"] - first_index = candidate_index[first] - last_index = candidate_index[last] - - prev_interval = candidates_sorted[first_index - 1] if first_index > 0 else None - next_interval = ( - candidates_sorted[last_index + 1] if last_index + 1 < len(candidates_sorted) else None - ) - - prev_contiguous = False - next_contiguous = False - if prev_interval is not None: - prev_contiguous = _interval_start(candidate_map[first]) - _interval_start( - prev_interval - ) == timedelta(minutes=interval_minutes) - if next_interval is not None: - next_contiguous = _interval_start(next_interval) - _interval_start( - candidate_map[last] - ) == timedelta(minutes=interval_minutes) - - options: list[dict[str, Any]] = [] - if prev_interval is not None and prev_contiguous and prev_interval["startsAt"] not in selected_map: - options.append(prev_interval) - if next_interval is not None and next_contiguous and next_interval["startsAt"] not in selected_map: - options.append(next_interval) - if not options: - warnings.append("min_charge_duration_unreachable") - break - - cheapest = min(options, key=lambda interval: (_sort_price(interval), _interval_start(interval))) - added = _add_interval_if_available( - selected_map, - candidate_map, - cheapest["startsAt"], - power_w=minimum_power_w, - charging_efficiency=charging_efficiency, - interval_minutes=interval_minutes, - ) - if not added: - break - progress = True - selected_intervals = sorted(selected_map.values(), key=_interval_start) - segment = next( - seg - for seg in group_intervals_into_segments(selected_intervals) - if first in {iv["startsAt"] for iv in seg["intervals"]} - ) + _extend_for_min_duration( + selected_map, + candidate_map=candidate_map, + candidates_sorted=candidates_sorted, + candidate_index=candidate_index, + minimum_power_w=minimum_power_w, + charging_efficiency=charging_efficiency, + interval_minutes=interval_minutes, + min_charge_duration_minutes=min_charge_duration_minutes, + warnings=warnings, + ) if max_cycles_per_day: - while True: - selected_intervals = sorted(selected_map.values(), key=_interval_start) - segments = group_intervals_into_segments(selected_intervals) - if len(segments) <= max_cycles_per_day: - break + _merge_for_max_cycles( + selected_map, + candidate_map=candidate_map, + candidates_sorted=candidates_sorted, + candidate_index=candidate_index, + minimum_power_w=minimum_power_w, + charging_efficiency=charging_efficiency, + interval_minutes=interval_minutes, + max_cycles_per_day=max_cycles_per_day, + warnings=warnings, + ) - best_gap: tuple[float, list[dict[str, Any]]] | None = None - for left, right in pairwise(segments): - left_end_index = candidate_index[left["intervals"][-1]["startsAt"]] - right_start_index = candidate_index[right["intervals"][0]["startsAt"]] - gap = candidates_sorted[left_end_index + 1 : right_start_index] - if not gap: - continue - if any(interval["startsAt"] in selected_map for interval in gap): - continue - if any( - _interval_start(gap[index + 1]) - _interval_start(gap[index]) != timedelta(minutes=interval_minutes) - for index in range(len(gap) - 1) - ): - continue - penalty = sum(_sort_price(interval) for interval in gap) - if best_gap is None or penalty < best_gap[0]: - best_gap = (penalty, gap) - - if best_gap is None: - warnings.append("max_cycles_unreachable") - break - - for interval in best_gap[1]: - _add_interval_if_available( - selected_map, - candidate_map, - interval["startsAt"], - power_w=minimum_power_w, - charging_efficiency=charging_efficiency, - interval_minutes=interval_minutes, - ) + if target_grid_energy_kwh is not None: + selected_map = _trim_to_target_energy( + selected_map, + target_grid_energy_kwh=target_grid_energy_kwh, + max_cycles_per_day=max_cycles_per_day, + min_charge_duration_minutes=min_charge_duration_minutes, + interval_minutes=interval_minutes, + protected_starts=protected_starts, + ) selected_intervals = sorted(selected_map.values(), key=_interval_start) segments = group_intervals_into_segments(selected_intervals) diff --git a/custom_components/tibber_prices/services/plan_charging.py b/custom_components/tibber_prices/services/plan_charging.py index 39112ba..865c023 100644 --- a/custom_components/tibber_prices/services/plan_charging.py +++ b/custom_components/tibber_prices/services/plan_charging.py @@ -576,12 +576,23 @@ def _attempt_plan( if schedule["unallocated_grid_energy_kwh"] > 1e-6: return None, "energy_unreachable" + # Deadline-critical intervals (selected to satisfy must_reach_soc by the deadline) must + # never be dropped by segment-constraint trimming, or deadline_met could silently become + # False even though the overall energy target is still satisfied. + protected_starts = ( + frozenset(interval["startsAt"] for interval in schedule["pre_deadline"]["intervals"]) + if "pre_deadline" in schedule + else None + ) + schedule, warnings = apply_segment_constraints( schedule, candidates, charging_efficiency=ctx.charging_efficiency, min_charge_duration_minutes=ctx.min_charge_duration_minutes, max_cycles_per_day=ctx.max_cycles_per_day, + target_grid_energy_kwh=effective_energy_needed_grid_kwh, + protected_starts=protected_starts, interval_minutes=INTERVAL_MINUTES, ) scheduled_intervals = build_soc_progression_from_schedule( diff --git a/tests/services/test_plan_charging.py b/tests/services/test_plan_charging.py index 92c5cb4..787d548 100644 --- a/tests/services/test_plan_charging.py +++ b/tests/services/test_plan_charging.py @@ -203,6 +203,60 @@ async def test_plan_charging_can_meet_deadline_before_peak_period(monkeypatch: p assert response["deadline"]["achieved_soc_kwh"] >= 4.0 +@pytest.mark.asyncio +async def test_plan_charging_max_cycles_does_not_break_deadline(monkeypatch: pytest.MonkeyPatch) -> None: + """Cycle merging/trimming must never discard intervals required to meet a deadline. + + Regression test for a bug where ``apply_segment_constraints`` could remove the + deadline-critical interval while trimming excess energy added by cycle bridging, + silently turning ``deadline_met`` false even though the overall target was reached. + """ + intervals = _make_intervals([0.80, 0.95, 0.95, 0.05]) + fake_tuple = _build_fake_entry_and_coordinator(intervals) + + monkeypatch.setattr(charging_module, "get_entry_and_data", lambda _hass, _entry_id: fake_tuple) + monkeypatch.setattr(charging_module, "resolve_home_timezone", lambda _coord, _home_id: "UTC") + monkeypatch.setattr( + charging_module, + "resolve_search_range", + lambda _call_data, _now, _home_tz: ( + datetime(2026, 1, 1, 0, 0, tzinfo=UTC), + datetime(2026, 1, 1, 1, 0, tzinfo=UTC), + ), + ) + monkeypatch.setattr(charging_module, "get_display_unit_factor", lambda _entry: 1) + monkeypatch.setattr(charging_module, "get_display_unit_string", lambda _entry, _currency: "EUR/kWh") + monkeypatch.setattr(charging_module.dt_util, "now", lambda: datetime(2026, 1, 1, 0, 0, tzinfo=UTC)) + + call = SimpleNamespace( + hass=object(), + data={ + "battery_capacity_kwh": 10.0, + "current_soc_percent": 20.0, + "target_soc_percent": 40.0, + "must_reach_soc_percent": 30.0, + "must_reach_by": datetime(2026, 1, 1, 0, 15, tzinfo=UTC), + "max_charge_power_w": 4000, + "max_cycles_per_day": 1, + "charging_efficiency": 1.0, + "use_base_unit": True, + "allow_relaxation": False, + }, + ) + + response = cast("dict[str, Any]", await handle_plan_charging(cast("ServiceCall", call))) + + assert response["intervals_found"] is True + assert response["charging"]["schedule"]["segment_count"] == 1 + assert response["deadline"]["deadline_met"] is True + assert response["deadline"]["achieved_soc_kwh"] == 3.0 + assert response["battery"]["target_met"] is True + assert response["battery"]["achieved_soc_kwh"] == 4.0 + + scheduled = cast("list[dict[str, Any]]", response["charging"]["schedule"]["intervals"]) + assert scheduled[0]["price"] == 0.8 # the deadline-critical interval must survive trimming + + @pytest.mark.asyncio async def test_plan_charging_can_filter_by_profitability(monkeypatch: pytest.MonkeyPatch) -> None: """Economic filtering should keep only profitable intervals when reserve_for_discharge is enabled.""" @@ -292,7 +346,7 @@ async def test_plan_charging_respects_min_charge_duration(monkeypatch: pytest.Mo @pytest.mark.asyncio async def test_plan_charging_respects_max_cycles_per_day(monkeypatch: pytest.MonkeyPatch) -> None: - """Multiple cheap isolated intervals should be bridged to satisfy max cycle limits.""" + """Cycle merging must not overcharge beyond the requested target energy.""" intervals = _make_intervals([0.10, 0.80, 0.11, 0.90, 0.12, 0.95, 0.50, 0.60]) fake_tuple = _build_fake_entry_and_coordinator(intervals) @@ -327,7 +381,10 @@ async def test_plan_charging_respects_max_cycles_per_day(monkeypatch: pytest.Mon assert response["intervals_found"] is True assert response["charging"]["schedule"]["segment_count"] == 1 + assert response["charging"]["total_energy_kwh"] == 3.0 + assert response["battery"]["achieved_soc_kwh"] == 5.0 + assert response["battery"]["target_met"] is True scheduled = cast("list[dict[str, Any]]", response["charging"]["schedule"]["intervals"]) - assert [interval["price"] for interval in scheduled[:5]] == [0.1, 0.8, 0.11, 0.9, 0.12] + assert [interval["price"] for interval in scheduled] == [0.1, 0.8, 0.11] assert response["warnings"] is None diff --git a/tests/services/test_power_scheduler.py b/tests/services/test_power_scheduler.py index 366a333..b15fb54 100644 --- a/tests/services/test_power_scheduler.py +++ b/tests/services/test_power_scheduler.py @@ -6,7 +6,10 @@ from datetime import UTC, datetime, timedelta from zoneinfo import ZoneInfo from custom_components.tibber_prices.services.charging.deadline_solver import resolve_deadline -from custom_components.tibber_prices.services.charging.power_scheduler import build_power_schedule +from custom_components.tibber_prices.services.charging.power_scheduler import ( + apply_segment_constraints, + build_power_schedule, +) def _make_intervals(prices: list[float]) -> list[dict[str, object]]: @@ -51,6 +54,85 @@ def test_stepped_mode_uses_smallest_sufficient_step() -> None: assert sorted(interval["power_w"] for interval in result["intervals"]) == [2000, 4000, 4000] +def test_apply_segment_constraints_bridges_and_trims_to_target() -> None: + """Bridging isolated cheap intervals must not leave more energy than requested. + + Direct unit-level regression for the ``max_cycles_per_day`` overcharge bug: bridging + fills the gaps between segments with extra (non-essential) intervals, and the + subsequent trim step must remove exactly that surplus again. + """ + candidates = _make_intervals([0.10, 0.80, 0.11, 0.90, 0.12, 0.95, 0.50, 0.60]) + schedule = build_power_schedule(candidates, 3.0, max_charge_power_w=4000, charging_efficiency=1.0) + assert schedule["total_grid_energy_kwh"] == 3.0 + assert len(schedule["segments"]) == 3 # three isolated cheap intervals + + schedule, warnings = apply_segment_constraints( + schedule, + candidates, + charging_efficiency=1.0, + max_cycles_per_day=1, + target_grid_energy_kwh=3.0, + interval_minutes=15, + ) + + assert warnings == [] + assert len(schedule["segments"]) == 1 + assert schedule["total_grid_energy_kwh"] == 3.0 + assert [interval["total"] for interval in schedule["intervals"]] == [0.10, 0.80, 0.11] + + +def test_apply_segment_constraints_never_trims_below_target() -> None: + """Trimming must never drop the plan below the required energy. + + Fixed-power mode cannot reduce the last interval's power, so it can legitimately + overshoot the target by up to one interval's energy. This is expected physical + behavior (not constraint bridging) and must survive trimming unchanged. + """ + candidates = _make_intervals([0.10, 0.20]) + schedule = build_power_schedule(candidates, 1.5, max_charge_power_w=4000, charging_efficiency=1.0) + assert schedule["total_grid_energy_kwh"] == 2.0 # rounded up from 1.5 + + schedule, warnings = apply_segment_constraints( + schedule, + candidates, + charging_efficiency=1.0, + target_grid_energy_kwh=1.5, + interval_minutes=15, + ) + + assert warnings == [] + assert schedule["total_grid_energy_kwh"] == 2.0 # both intervals kept + assert len(schedule["intervals"]) == 2 + + +def test_apply_segment_constraints_respects_protected_starts() -> None: + """Protected intervals (e.g. deadline-critical) must survive trimming. + + Even when a protected interval is the most expensive edge of the merged segment, + trimming must skip it and remove the next-best removable edge instead. + """ + candidates = _make_intervals([0.80, 0.95, 0.95, 0.05]) + protected_start = candidates[0]["startsAt"] + + schedule = build_power_schedule(candidates, 2.0, max_charge_power_w=4000, charging_efficiency=1.0) + assert [interval["total"] for interval in schedule["intervals"]] == [0.80, 0.05] + + schedule, warnings = apply_segment_constraints( + schedule, + candidates, + charging_efficiency=1.0, + max_cycles_per_day=1, + target_grid_energy_kwh=2.0, + protected_starts=frozenset({protected_start}), + interval_minutes=15, + ) + + assert warnings == [] + assert schedule["total_grid_energy_kwh"] == 2.0 + starts = {interval["startsAt"] for interval in schedule["intervals"]} + assert protected_start in starts + + def test_resolve_deadline_next_peak_period() -> None: """Deadline helper should resolve the next future peak period start.""" now = datetime(2026, 1, 1, 0, 0, tzinfo=UTC)