mirror of
https://github.com/jpawlowski/hass.tibber_prices.git
synced 2026-05-28 18:43:40 +00:00
feat(sensor): add price rank sensors and IQR-based volatility attributes
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.
This commit is contained in:
Julian Pawlowski
parent
c89248d493
commit
6f5261785b
7 changed files with 308 additions and 6 deletions
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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(),
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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
|
||||
# ================================================================
|
||||
|
|
|
|||
|
|
@ -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]],
|
||||
|
|
|
|||
Loading…
Reference in a new issue