jpawlowski/hass.tibber_prices
1
0
Fork 0
mirror of https://github.com/jpawlowski/hass.tibber_prices.git synced 2026-05-28 18:43:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

fix(periods): keep negative best-price windows strictly local

Browse source 
Always treat prices at or below zero as valid best-price intervals, rescue short
negative cores with directly adjacent cheap shoulders before min-length filtering,
and block geometric or shape-based widening for periods that already contain a
negative-price core.

Impact: Negative best-price periods no longer expand into positive edge intervals on days with extreme negative prices.
This commit is contained in:
Julian Pawlowski 2026-04-25 20:00:04 +00:00
parent c8f40e0b8a
commit 10c83d6720
6 changed files with 508 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

12
custom_components/tibber_prices/coordinator/period_handlers/core.py
View file

@ -16,6 +16,7 @@ from .period_building import (
add_interval_ends,
build_periods,
calculate_reference_prices,
extend_negative_core_periods_for_min_length,
extend_periods_across_midnight,
filter_periods_by_end_date,
filter_periods_by_min_length,
@ -201,6 +202,17 @@ def calculate_periods(
len(raw_periods),
)
# Step 3.75: Rescue short negative best-price cores before min-length filtering.
# This keeps <= 0 prices as the hard core and only adds directly adjacent cheap
# shoulders when needed to reach the configured minimum length.
if not reverse_sort:
raw_periods = extend_negative_core_periods_for_min_length(
raw_periods,
all_prices_sorted,
min_period_length,
time=time,
)
# Step 4: Filter by minimum length
raw_periods = filter_periods_by_min_length(raw_periods, min_period_length, time=time)
_LOGGER.debug(

30
custom_components/tibber_prices/coordinator/period_handlers/level_filtering.py
View file

@ -132,6 +132,17 @@ def check_interval_criteria(
Tuple of (in_flex, meets_min_distance)
"""
# ============================================================
# FAST PATH: Negative/zero prices always qualify as best price
# ============================================================
# When price ≤ 0 the consumer is paid or gets free electricity.
# This is unconditionally the cheapest possible outcome regardless
# of daily average, flex setting, or level filter.
# Bypasses both flex AND min_distance: a negative price is always
# maximally "far below average" in the economically meaningful sense.
if not criteria.reverse_sort and price <= 0:
return True, True
# Normalize inputs to absolute values for consistent calculation
flex_abs = abs(criteria.flex)
min_distance_abs = abs(criteria.min_distance_from_avg)
@ -143,22 +154,19 @@ def check_interval_criteria(
# - Peak price (reverse_sort=True): daily MAXIMUM
# - Best price (reverse_sort=False): daily MINIMUM
#
# Standard formula (positive daily minimum):
# Flex base = max(price_span, abs(ref_price)):
# - On V-shape days (tiny minimum, large span): span wins → meaningful flex band
# - On flat days (large minimum, small span): ref_price wins → same as before
#
# WHY NOT plain ref_price * flex: When daily_min is a single low outlier
# (e.g., min=1 ct, avg=19 ct), the flex band collapses to near-zero
# (1 ct * 15% = 0.15 ct) and no period of sufficient length can be found.
#
# WHY NOT plain span * flex: On flat days (e.g., min=30 ct, span=3 ct),
# this makes the band much narrower than before, breaking existing behaviour.
#
# Examples with flex=15%:
# - V-shape: min=1 ct, avg=19 ct → span=18 ct → flex_base=18 → threshold=1+2.7=3.7 ct (spans fixed)
# - Flat: min=30 ct, avg=33 ct → span=3 ct → flex_base=30 → threshold=30+4.5=34.5 ct (unchanged)
# - Normal: min=10 ct, avg=20 ct → span=10 ct → flex_base=10 → threshold=10+1.5=11.5 ct (unchanged)
# Examples with flex=15% (positive minimum):
# - V-shape: min=1 ct, avg=19 ct → span=18 ct → flex_base=18 → threshold=1+2.7=3.7 ct
# - Flat: min=30 ct, avg=33 ct → span=3 ct → flex_base=30 → threshold=30+4.5=34.5 ct
# - Normal: min=10 ct, avg=20 ct → span=10 ct → flex_base=10 → threshold=10+1.5=11.5 ct
# Positive shoulders around a short negative core are handled later in the
# raw-period pipeline, where adjacency can be evaluated locally. Keeping the
# interval filter day-agnostic avoids creating a global halo across the whole day.
price_span = abs(criteria.avg_price - criteria.ref_price)
flex_base = max(price_span, abs(criteria.ref_price))

191
custom_components/tibber_prices/coordinator/period_handlers/period_building.py
View file

@ -6,7 +6,7 @@ from datetime import date, datetime, timedelta
import logging
from typing import TYPE_CHECKING, Any
from custom_components.tibber_prices.const import PRICE_LEVEL_MAPPING
from custom_components.tibber_prices.const import PRICE_LEVEL_CHEAP, PRICE_LEVEL_MAPPING, PRICE_LEVEL_VERY_CHEAP
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
@ -19,6 +19,7 @@ _LOGGER_DETAILS = logging.getLogger(__name__ + ".details")
# Module-local log indentation (each module starts at level 0)
INDENT_L0 = "" # Entry point / main function
NEGATIVE_CORE_NO_SHOULDER_INTERVALS = 8 # 2 hours at 15-min resolution
def split_intervals_by_day(
@ -66,6 +67,182 @@ def _trim_trailing_gaps(period: list[dict]) -> list[dict]:
return period
def _build_period_interval(price_data: dict, *, time: TibberPricesTimeService) -> dict | None:
"""Build the internal interval representation used by raw periods."""
starts_at = time.get_interval_time(price_data)
if starts_at is None:
return None
price_original = float(price_data.get("_original_price", price_data["total"]))
return {
"interval_hour": starts_at.hour,
"interval_minute": starts_at.minute,
"interval_time": f"{starts_at.hour:02d}:{starts_at.minute:02d}",
"price": price_original,
"interval_start": starts_at,
"smoothing_was_impactful": False,
"is_level_gap": False,
"geometric_bonus_applied": False,
}
def _get_longest_negative_core_length(period: list[dict]) -> int:
"""Return the longest contiguous run of intervals with price <= 0."""
longest = 0
current = 0
for interval in period:
if float(interval.get("price", 0.0)) <= 0:
current += 1
longest = max(longest, current)
else:
current = 0
return longest
def _collect_contiguous_best_price_side(
interval_index: dict[datetime, dict],
start_cursor: datetime,
step: timedelta,
*,
max_intervals: int,
time: TibberPricesTimeService,
) -> list[dict]:
"""Collect directly adjacent favourable intervals on one side of a negative core."""
for target_level in (PRICE_LEVEL_VERY_CHEAP, PRICE_LEVEL_CHEAP):
additions: list[dict] = []
cursor = start_cursor
for _ in range(max_intervals):
price_data = interval_index.get(cursor)
if price_data is None or price_data.get("level") != target_level:
break
period_interval = _build_period_interval(price_data, time=time)
if period_interval is None:
break
additions.append(period_interval)
cursor += step
if additions:
if step < timedelta(0):
additions.reverse()
return additions
return []
def _select_nearest_extensions(
left_candidates: list[dict],
right_candidates: list[dict],
*,
max_total_additions: int,
) -> tuple[list[dict], list[dict]]:
"""Select the nearest left/right additions until the target length is reached."""
left_nearest = list(reversed(left_candidates))
right_nearest = right_candidates.copy()
selected_left_nearest: list[dict] = []
selected_right: list[dict] = []
prefer_left = bool(left_nearest)
while max_total_additions > 0 and (left_nearest or right_nearest):
if prefer_left and left_nearest:
selected_left_nearest.append(left_nearest.pop(0))
max_total_additions -= 1
elif not prefer_left and right_nearest:
selected_right.append(right_nearest.pop(0))
max_total_additions -= 1
elif left_nearest:
selected_left_nearest.append(left_nearest.pop(0))
max_total_additions -= 1
elif right_nearest:
selected_right.append(right_nearest.pop(0))
max_total_additions -= 1
prefer_left = not prefer_left
return list(reversed(selected_left_nearest)), selected_right
def extend_negative_core_periods_for_min_length(
periods: list[list[dict]],
all_prices: list[dict],
min_period_length: int,
*,
time: TibberPricesTimeService,
) -> list[list[dict]]:
"""Locally extend short negative best-price cores into directly adjacent cheap shoulders.
This rescue step is intentionally narrow:
- only periods that already contain a negative/zero core are considered
- only periods shorter than the configured minimum length are extended
- only directly adjacent VERY_CHEAP/CHEAP intervals may be added
- multi-hour negative blocks stay untouched to preserve a strict negative-only period
"""
if not periods:
return periods
min_intervals = time.minutes_to_intervals(min_period_length)
if min_intervals <= 0:
return periods
interval_index: dict[datetime, dict] = {}
for price_data in all_prices:
starts_at = time.get_interval_time(price_data)
if starts_at is not None:
interval_index[starts_at] = price_data
interval_duration = time.get_interval_duration()
extended_periods: list[list[dict]] = []
for period in periods:
negative_core_length = _get_longest_negative_core_length(period)
if (
negative_core_length == 0
or negative_core_length >= NEGATIVE_CORE_NO_SHOULDER_INTERVALS
or len(period) >= min_intervals
):
extended_periods.append(period)
continue
period_start = period[0].get("interval_start")
period_end = period[-1].get("interval_start")
if period_start is None or period_end is None:
extended_periods.append(period)
continue
needed_intervals = min_intervals - len(period)
left_candidates = _collect_contiguous_best_price_side(
interval_index,
period_start - interval_duration,
-interval_duration,
max_intervals=needed_intervals,
time=time,
)
right_candidates = _collect_contiguous_best_price_side(
interval_index,
period_end + interval_duration,
interval_duration,
max_intervals=needed_intervals,
time=time,
)
selected_left, selected_right = _select_nearest_extensions(
left_candidates,
right_candidates,
max_total_additions=needed_intervals,
)
if selected_left or selected_right:
extended_periods.append([*selected_left, *period, *selected_right])
else:
extended_periods.append(period)
return extended_periods
def build_periods(
all_prices: list[dict],
price_context: dict[str, Any],
@ -144,7 +321,6 @@ def build_periods(
)
for day in ref_prices
}
for price_data in all_prices:
starts_at = time.get_interval_time(price_data)
if starts_at is None:
@ -173,9 +349,16 @@ def build_periods(
# Check flex and minimum distance criteria (using smoothed price and interval's own day reference)
criteria = criteria_by_day[ref_date]
# Compute geometric flex bonus if pattern-aware expansion is enabled
# Compute geometric flex bonus if pattern-aware expansion is enabled.
# Best-price days with a negative daily minimum are handled by the dedicated
# negative-core logic; applying a day-wide geometric valley bonus there would
# reintroduce broad positive shoulders around a negative core.
geo_bonus = 0.0
if geometric_extra_flex > 0 and day_patterns_by_date is not None:
if (
geometric_extra_flex > 0
and day_patterns_by_date is not None
and not (not reverse_sort and criteria.ref_price < 0)
):
day_pattern_for_date = day_patterns_by_date.get(ref_date)
geo_bonus = compute_geometric_flex_bonus(
starts_at,

22
custom_components/tibber_prices/coordinator/period_handlers/shape_extension.py
View file

@ -47,6 +47,7 @@ if TYPE_CHECKING:
from .types import TibberPricesThresholdConfig
_INTERVAL_DURATION = timedelta(minutes=15)
NEGATIVE_CORE_DISABLE_EXTENSION_INTERVALS = 1
def extend_periods_for_shape(
@ -269,6 +270,12 @@ def _extend_period_edges(
# Collect original intervals early – needed for the majority gate below.
original_intervals = _collect_original_intervals(start, end, interval_index)
# Negative-price best-price periods use dedicated core/shoulder handling earlier
# in the pipeline. Do not widen them again here just because adjacent intervals
# are labelled VERY_CHEAP/CHEAP.
if not reverse_sort and _contains_negative_core(original_intervals):
return period
# ── walk LEFT (earlier than period start) ─────────────────────────────────
left_cursor = start - _INTERVAL_DURATION
left_additions = _walk_contiguous(interval_index, left_cursor, backward_step, primary_level, max_intervals)
@ -401,3 +408,18 @@ def _collect_original_intervals(
result.append(iv)
cursor += _INTERVAL_DURATION
return result
def _contains_negative_core(intervals: list[dict[str, Any]]) -> bool:
"""Return True when the period contains at least one negative/zero-price interval."""
negative_run = 0
for interval in intervals:
if float(interval.get("total", 0.0)) <= 0:
negative_run += 1
if negative_run >= NEGATIVE_CORE_DISABLE_EXTENSION_INTERVALS:
return True
else:
negative_run = 0
return False

97
tests/test_level_filtering.py
View file

@ -679,3 +679,100 @@ class TestRealWorldScenarios:
assert in_flex_min is True, "Negative minimum should pass"
assert in_flex_within is True, "Within flex of negative min should pass"
@pytest.mark.unit
class TestNegativePriceFastPath:
"""
Tests for the negative/zero price fast path.
Regression Tests:
- Negative prices not fully included in best price (Apr 2026): Even with
VERY_CHEAP extension and max flex (50%), the old formula excluded large
parts of a negative-price day because the flex band was anchored at the
negative daily minimum. Example: min=-38 ct, flex=50% → threshold=-19 ct,
excluding -18 ct, -10 ct, 0 ct, etc.
"""
def test_negative_price_always_qualifies_best_price(self) -> None:
"""Any negative price unconditionally qualifies as best price (fast path)."""
# Simulate tomorrow: min=-38 ct, avg=-3 ct
criteria = TibberPricesIntervalCriteria(
ref_price=-38.0,
avg_price=-3.0,
flex=0.15,
min_distance_from_avg=5.0,
reverse_sort=False,
)
for price in [-38.0, -20.0, -5.0, -0.01]:
in_flex, meets_distance = check_interval_criteria(price, criteria)
assert in_flex is True, f"Negative price {price} ct should always pass flex"
assert meets_distance is True, f"Negative price {price} ct should always pass distance"
def test_zero_price_always_qualifies_best_price(self) -> None:
"""Zero price unconditionally qualifies as best price (fast path)."""
criteria = TibberPricesIntervalCriteria(
ref_price=-10.0,
avg_price=5.0,
flex=0.15,
min_distance_from_avg=5.0,
reverse_sort=False,
)
in_flex, meets_distance = check_interval_criteria(0.0, criteria)
assert in_flex is True, "Zero price should always pass flex"
assert meets_distance is True, "Zero price should always pass distance"
def test_negative_price_does_not_qualify_peak_price(self) -> None:
"""Negative prices should NOT qualify as peak prices."""
# Even with high average, a negative price is never a "peak"
criteria = TibberPricesIntervalCriteria(
ref_price=30.0,
avg_price=20.0,
flex=0.50,
min_distance_from_avg=5.0,
reverse_sort=True,
)
in_flex, _meets_distance = check_interval_criteria(-5.0, criteria)
assert in_flex is False, "Negative price should fail peak price flex check"
def test_positive_price_is_not_auto_qualified_on_negative_day(self) -> None:
"""Positive prices must still pass the normal interval criteria."""
criteria = TibberPricesIntervalCriteria(
ref_price=-38.0,
avg_price=5.0,
flex=0.15,
min_distance_from_avg=5.0,
reverse_sort=False,
)
in_flex, meets_distance = check_interval_criteria(2.0, criteria)
assert in_flex is False, "Positive prices should not get a day-global negative halo"
assert meets_distance is True, "Distance may still pass even when flex fails"
def test_regression_old_formula_would_exclude_most_intervals(self) -> None:
"""
Regression: the old flex formula (anchored at negative min) excluded
almost all intervals on a day with extreme negative prices.
Old formula (BROKEN):
flex_base = max(|avg - min|, |min|) = max(35, 38) = 38
threshold = -38 + 38 * 0.15 = -32.3 ct
→ only prices ≤ -32.3 ct qualify!
New formula:
Fast path: price ≤ 0 → always (True, True)
Zero-anchored: threshold = abs(-3) * 0.15 = 0.45 ct above zero
→ ALL negative prices qualify, plus tiny halo near 0
"""
criteria = TibberPricesIntervalCriteria(
ref_price=-38.0,
avg_price=-3.0,
flex=0.15,
min_distance_from_avg=5.0,
reverse_sort=False,
)
# All these prices should pass with new logic but would have FAILED with old logic
prices_that_now_pass = [-32.0, -20.0, -10.0, -5.0, -1.0, -0.01, 0.0]
for price in prices_that_now_pass:
in_flex, meets_distance = check_interval_criteria(price, criteria)
assert in_flex is True, f"Price {price} ct should pass with new logic (regression)"
assert meets_distance is True, f"Price {price} ct should pass distance (regression)"

171
tests/test_negative_best_price_periods.py Normal file
View file

@ -0,0 +1,171 @@
"""Regression tests for best-price periods with negative prices."""
from __future__ import annotations
from datetime import datetime
from zoneinfo import ZoneInfo
import pytest
from custom_components.tibber_prices.coordinator.period_handlers.core import calculate_periods
from custom_components.tibber_prices.coordinator.period_handlers.types import TibberPricesPeriodConfig
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
def _create_interval(dt: datetime, price: float, level: str) -> dict:
"""Create a single interval dict."""
rating = "LOW" if price <= 5.0 else "NORMAL"
return {
"startsAt": dt,
"total": price,
"level": level,
"rating_level": rating,
"_original_price": price,
}
def _build_day_with_overrides(overrides: dict[tuple[int, int], tuple[float, str]]) -> list[dict]:
"""Build a day of 15-minute intervals with targeted overrides."""
tz = ZoneInfo("Europe/Berlin")
base = datetime(2025, 4, 25, 0, 0, 0, tzinfo=tz)
intervals: list[dict] = []
for hour in range(24):
for minute in (0, 15, 30, 45):
price, level = overrides.get((hour, minute), (20.0, "NORMAL"))
intervals.append(_create_interval(base.replace(hour=hour, minute=minute), price, level))
return intervals
def _create_time_service() -> TibberPricesTimeService:
"""Create a deterministic time service for period calculation."""
tz = ZoneInfo("Europe/Berlin")
return TibberPricesTimeService(datetime(2025, 4, 25, 12, 0, 0, tzinfo=tz))
def _create_day_pattern(valley_start: tuple[int, int], valley_end: tuple[int, int]) -> dict:
"""Create a minimal day-pattern dict for geometric flex tests."""
tz = ZoneInfo("Europe/Berlin")
base = datetime(2025, 4, 25, 0, 0, 0, tzinfo=tz)
return {
"pattern": "valley",
"confidence": 1.0,
"day_cv_percent": 100.0,
"segments": [],
"extreme_time": base.replace(hour=13, minute=30),
"valley_start": base.replace(hour=valley_start[0], minute=valley_start[1]),
"valley_end": base.replace(hour=valley_end[0], minute=valley_end[1]),
"peak_start": None,
"peak_end": None,
}
@pytest.mark.unit
class TestNegativeBestPricePeriods:
"""Validate local shoulder rescue around short negative cores only."""
def test_short_negative_dip_can_be_rescued_by_local_shoulders(self) -> None:
"""A short negative core may extend into directly adjacent cheap shoulders."""
intervals = _build_day_with_overrides(
{
(10, 45): (5.0, "CHEAP"),
(11, 0): (2.0, "CHEAP"),
(11, 15): (-1.5, "VERY_CHEAP"),
(11, 30): (-1.0, "VERY_CHEAP"),
(11, 45): (2.0, "CHEAP"),
(12, 0): (5.0, "CHEAP"),
}
)
config = TibberPricesPeriodConfig(
reverse_sort=False,
flex=0.15,
min_distance_from_avg=5.0,
min_period_length=60,
)
result = calculate_periods(intervals, config=config, time=_create_time_service())
periods = result["periods"]
assert len(periods) == 1, "Expected the short negative dip to survive as one local period"
assert periods[0]["start"].hour == 11 and periods[0]["start"].minute == 0
assert periods[0]["end"].hour == 12 and periods[0]["end"].minute == 0
assert periods[0]["duration_minutes"] == 60
def test_long_negative_block_stays_negative_only(self) -> None:
"""A multi-hour negative block must not pull in positive shoulders."""
intervals = _build_day_with_overrides(
{
(10, 30): (2.0, "CHEAP"),
(10, 45): (2.0, "CHEAP"),
(11, 0): (-1.5, "VERY_CHEAP"),
(11, 15): (-1.4, "VERY_CHEAP"),
(11, 30): (-1.3, "VERY_CHEAP"),
(11, 45): (-1.2, "VERY_CHEAP"),
(12, 0): (-1.1, "VERY_CHEAP"),
(12, 15): (-1.0, "VERY_CHEAP"),
(12, 30): (-0.9, "VERY_CHEAP"),
(12, 45): (-0.8, "VERY_CHEAP"),
(13, 0): (2.0, "CHEAP"),
(13, 15): (2.0, "CHEAP"),
}
)
config = TibberPricesPeriodConfig(
reverse_sort=False,
flex=0.15,
min_distance_from_avg=5.0,
min_period_length=180,
)
result = calculate_periods(intervals, config=config, time=_create_time_service())
assert result["periods"] == [], "Long negative blocks should not be widened with positive shoulders"
def test_negative_core_ignores_geometric_and_shape_extension(self) -> None:
"""Negative best-price periods must not widen via geometric or shape extension."""
intervals = _build_day_with_overrides(
{
(11, 45): (7.93, "VERY_CHEAP"),
(12, 0): (4.5, "VERY_CHEAP"),
(12, 15): (-1.0, "VERY_CHEAP"),
(12, 30): (-2.0, "VERY_CHEAP"),
(12, 45): (-3.0, "VERY_CHEAP"),
(13, 0): (-4.0, "VERY_CHEAP"),
(13, 15): (-5.36, "VERY_CHEAP"),
(13, 30): (-4.5, "VERY_CHEAP"),
(13, 45): (-3.5, "VERY_CHEAP"),
(14, 0): (-2.5, "VERY_CHEAP"),
(14, 15): (-1.5, "VERY_CHEAP"),
(14, 30): (-0.5, "VERY_CHEAP"),
(14, 45): (2.0, "VERY_CHEAP"),
(15, 0): (4.0, "VERY_CHEAP"),
(15, 15): (7.0, "VERY_CHEAP"),
}
)
config = TibberPricesPeriodConfig(
reverse_sort=False,
flex=0.15,
min_distance_from_avg=5.0,
min_period_length=60,
extend_to_extreme=True,
max_extension_intervals=4,
geometric_extra_flex=0.20,
)
day_patterns_by_date = {
datetime(2025, 4, 25, 0, 0, 0, tzinfo=ZoneInfo("Europe/Berlin")).date(): _create_day_pattern(
(11, 45), (15, 15)
)
}
result = calculate_periods(
intervals,
config=config,
time=_create_time_service(),
day_patterns_by_date=day_patterns_by_date,
)
assert len(result["periods"]) == 1
assert result["periods"][0]["start"].hour == 12 and result["periods"][0]["start"].minute == 15
assert result["periods"][0]["end"].hour == 14 and result["periods"][0]["end"].minute == 45
assert result["periods"][0].get("geometric_extension_active") is None
assert result["periods"][0].get("extension_intervals_added") is None