jpawlowski/hass.tibber_prices
1
0
Fork 0
mirror of https://github.com/jpawlowski/hass.tibber_prices.git synced 2026-03-30 21:33:39 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 2
hass.tibber_prices/custom_components/tibber_prices/utils/average.py
Julian Pawlowski 51a99980df feat(sensors)!: add configurable median/mean display for average sensors 
Add user-configurable option to choose between median and arithmetic mean
as the displayed value for all 14 average price sensors, with the alternate
value exposed as attribute.

BREAKING CHANGE: Average sensor default changed from arithmetic mean to
median. Users who rely on arithmetic mean behavior may use the price_mean attribue now, or must manually reconfigure
via Settings → Devices & Services → Tibber Prices → Configure → General
Settings → "Average Sensor Display" → Select "Arithmetic Mean" to get this as sensor state.

Affected sensors (14 total):
- Daily averages: average_price_today, average_price_tomorrow
- 24h windows: trailing_price_average, leading_price_average
- Rolling hour: current_hour_average_price, next_hour_average_price
- Future forecasts: next_avg_3h, next_avg_6h, next_avg_9h, next_avg_12h

Implementation:
- All average calculators now return (mean, median) tuples
- User preference controls which value appears in sensor state
- Alternate value automatically added to attributes
- Period statistics (best_price/peak_price) extended with both values

Technical changes:
- New config option: CONF_AVERAGE_SENSOR_DISPLAY (default: "median")
- Calculator functions return tuples: (avg, median)
- Attribute builders: add_alternate_average_attribute() helper function
- Period statistics: price_avg → price_mean + price_median
- Translations: Updated all 5 languages (de, en, nb, nl, sv)
- Documentation: AGENTS.md, period-calculation.md, recorder-optimization.md

Migration path:
Users can switch back to arithmetic mean via:
Settings → Integrations → Tibber Prices → Configure
→ General Settings → "Average Sensor Display" → "Arithmetic Mean"

Impact: Median is more resistant to price spikes, providing more stable
automation triggers. Statistical analysis from coordinator still uses
arithmetic mean (e.g., trailing_avg_24h for rating calculations).

Co-developed-with: GitHub Copilot <copilot@github.com>
2025-12-08 17:53:40 +00:00

512 lines
16 KiB
Python
Raw Blame History

