From 6f5261785bdf4cee8263f057305420aa02bf3811 Mon Sep 17 00:00:00 2001 From: Julian Pawlowski Date: Sun, 12 Apr 2026 14:13:47 +0000 Subject: [PATCH] feat(sensor): add price rank sensors and IQR-based volatility attributes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add three new price rank sensors that show where today's/tomorrow's/combined average price falls relative to all intervals in the evaluated window: - price_rank_today: today's average price percentile rank (0–100%) - price_rank_tomorrow: tomorrow's average price percentile rank - price_rank_today_tomorrow: combined today+tomorrow percentile rank Extend all volatility sensors with IQR-based band statistics: - price_typical_spread: interquartile range (IQR) in currency subunit - price_typical_spread_%: IQR as percentage of daily average - price_spike_count: number of intervals outside Tukey fences (outliers) Add calculate_iqr_stats() utility function in utils/price.py that computes the 25th/75th percentiles, IQR, outer fences (Q1 - 1.5×IQR / Q3 + 1.5×IQR), and outlier count for any list of price values. Entity keys and attribute names use plain language (`price_rank`, `price_typical_spread`) as primary labels; technical terms (percentile rank, IQR) are included parenthetically in descriptions and documentation. Impact: Users can now see where current day prices rank compared to their window and how tightly clustered or spike-prone a day's prices are. --- .../sensor/attributes/__init__.py | 6 +- .../sensor/attributes/volatility.py | 51 ++++++++++ .../sensor/calculators/volatility.py | 99 ++++++++++++++++++- .../tibber_prices/sensor/core.py | 3 + .../tibber_prices/sensor/definitions.py | 50 ++++++++++ .../tibber_prices/sensor/value_getters.py | 6 ++ .../tibber_prices/utils/price.py | 99 +++++++++++++++++++ 7 files changed, 308 insertions(+), 6 deletions(-) diff --git a/custom_components/tibber_prices/sensor/attributes/__init__.py b/custom_components/tibber_prices/sensor/attributes/__init__.py index ac14568..6638a86 100644 --- a/custom_components/tibber_prices/sensor/attributes/__init__.py +++ b/custom_components/tibber_prices/sensor/attributes/__init__.py @@ -47,7 +47,7 @@ from .lifecycle import build_lifecycle_attributes from .metadata import get_day_pattern_attributes from .timing import _is_timing_or_volatility_sensor from .trend import _add_cached_trend_attributes, _add_timing_or_volatility_attributes -from .volatility import add_volatility_type_attributes, get_prices_for_volatility +from .volatility import add_percentile_rank_attributes, add_volatility_type_attributes, get_prices_for_volatility from .window_24h import add_average_price_attributes __all__ = [ @@ -65,6 +65,7 @@ __all__ = [ "TrendAttributes", "VolatilityAttributes", "Window24hAttributes", + "add_percentile_rank_attributes", "add_volatility_type_attributes", "build_extra_state_attributes", "build_sensor_attributes", @@ -190,6 +191,9 @@ def build_sensor_attributes( # noqa: PLR0912 elif _is_timing_or_volatility_sensor(key): _add_timing_or_volatility_attributes(attributes, key, cached_data, native_value, time=time) + elif key in ("price_rank_today", "price_rank_tomorrow", "price_rank_today_tomorrow"): + add_percentile_rank_attributes(attributes, cached_data, time=time) + elif key in ("day_pattern_yesterday", "day_pattern_today", "day_pattern_tomorrow"): day = key.removeprefix("day_pattern_") day_attrs = get_day_pattern_attributes(coordinator, day) diff --git a/custom_components/tibber_prices/sensor/attributes/volatility.py b/custom_components/tibber_prices/sensor/attributes/volatility.py index 9a25c17..f92d310 100644 --- a/custom_components/tibber_prices/sensor/attributes/volatility.py +++ b/custom_components/tibber_prices/sensor/attributes/volatility.py @@ -164,3 +164,54 @@ def add_volatility_type_attributes( # Add time window info now = time.now() volatility_attributes["timestamp"] = now + + +def add_percentile_rank_attributes( + attributes: dict, + cached_data: dict, + *, + time: TibberPricesTimeService, +) -> None: + """ + Add attributes for percentile rank sensors. + + Sets the timestamp based on the percentile type stored in cached_data: + - "today" / "today_tomorrow": today's first interval start (midnight context) + - "tomorrow": tomorrow's first interval start + + Args: + attributes: Dictionary to add attributes to + cached_data: Dictionary containing cached sensor data (percentile_rank_attributes, + percentile_rank_type, coordinator_data) + time: TibberPricesTimeService instance (required) + + """ + from datetime import timedelta # noqa: PLC0415 - local import to avoid circular + + rank_attrs = cached_data.get("percentile_rank_attributes") + if rank_attrs: + attributes.update(rank_attrs) + + # Set timestamp based on period type + percentile_type = cached_data.get("percentile_rank_type", "today") + coordinator_data = cached_data.get("coordinator_data") + + if coordinator_data: + from custom_components.tibber_prices.coordinator.helpers import ( # noqa: PLC0415 + get_intervals_for_day_offsets, + ) + + all_intervals = get_intervals_for_day_offsets(coordinator_data, [-1, 0, 1]) + now = time.now() + today_date = now.date() + tomorrow_date = (now + timedelta(days=1)).date() + + if percentile_type == "tomorrow": + tomorrow_data = [p for p in all_intervals if p.get("startsAt") and p["startsAt"].date() == tomorrow_date] + if tomorrow_data: + attributes["timestamp"] = tomorrow_data[0].get("startsAt") + else: + # today / today_tomorrow → use today's midnight + today_data = [p for p in all_intervals if p.get("startsAt") and p["startsAt"].date() == today_date] + if today_data: + attributes["timestamp"] = today_data[0].get("startsAt") diff --git a/custom_components/tibber_prices/sensor/calculators/volatility.py b/custom_components/tibber_prices/sensor/calculators/volatility.py index 3057683..c250337 100644 --- a/custom_components/tibber_prices/sensor/calculators/volatility.py +++ b/custom_components/tibber_prices/sensor/calculators/volatility.py @@ -2,6 +2,7 @@ from __future__ import annotations +import bisect from typing import TYPE_CHECKING from custom_components.tibber_prices.const import ( @@ -19,7 +20,11 @@ from custom_components.tibber_prices.sensor.attributes import ( get_prices_for_volatility, ) from custom_components.tibber_prices.utils.average import calculate_mean -from custom_components.tibber_prices.utils.price import calculate_volatility_with_cv +from custom_components.tibber_prices.utils.price import ( + calculate_iqr_stats, + calculate_percentile_rank, + calculate_volatility_with_cv, +) from .base import TibberPricesBaseCalculator @@ -46,6 +51,7 @@ class TibberPricesVolatilityCalculator(TibberPricesBaseCalculator): """ super().__init__(*args, **kwargs) self._last_volatility_attributes: dict[str, Any] = {} + self._last_percentile_rank_attributes: dict[str, Any] = {} def get_volatility_value(self, *, volatility_type: str) -> str | None: """ @@ -101,17 +107,33 @@ class TibberPricesVolatilityCalculator(TibberPricesBaseCalculator): # Calculate volatility level AND coefficient of variation volatility, cv = calculate_volatility_with_cv(prices_to_analyze, **thresholds) + # Calculate IQR statistics (robust to outliers) + iqr_stats = calculate_iqr_stats(prices_to_analyze) + # Store attributes for this sensor - self._last_volatility_attributes = { - "price_spread": round(spread_display, 2), - "price_coefficient_variation_%": round(cv, 2) if cv is not None else None, + # Build attributes with all price_* together, interval_count last + attrs: dict[str, Any] = { "price_volatility": volatility.lower(), + "price_coefficient_variation_%": round(cv, 2) if cv is not None else None, + "price_spread": round(spread_display, 2), "price_min": round(price_min * factor, 2), "price_max": round(price_max * factor, 2), "price_mean": round(price_mean * factor, 2), - "interval_count": len(prices_to_analyze), } + # Add IQR attributes when enough data is available (stay in price_* group) + if iqr_stats is not None: + attrs["price_median"] = round(iqr_stats["median"] * factor, 2) + attrs["price_q25"] = round(iqr_stats["q25"] * factor, 2) + attrs["price_q75"] = round(iqr_stats["q75"] * factor, 2) + attrs["price_typical_spread"] = round(iqr_stats["iqr"] * factor, 2) + if iqr_stats["iqr_pct"] is not None: + attrs["price_typical_spread_%"] = round(iqr_stats["iqr_pct"], 2) + attrs["price_spike_count"] = iqr_stats["outlier_count"] + + attrs["interval_count"] = len(prices_to_analyze) + self._last_volatility_attributes = attrs + # Add icon_color for dynamic styling add_icon_color_attribute(self._last_volatility_attributes, key="volatility", state_value=volatility) @@ -136,3 +158,70 @@ class TibberPricesVolatilityCalculator(TibberPricesBaseCalculator): """ return self._last_volatility_attributes + + def get_percentile_rank_value(self, *, percentile_type: str) -> float | None: + """ + Calculate the percentile rank of the current price within a reference set. + + The result is 0-100: percentage of reference prices strictly cheaper than + the current interval price. 0% = cheapest, ~99% = most expensive. + + Also stores detailed attributes in self._last_percentile_rank_attributes + for use in extra_state_attributes. + + Args: + percentile_type: One of "today", "tomorrow", "today_tomorrow". + + Returns: + Percentile rank (0.0-100.0) or None if unavailable. + + """ + if not self.has_data(): + return None + + # Get current interval price + current_interval = self.coordinator.get_current_interval() + if current_interval is None: + return None + current_price_raw = current_interval.get("total") + if current_price_raw is None: + return None + current_price = float(current_price_raw) + + # Get reference prices for this type (reuse volatility helper) + reference_prices = get_prices_for_volatility( + percentile_type, + self.coordinator.data, + time=self.coordinator.time, + ) + if not reference_prices: + return None + + # Calculate percentile rank + rank = calculate_percentile_rank(current_price, reference_prices) + if rank is None: + return None + + # Convert to display units for attribute storage + factor = get_display_unit_factor(self.config_entry) + + self._last_percentile_rank_attributes = { + "current_price": round(current_price * factor, 2), + "prices_below_count": bisect.bisect_left(sorted(reference_prices), current_price), + "interval_count": len(reference_prices), + "reference_min": round(min(reference_prices) * factor, 2), + "reference_max": round(max(reference_prices) * factor, 2), + "reference_mean": round(calculate_mean(reference_prices) * factor, 2), + } + + return rank + + def get_percentile_rank_attributes(self) -> dict[str, Any]: + """ + Get stored percentile rank attributes from last calculation. + + Returns: + Dictionary of percentile rank attributes, or empty dict if no calculation yet. + + """ + return self._last_percentile_rank_attributes diff --git a/custom_components/tibber_prices/sensor/core.py b/custom_components/tibber_prices/sensor/core.py index 43add84..71fdb97 100644 --- a/custom_components/tibber_prices/sensor/core.py +++ b/custom_components/tibber_prices/sensor/core.py @@ -1164,6 +1164,9 @@ class TibberPricesSensor(TibberPricesEntity, RestoreSensor): "current_trend_attributes": self._trend_calculator.get_current_trend_attributes(), "trend_change_attributes": self._trend_calculator.get_trend_change_attributes(), "volatility_attributes": self._volatility_calculator.get_volatility_attributes(), + "percentile_rank_attributes": self._volatility_calculator.get_percentile_rank_attributes(), + "percentile_rank_type": key.removeprefix("price_rank_") if key.startswith("price_rank_") else None, + "coordinator_data": self.coordinator.data, "last_extreme_interval": self._daily_stat_calculator.get_last_extreme_interval(), "last_energy_tax_averages": self._daily_stat_calculator.get_last_energy_tax_averages(), "last_price_level": self._interval_calculator.get_last_price_level(), diff --git a/custom_components/tibber_prices/sensor/definitions.py b/custom_components/tibber_prices/sensor/definitions.py index a9ad3b1..26ff87a 100644 --- a/custom_components/tibber_prices/sensor/definitions.py +++ b/custom_components/tibber_prices/sensor/definitions.py @@ -736,6 +736,55 @@ VOLATILITY_SENSORS = ( ), ) +# ---------------------------------------------------------------------------- +# 6b. PRICE PERCENTILE RANK SENSORS +# ---------------------------------------------------------------------------- +# These sensors show where the current price ranks within a reference period. +# The state (0-100%) answers: "What percentage of reference prices are cheaper +# than the current price?" +# +# 0% = current price is the cheapest in the reference period +# 50% = half the prices are cheaper (current price at median level) +# ~99% = almost everything is cheaper (current price near the maximum) +# +# Reference periods: +# - today: 96 intervals of today (local calendar day) +# - tomorrow: 96 intervals of tomorrow (once data is available) +# - today_tomorrow: 192 combined intervals when tomorrow is available +# +# Use case: "Is now the right time to run a large appliance?" +# - price_rank_today < 25 → bottom quartile, great time to use energy +# - price_rank_today > 75 → top quartile, consider delaying consumption + +PERCENTILE_RANK_SENSORS = ( + SensorEntityDescription( + key="price_rank_today", + translation_key="price_rank_today", + icon="mdi:percent", + native_unit_of_measurement=PERCENTAGE, + state_class=None, # Position metric: no statistics + suggested_display_precision=0, + ), + SensorEntityDescription( + key="price_rank_tomorrow", + translation_key="price_rank_tomorrow", + icon="mdi:percent", + native_unit_of_measurement=PERCENTAGE, + state_class=None, # Position metric: no statistics + suggested_display_precision=0, + entity_registry_enabled_default=False, # Available once tomorrow's data arrives + ), + SensorEntityDescription( + key="price_rank_today_tomorrow", + translation_key="price_rank_today_tomorrow", + icon="mdi:percent", + native_unit_of_measurement=PERCENTAGE, + state_class=None, # Position metric: no statistics + suggested_display_precision=0, + entity_registry_enabled_default=False, # Advanced overview use case + ), +) + # ---------------------------------------------------------------------------- # 7. BEST/PEAK PRICE TIMING SENSORS (period-based time tracking) # ---------------------------------------------------------------------------- @@ -1116,6 +1165,7 @@ ENTITY_DESCRIPTIONS = ( *FUTURE_TREND_SENSORS, *PRICE_TRAJECTORY_SENSORS, *VOLATILITY_SENSORS, + *PERCENTILE_RANK_SENSORS, *BEST_PRICE_TIMING_SENSORS, *PEAK_PRICE_TIMING_SENSORS, *DAY_PATTERN_SENSORS, diff --git a/custom_components/tibber_prices/sensor/value_getters.py b/custom_components/tibber_prices/sensor/value_getters.py index d833e07..68a2b5b 100644 --- a/custom_components/tibber_prices/sensor/value_getters.py +++ b/custom_components/tibber_prices/sensor/value_getters.py @@ -249,6 +249,12 @@ def get_value_getter_mapping( # noqa: PLR0913 - needs all calculators as parame "today_tomorrow_volatility": lambda: volatility_calculator.get_volatility_value( volatility_type="today_tomorrow" ), + # Price rank sensors (via VolatilityCalculator - reuses same price extraction) + "price_rank_today": lambda: volatility_calculator.get_percentile_rank_value(percentile_type="today"), + "price_rank_tomorrow": lambda: volatility_calculator.get_percentile_rank_value(percentile_type="tomorrow"), + "price_rank_today_tomorrow": lambda: volatility_calculator.get_percentile_rank_value( + percentile_type="today_tomorrow" + ), # ================================================================ # BEST/PEAK PRICE TIMING SENSORS - via TimingCalculator # ================================================================ diff --git a/custom_components/tibber_prices/utils/price.py b/custom_components/tibber_prices/utils/price.py index 77e3f6a..54ac98a 100644 --- a/custom_components/tibber_prices/utils/price.py +++ b/custom_components/tibber_prices/utils/price.py @@ -2,6 +2,7 @@ from __future__ import annotations +import bisect import logging import statistics from datetime import datetime, timedelta @@ -176,6 +177,104 @@ def calculate_volatility_level( return level +MIN_PRICES_FOR_IQR = 4 # Minimum price values needed for meaningful IQR calculation + + +def calculate_iqr_stats(prices: list[float]) -> dict[str, Any] | None: + """ + Calculate Interquartile Range (IQR) statistics from a price list. + + IQR = Q75 - Q25, representing the spread of the central 50% of prices. + This is more robust to outliers than coefficient of variation because + extreme values (price spikes or negative prices) don't distort the result. + + Args: + prices: List of price values (in any unit, e.g. EUR or NOK per kWh) + + Returns: + Dict with keys: + - q25: 25th percentile (lower quartile) + - median: 50th percentile (median) + - q75: 75th percentile (upper quartile) + - iqr: Interquartile range (q75 - q25) + - iqr_pct: Relative IQR as percentage of median (None if median is 0) + - outlier_count: Intervals outside Tukey fences [Q25 - 1.5xIQR, Q75 + 1.5xIQR] + Returns None if fewer than MIN_PRICES_FOR_IQR prices are provided. + + Examples: + - iqr_pct ~5%: Very tight price band, stability similar to CV + - iqr_pct ~20%: Moderate spread in the core price range + - iqr_pct ~50%: Wide core spread, significant optimization potential + - outlier_count > 0: Isolated price spikes/dips exist (CV-heavy days) + + """ + if len(prices) < MIN_PRICES_FOR_IQR: + return None + + quartiles = statistics.quantiles(prices, n=4) # Returns [Q25, Q50, Q75] + q25 = quartiles[0] + median = quartiles[1] + q75 = quartiles[2] + iqr = q75 - q25 + + # Relative IQR: normalized by median for cross-price-level comparison + iqr_pct = (iqr / abs(median) * 100) if median != 0 else None + + # Tukey fence outlier detection (standard method) + lower_fence = q25 - 1.5 * iqr + upper_fence = q75 + 1.5 * iqr + outlier_count = sum(1 for p in prices if p < lower_fence or p > upper_fence) + + return { + "q25": q25, + "median": median, + "q75": q75, + "iqr": iqr, + "iqr_pct": iqr_pct, + "outlier_count": outlier_count, + } + + +def calculate_percentile_rank(current_price: float, prices: list[float]) -> float | None: + """ + Calculate where the current price ranks among a reference price set. + + Returns the percentage of prices in the reference set that are strictly + cheaper than current_price. A value of 0% means the current price is at + or below the cheapest reference price; ~99% means nearly everything is + cheaper (current price near the maximum). + + The current interval's own price is included in today's reference set, + so the cheapest interval of the day always returns 0%. + + Args: + current_price: The price to rank (any unit, must match prices unit) + prices: Reference price list to rank against + + Returns: + Percentile rank as float 0.0-100.0 (1 decimal precision), or None if + reference list is empty. + + Examples (8 intervals: [8, 10, 12, 15, 15, 18, 20, 22]): + - current=8: 0/8 x 100 = 0.0% (cheapest) + - current=15: 3/8 x 100 = 37.5% (above 3 cheaper intervals) + - current=22: 7/8 x 100 = 87.5% (most expensive) + + Note: + Equal prices: All duplicate prices at the current level are counted as + "not below" the current price (bisect_left semantics). This matches + automation logic: "is now cheap?" returns False if current == minimum + is debatable but ensures 0% always means strictly cheapest. + + """ + if not prices: + return None + + sorted_prices = sorted(prices) + count_below = bisect.bisect_left(sorted_prices, current_price) + return round(count_below / len(sorted_prices) * 100, 1) + + def calculate_trailing_average_for_interval( interval_start: datetime, all_prices: list[dict[str, Any]],