"""Utility functions for calculating price averages."""
from __future__ import annotations
from datetime import datetime, timedelta
from typing import TYPE_CHECKING
from custom_components.tibber_prices.coordinator.helpers import get_intervals_for_day_offsets
if TYPE_CHECKING:
from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService
def calculate_median(prices: list[float]) -> float | None:
"""
Calculate median from a list of prices.
Args:
prices: List of price values
Returns:
Median price, or None if list is empty
"""
if not prices:
return None
sorted_prices = sorted(prices)
n = len(sorted_prices)
if n % 2 == 0:
# Even number of elements: average of middle two
return (sorted_prices[n // 2 - 1] + sorted_prices[n // 2]) / 2
# Odd number of elements: middle element
return sorted_prices[n // 2]
def calculate_trailing_24h_avg(all_prices: list[dict], interval_start: datetime) -> tuple[float | None, float | None]:
"""
Calculate trailing 24-hour average and median price for a given interval.
Args:
all_prices: List of all price data (yesterday, today, tomorrow combined)
interval_start: Start time of the interval to calculate average for
time: TibberPricesTimeService instance (required)
Returns:
Tuple of (average price, median price) for the 24 hours preceding the interval,
or (None, None) if no data in window
"""
# Define the 24-hour window: from 24 hours before interval_start up to interval_start
window_start = interval_start - timedelta(hours=24)
window_end = interval_start
# Filter prices within the 24-hour window
prices_in_window = []
for price_data in all_prices:
starts_at = price_data["startsAt"] # Already datetime object in local timezone
if starts_at is None:
continue
# Include intervals that start within the window (not including the current interval's end)
if window_start <= starts_at < window_end:
prices_in_window.append(float(price_data["total"]))
# Calculate average and median
# CRITICAL: Return None instead of 0.0 when no data available
# With negative prices, 0.0 could be misinterpreted as a real average value
if prices_in_window:
avg = sum(prices_in_window) / len(prices_in_window)
median = calculate_median(prices_in_window)
return avg, median
return None, None
def calculate_leading_24h_avg(all_prices: list[dict], interval_start: datetime) -> tuple[float | None, float | None]:
"""
Calculate leading 24-hour average and median price for a given interval.
Args:
all_prices: List of all price data (yesterday, today, tomorrow combined)
interval_start: Start time of the interval to calculate average for
time: TibberPricesTimeService instance (required)
Returns:
Tuple of (average price, median price) for up to 24 hours following the interval,
or (None, None) if no data in window
"""
# Define the 24-hour window: from interval_start up to 24 hours after
window_start = interval_start
window_end = interval_start + timedelta(hours=24)
# Filter prices within the 24-hour window
prices_in_window = []
for price_data in all_prices:
starts_at = price_data["startsAt"] # Already datetime object in local timezone
if starts_at is None:
continue
# Include intervals that start within the window
if window_start <= starts_at < window_end:
prices_in_window.append(float(price_data["total"]))
# Calculate average and median
# CRITICAL: Return None instead of 0.0 when no data available
# With negative prices, 0.0 could be misinterpreted as a real average value
if prices_in_window:
avg = sum(prices_in_window) / len(prices_in_window)
median = calculate_median(prices_in_window)
return avg, median
return None, None
def calculate_current_trailing_avg(
coordinator_data: dict,
*,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate the trailing 24-hour average for the current time.
Args:
coordinator_data: The coordinator data containing priceInfo
time: TibberPricesTimeService instance (required)
Returns:
Current trailing 24-hour average price, or None if unavailable
"""
if not coordinator_data:
return None
# Get all intervals (yesterday, today, tomorrow) via helper
all_prices = get_intervals_for_day_offsets(coordinator_data, [-1, 0, 1])
if not all_prices:
return None
now = time.now()
return calculate_trailing_24h_min(all_prices, now, time=time)
def calculate_current_leading_avg(
coordinator_data: dict,
*,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate the leading 24-hour average for the current time.
Args:
coordinator_data: The coordinator data containing priceInfo
time: TibberPricesTimeService instance (required)
Returns:
Current leading 24-hour average price, or None if unavailable
"""
if not coordinator_data:
return None
# Get all intervals (yesterday, today, tomorrow) via helper
all_prices = get_intervals_for_day_offsets(coordinator_data, [-1, 0, 1])
if not all_prices:
return None
now = time.now()
return calculate_leading_24h_min(all_prices, now, time=time)
def calculate_trailing_24h_min(
all_prices: list[dict],
interval_start: datetime,
*,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate trailing 24-hour minimum price for a given interval.
Args:
all_prices: List of all price data (yesterday, today, tomorrow combined)
interval_start: Start time of the interval to calculate minimum for
time: TibberPricesTimeService instance (required)
Returns:
Minimum price for the 24 hours preceding the interval, or None if no data in window
"""
# Define the 24-hour window: from 24 hours before interval_start up to interval_start
window_start = interval_start - timedelta(hours=24)
window_end = interval_start
# Filter prices within the 24-hour window
prices_in_window = []
for price_data in all_prices:
starts_at = time.get_interval_time(price_data)
if starts_at is None:
continue
# Include intervals that start within the window (not including the current interval's end)
if window_start <= starts_at < window_end:
prices_in_window.append(float(price_data["total"]))
# Calculate minimum
# CRITICAL: Return None instead of 0.0 when no data available
# With negative prices, 0.0 could be misinterpreted as a maximum value
if prices_in_window:
return min(prices_in_window)
return None
def calculate_trailing_24h_max(
all_prices: list[dict],
interval_start: datetime,
*,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate trailing 24-hour maximum price for a given interval.
Args:
all_prices: List of all price data (yesterday, today, tomorrow combined)
interval_start: Start time of the interval to calculate maximum for
time: TibberPricesTimeService instance (required)
Returns:
Maximum price for the 24 hours preceding the interval, or None if no data in window
"""
# Define the 24-hour window: from 24 hours before interval_start up to interval_start
window_start = interval_start - timedelta(hours=24)
window_end = interval_start
# Filter prices within the 24-hour window
prices_in_window = []
for price_data in all_prices:
starts_at = time.get_interval_time(price_data)
if starts_at is None:
continue
# Include intervals that start within the window (not including the current interval's end)
if window_start <= starts_at < window_end:
prices_in_window.append(float(price_data["total"]))
# Calculate maximum
# CRITICAL: Return None instead of 0.0 when no data available
# With negative prices, 0.0 could be misinterpreted as a real price value
if prices_in_window:
return max(prices_in_window)
return None
def calculate_leading_24h_min(
all_prices: list[dict],
interval_start: datetime,
*,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate leading 24-hour minimum price for a given interval.
Args:
all_prices: List of all price data (yesterday, today, tomorrow combined)
interval_start: Start time of the interval to calculate minimum for
time: TibberPricesTimeService instance (required)
Returns:
Minimum price for up to 24 hours following the interval, or None if no data in window
"""
# Define the 24-hour window: from interval_start up to 24 hours after
window_start = interval_start
window_end = interval_start + timedelta(hours=24)
# Filter prices within the 24-hour window
prices_in_window = []
for price_data in all_prices:
starts_at = time.get_interval_time(price_data)
if starts_at is None:
continue
# Include intervals that start within the window
if window_start <= starts_at < window_end:
prices_in_window.append(float(price_data["total"]))
# Calculate minimum
# CRITICAL: Return None instead of 0.0 when no data available
# With negative prices, 0.0 could be misinterpreted as a maximum value
if prices_in_window:
return min(prices_in_window)
return None
def calculate_leading_24h_max(
all_prices: list[dict],
interval_start: datetime,
*,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate leading 24-hour maximum price for a given interval.
Args:
all_prices: List of all price data (yesterday, today, tomorrow combined)
interval_start: Start time of the interval to calculate maximum for
time: TibberPricesTimeService instance (required)
Returns:
Maximum price for up to 24 hours following the interval, or None if no data in window
"""
# Define the 24-hour window: from interval_start up to 24 hours after
window_start = interval_start
window_end = interval_start + timedelta(hours=24)
# Filter prices within the 24-hour window
prices_in_window = []
for price_data in all_prices:
starts_at = time.get_interval_time(price_data)
if starts_at is None:
continue
# Include intervals that start within the window
if window_start <= starts_at < window_end:
prices_in_window.append(float(price_data["total"]))
# Calculate maximum
# CRITICAL: Return None instead of 0.0 when no data available
# With negative prices, 0.0 could be misinterpreted as a real price value
if prices_in_window:
return max(prices_in_window)
return None
def calculate_current_trailing_min(
coordinator_data: dict,
*,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate the trailing 24-hour minimum for the current time.
Args:
coordinator_data: The coordinator data containing priceInfo
time: TibberPricesTimeService instance (required)
Returns:
Current trailing 24-hour minimum price, or None if unavailable
"""
if not coordinator_data:
return None
# Get all intervals (yesterday, today, tomorrow) via helper
all_prices = get_intervals_for_day_offsets(coordinator_data, [-1, 0, 1])
if not all_prices:
return None
now = time.now()
return calculate_trailing_24h_min(all_prices, now, time=time)
def calculate_current_trailing_max(
coordinator_data: dict,
*,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate the trailing 24-hour maximum for the current time.
Args:
coordinator_data: The coordinator data containing priceInfo
time: TibberPricesTimeService instance (required)
Returns:
Current trailing 24-hour maximum price, or None if unavailable
"""
if not coordinator_data:
return None
# Get all intervals (yesterday, today, tomorrow) via helper
all_prices = get_intervals_for_day_offsets(coordinator_data, [-1, 0, 1])
if not all_prices:
return None
now = time.now()
return calculate_trailing_24h_max(all_prices, now, time=time)
def calculate_current_leading_min(
coordinator_data: dict,
*,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate the leading 24-hour minimum for the current time.
Args:
coordinator_data: The coordinator data containing priceInfo
time: TibberPricesTimeService instance (required)
Returns:
Current leading 24-hour minimum price, or None if unavailable
"""
if not coordinator_data:
return None
# Get all intervals (yesterday, today, tomorrow) via helper
all_prices = get_intervals_for_day_offsets(coordinator_data, [-1, 0, 1])
if not all_prices:
return None
now = time.now()
# calculate_leading_24h_avg returns (avg, median) - we just need the avg
result = calculate_leading_24h_avg(all_prices, now)
if isinstance(result, tuple):
return result[0] # Return avg only
return None
def calculate_current_leading_max(
coordinator_data: dict,
*,
time: TibberPricesTimeService,
) -> float | None:
"""
Calculate the leading 24-hour maximum for the current time.
Args:
coordinator_data: The coordinator data containing priceInfo
time: TibberPricesTimeService instance (required)
Returns:
Current leading 24-hour maximum price, or None if unavailable
"""
if not coordinator_data:
return None
# Get all intervals (yesterday, today, tomorrow) via helper
all_prices = get_intervals_for_day_offsets(coordinator_data, [-1, 0, 1])
if not all_prices:
return None
now = time.now()
return calculate_leading_24h_max(all_prices, now, time=time)
def calculate_next_n_hours_avg(
coordinator_data: dict,
hours: int,
*,
time: TibberPricesTimeService,
) -> tuple[float | None, float | None]:
"""
Calculate average and median price for the next N hours starting from the next interval.
This function computes the average and median of all 15-minute intervals starting from
the next interval (not current) up to N hours into the future.
Args:
coordinator_data: The coordinator data containing priceInfo
hours: Number of hours to look ahead (1, 2, 3, 4, 5, 6, 8, 12, etc.)
time: TibberPricesTimeService instance (required)
Returns:
Tuple of (average price, median price) for the next N hours,
or (None, None) if insufficient data
"""
if not coordinator_data or hours <= 0:
return None, None
# Get all intervals (yesterday, today, tomorrow) via helper
all_prices = get_intervals_for_day_offsets(coordinator_data, [-1, 0, 1])
if not all_prices:
return None, None
# Find the current interval index
current_idx = None
for idx, price_data in enumerate(all_prices):
starts_at = time.get_interval_time(price_data)
if starts_at is None:
continue
interval_end = starts_at + time.get_interval_duration()
if time.is_current_interval(starts_at, interval_end):
current_idx = idx
break
if current_idx is None:
return None, None
# Calculate how many intervals are in N hours
intervals_needed = time.minutes_to_intervals(hours * 60)
# Collect prices starting from NEXT interval (current_idx + 1)
prices_in_window = []
for offset in range(1, intervals_needed + 1):
idx = current_idx + offset
if idx >= len(all_prices):
# Not enough future data available
break
price = all_prices[idx].get("total")
if price is not None:
prices_in_window.append(float(price))
# Return None if no data at all
if not prices_in_window:
return None, None
# Return average and median (prefer full period, but allow graceful degradation)
avg = sum(prices_in_window) / len(prices_in_window)
median = calculate_median(prices_in_window)
return avg, median
Reference in a new issue View git blame Copy permalink