"""Utility functions for price data calculations.""" from __future__ import annotations import logging import statistics from datetime import datetime, timedelta from typing import TYPE_CHECKING, Any if TYPE_CHECKING: from custom_components.tibber_prices.coordinator.time_service import TibberPricesTimeService from custom_components.tibber_prices.const import ( DEFAULT_VOLATILITY_THRESHOLD_HIGH, DEFAULT_VOLATILITY_THRESHOLD_MODERATE, DEFAULT_VOLATILITY_THRESHOLD_VERY_HIGH, PRICE_LEVEL_MAPPING, PRICE_LEVEL_NORMAL, PRICE_RATING_NORMAL, VOLATILITY_HIGH, VOLATILITY_LOW, VOLATILITY_MODERATE, VOLATILITY_VERY_HIGH, ) from custom_components.tibber_prices.coordinator.helpers import get_intervals_for_day_offsets _LOGGER = logging.getLogger(__name__) MIN_PRICES_FOR_VOLATILITY = 2 # Minimum number of price values needed for volatility calculation # Volatility factors for adaptive trend thresholds # These multipliers adjust the base trend thresholds based on price volatility. # The volatility *ranges* are user-configurable (threshold_moderate, threshold_high), # but the *reaction strength* (factors) is fixed for predictable behavior. # This separation allows users to adjust volatility classification without # unexpectedly changing trend sensitivity. # # Factor selection based on lookahead volatility: # - Below moderate threshold (e.g., <15%): Use 0.6 → 40% more sensitive # - Moderate to high (e.g., 15-30%): Use 1.0 → as configured by user # - High and above (e.g., ≥30%): Use 1.4 → 40% less sensitive (filters noise) VOLATILITY_FACTOR_SENSITIVE = 0.6 # Low volatility → more responsive VOLATILITY_FACTOR_NORMAL = 1.0 # Moderate volatility → baseline VOLATILITY_FACTOR_INSENSITIVE = 1.4 # High volatility → noise filtering def calculate_volatility_level( prices: list[float], threshold_moderate: float | None = None, threshold_high: float | None = None, threshold_very_high: float | None = None, ) -> str: """ Calculate volatility level from price list using coefficient of variation. Volatility indicates how much prices fluctuate during a period, which helps determine whether active load shifting is worthwhile. Uses the coefficient of variation (CV = std_dev / mean * 100%) for relative comparison that works across different price levels and period lengths. Args: prices: List of price values (in any unit, typically major currency units like EUR or NOK) threshold_moderate: Custom threshold for MODERATE level (default: use DEFAULT_VOLATILITY_THRESHOLD_MODERATE) threshold_high: Custom threshold for HIGH level (default: use DEFAULT_VOLATILITY_THRESHOLD_HIGH) threshold_very_high: Custom threshold for VERY_HIGH level (default: use DEFAULT_VOLATILITY_THRESHOLD_VERY_HIGH) Returns: Volatility level: "LOW", "MODERATE", "HIGH", or "VERY_HIGH" (uppercase) Examples: - CV < 15%: LOW → minimal optimization potential, prices relatively stable - 15% ≤ CV < 30%: MODERATE → some optimization worthwhile, noticeable variation - 30% ≤ CV < 50%: HIGH → strong optimization recommended, significant swings - CV ≥ 50%: VERY_HIGH → maximum optimization potential, extreme volatility Note: Requires at least 2 price values for calculation. Returns LOW if insufficient data. Works identically for short periods (2-3 intervals) and long periods (96 intervals/day). """ # Need at least 2 values for standard deviation if len(prices) < MIN_PRICES_FOR_VOLATILITY: return VOLATILITY_LOW # Use provided thresholds or fall back to constants t_moderate = threshold_moderate if threshold_moderate is not None else DEFAULT_VOLATILITY_THRESHOLD_MODERATE t_high = threshold_high if threshold_high is not None else DEFAULT_VOLATILITY_THRESHOLD_HIGH t_very_high = threshold_very_high if threshold_very_high is not None else DEFAULT_VOLATILITY_THRESHOLD_VERY_HIGH # Calculate coefficient of variation # CRITICAL: Use absolute value of mean for negative prices (Norway/Germany) # Negative electricity prices are valid and should have measurable volatility mean = statistics.mean(prices) if mean == 0: # Division by zero case (all prices exactly zero) return VOLATILITY_LOW std_dev = statistics.stdev(prices) coefficient_of_variation = (std_dev / abs(mean)) * 100 # As percentage, use abs(mean) # Classify based on thresholds if coefficient_of_variation < t_moderate: return VOLATILITY_LOW if coefficient_of_variation < t_high: return VOLATILITY_MODERATE if coefficient_of_variation < t_very_high: return VOLATILITY_HIGH return VOLATILITY_VERY_HIGH def calculate_trailing_average_for_interval( interval_start: datetime, all_prices: list[dict[str, Any]], ) -> float | None: """ Calculate the trailing 24-hour average price for a specific interval. Args: interval_start: The start time of the interval we're calculating for all_prices: List of all available price intervals (yesterday + today + tomorrow) Returns: The average price of all intervals in the 24 hours before interval_start, or None if insufficient data is available. """ if not all_prices: return None # Calculate the lookback period: 24 hours before this interval lookback_start = interval_start - timedelta(hours=24) # Collect all prices that fall within the 24-hour lookback window matching_prices = [] for price_data in all_prices: price_time = price_data.get("startsAt") # Already datetime object in local timezone if not price_time: continue # Check if this price falls within our lookback window # Include prices that start >= lookback_start and start < interval_start if lookback_start <= price_time < interval_start: total_price = price_data.get("total") if total_price is not None: matching_prices.append(float(total_price)) if not matching_prices: return None # CRITICAL: Warn if we have less than 24 hours of data (partial average) # 24 hours = 96 intervals (4 per hour) # Note: This is expected for intervals in the first 24h window # Calculate and return the average return sum(matching_prices) / len(matching_prices) def calculate_difference_percentage( current_interval_price: float, trailing_average: float | None, ) -> float | None: """ Calculate the difference percentage between current price and trailing average. This mimics the API's "difference" field from priceRating endpoint. CRITICAL: For negative averages, use absolute value to get meaningful percentage. Example: current=10 ct, average=-10 ct - Wrong: (10-(-10))/-10 = -200% (would rate as "cheap" despite being expensive) - Right: (10-(-10))/abs(-10) = +200% (correctly rates as "expensive") Args: current_interval_price: The current interval's price trailing_average: The 24-hour trailing average price Returns: The percentage difference: ((current - average) / abs(average)) * 100 or None if trailing_average is None or zero. """ if trailing_average is None or trailing_average == 0: return None # Use absolute value of average to handle negative prices correctly return ((current_interval_price - trailing_average) / abs(trailing_average)) * 100 def calculate_rating_level( difference: float | None, threshold_low: float, threshold_high: float, ) -> str | None: """ Calculate the rating level based on difference percentage and thresholds. This mimics the API's "level" field from priceRating endpoint. Args: difference: The difference percentage (from calculate_difference_percentage) threshold_low: The low threshold percentage (typically -100 to 0) threshold_high: The high threshold percentage (typically 0 to 100) Returns: "LOW" if difference <= threshold_low "HIGH" if difference >= threshold_high "NORMAL" otherwise None if difference is None """ if difference is None: return None # CRITICAL: Validate threshold configuration # threshold_low must be less than threshold_high for meaningful classification if threshold_low >= threshold_high: _LOGGER.warning( "Invalid rating thresholds: threshold_low (%.2f) >= threshold_high (%.2f). " "Using NORMAL as fallback. Please check configuration.", threshold_low, threshold_high, ) return PRICE_RATING_NORMAL # Classify based on thresholds if difference <= threshold_low: return "LOW" if difference >= threshold_high: return "HIGH" return PRICE_RATING_NORMAL def _process_price_interval( price_interval: dict[str, Any], all_prices: list[dict[str, Any]], threshold_low: float, threshold_high: float, ) -> None: """ Process a single price interval and add difference and rating_level. Args: price_interval: The price interval to process (modified in place) all_prices: All available price intervals for lookback calculation threshold_low: Low threshold percentage threshold_high: High threshold percentage day_label: Label for logging ("today" or "tomorrow") """ starts_at = price_interval.get("startsAt") # Already datetime object in local timezone if not starts_at: return current_interval_price = price_interval.get("total") if current_interval_price is None: return # Calculate trailing average trailing_avg = calculate_trailing_average_for_interval(starts_at, all_prices) # Calculate and set the difference and rating_level if trailing_avg is not None: difference = calculate_difference_percentage(float(current_interval_price), trailing_avg) price_interval["difference"] = difference # Calculate rating_level based on difference rating_level = calculate_rating_level(difference, threshold_low, threshold_high) price_interval["rating_level"] = rating_level else: # Set to None if we couldn't calculate (expected for intervals in first 24h) price_interval["difference"] = None price_interval["rating_level"] = None def enrich_price_info_with_differences( all_intervals: list[dict[str, Any]], *, threshold_low: float | None = None, threshold_high: float | None = None, time: TibberPricesTimeService | None = None, # noqa: ARG001 # Used in production (via coordinator), kept for compatibility ) -> list[dict[str, Any]]: """ Enrich price intervals with calculated 'difference' and 'rating_level' values. Computes the trailing 24-hour average, difference percentage, and rating level for intervals that have sufficient lookback data (in-place modification). CRITICAL: Only enriches intervals that have at least 24 hours of prior data available. This is determined by checking if (interval_start - earliest_interval_start) >= 24h. Works independently of interval density (24 vs 96 intervals/day) and handles transition periods (e.g., Oct 1, 2025) correctly. Args: all_intervals: Flat list of all price intervals (day_before_yesterday + yesterday + today + tomorrow). threshold_low: Low threshold percentage for rating_level (defaults to -10) threshold_high: High threshold percentage for rating_level (defaults to 10) time: TibberPricesTimeService instance (kept for API compatibility, not used) Returns: Same list (modified in-place) with 'difference' and 'rating_level' added to intervals that have full 24h lookback data. Intervals within the first 24 hours remain unenriched. Note: Interval density changed on Oct 1, 2025 from 24 to 96 intervals/day. This function works correctly across this transition by using time-based rather than count-based logic. """ threshold_low = threshold_low if threshold_low is not None else -10 threshold_high = threshold_high if threshold_high is not None else 10 if not all_intervals: return all_intervals # Find the earliest interval timestamp (start of available data) earliest_start: datetime | None = None for interval in all_intervals: starts_at = interval.get("startsAt") if starts_at and (earliest_start is None or starts_at < earliest_start): earliest_start = starts_at if earliest_start is None: # No valid intervals - return as-is return all_intervals # Calculate the 24-hour boundary from earliest data # Only intervals starting at or after this boundary have full 24h lookback enrichment_boundary = earliest_start + timedelta(hours=24) # Process intervals (modifies in-place) # CRITICAL: Only enrich intervals that start >= 24h after earliest data enriched_count = 0 skipped_count = 0 for price_interval in all_intervals: starts_at = price_interval.get("startsAt") if not starts_at: skipped_count += 1 continue # Skip if interval doesn't have full 24h lookback if starts_at < enrichment_boundary: skipped_count += 1 continue _process_price_interval(price_interval, all_intervals, threshold_low, threshold_high) enriched_count += 1 return all_intervals def find_price_data_for_interval( coordinator_data: dict, target_time: datetime, *, time: TibberPricesTimeService, ) -> dict | None: """ Find the price data for a specific 15-minute interval timestamp. Args: coordinator_data: The coordinator data dict target_time: The target timestamp to find price data for time: TibberPricesTimeService instance (required) Returns: Price data dict if found, None otherwise """ # Round to nearest quarter-hour to handle edge cases where we're called # slightly before the boundary (e.g., 14:59:59.999 → 15:00:00) rounded_time = time.round_to_nearest_quarter(target_time) rounded_date = rounded_time.date() # Get all intervals (yesterday, today, tomorrow) via helper all_intervals = get_intervals_for_day_offsets(coordinator_data, [-1, 0, 1]) # Search for matching interval for price_data in all_intervals: starts_at = time.get_interval_time(price_data) if starts_at is None: continue # Exact match after rounding (both time and date must match) if starts_at == rounded_time and starts_at.date() == rounded_date: return price_data return None def aggregate_price_levels(levels: list[str]) -> str: """ Aggregate multiple price levels into a single representative level using median. Takes a list of price level strings (e.g., "VERY_CHEAP", "NORMAL", "EXPENSIVE") and returns the median level after sorting by numeric values. This naturally tends toward "NORMAL" when levels are mixed, which is the desired conservative behavior for period/window aggregations. Args: levels: List of price level strings from intervals Returns: The median price level string, or PRICE_LEVEL_NORMAL if input is empty Note: For even-length lists, uses upper-middle value (len // 2) to bias toward NORMAL rather than cheaper levels. This provides conservative recommendations when periods contain mixed price levels. Example: [-2, -1, 0, 1] → index 2 → value 0 (NORMAL) This is intentional: we prefer saying "NORMAL" over "CHEAP" when ambiguous. """ if not levels: return PRICE_LEVEL_NORMAL # Convert levels to numeric values and sort numeric_values = [PRICE_LEVEL_MAPPING.get(level, 0) for level in levels] numeric_values.sort() # Get median: middle value for odd length, upper-middle for even length # Upper-middle (len // 2) intentionally biases toward NORMAL (0) for even counts median_idx = len(numeric_values) // 2 median_value = numeric_values[median_idx] # Convert back to level string for level, value in PRICE_LEVEL_MAPPING.items(): if value == median_value: return level return PRICE_LEVEL_NORMAL def aggregate_price_rating(differences: list[float], threshold_low: float, threshold_high: float) -> tuple[str, float]: """ Aggregate multiple price differences into a single rating level. Calculates the average difference percentage across multiple intervals and applies thresholds to determine the overall rating level. Args: differences: List of difference percentages from intervals threshold_low: The low threshold percentage for LOW rating threshold_high: The high threshold percentage for HIGH rating Returns: Tuple of (rating_level, average_difference) rating_level: "LOW", "NORMAL", or "HIGH" average_difference: The averaged difference percentage """ if not differences: return PRICE_RATING_NORMAL, 0.0 # Filter out None values valid_differences = [d for d in differences if d is not None] if not valid_differences: return PRICE_RATING_NORMAL, 0.0 # Calculate average difference avg_difference = sum(valid_differences) / len(valid_differences) # Apply thresholds rating_level = calculate_rating_level(avg_difference, threshold_low, threshold_high) return rating_level or PRICE_RATING_NORMAL, avg_difference def aggregate_period_levels(interval_data_list: list[dict[str, Any]]) -> str | None: """ Aggregate price levels across multiple intervals in a period. Extracts "level" from each interval and uses the same logic as aggregate_price_levels() to determine the overall level for the period. Args: interval_data_list: List of price interval dictionaries with "level" keys Returns: The aggregated level string in lowercase (e.g., "very_cheap", "normal", "expensive"), or None if no valid levels found """ levels: list[str] = [] for interval in interval_data_list: level = interval.get("level") if level is not None and isinstance(level, str): levels.append(level) if not levels: return None aggregated = aggregate_price_levels(levels) # Convert to lowercase for consistency with other enum sensors return aggregated.lower() if aggregated else None def aggregate_period_ratings( interval_data_list: list[dict[str, Any]], threshold_low: float, threshold_high: float, ) -> tuple[str | None, float | None]: """ Aggregate price ratings across multiple intervals in a period. Extracts "difference" from each interval and uses the same logic as aggregate_price_rating() to determine the overall rating for the period. Args: interval_data_list: List of price interval dictionaries with "difference" keys threshold_low: The low threshold percentage for LOW rating threshold_high: The high threshold percentage for HIGH rating Returns: Tuple of (rating_level, average_difference) rating_level: "low", "normal", "high" (lowercase), or None if no valid data average_difference: The averaged difference percentage, or None if no valid data """ differences: list[float] = [] for interval in interval_data_list: diff = interval.get("difference") if diff is not None: differences.append(float(diff)) if not differences: return None, None rating_level, avg_diff = aggregate_price_rating(differences, threshold_low, threshold_high) # Convert to lowercase for consistency with other enum sensors return rating_level.lower() if rating_level else None, avg_diff def _calculate_lookahead_volatility_factor( all_intervals: list[dict[str, Any]], lookahead_intervals: int, volatility_threshold_moderate: float, volatility_threshold_high: float, ) -> float: """ Calculate volatility factor for adaptive thresholds based on lookahead period. Uses the same volatility calculation (coefficient of variation) as volatility sensors, ensuring consistent volatility interpretation across the integration. Args: all_intervals: List of price intervals (today + tomorrow) lookahead_intervals: Number of intervals to analyze for volatility volatility_threshold_moderate: Threshold for moderate volatility (%, e.g., 15) volatility_threshold_high: Threshold for high volatility (%, e.g., 30) Returns: Multiplier for base threshold: - 0.6 for low volatility (< moderate threshold) - 1.0 for moderate volatility (moderate to high threshold) - 1.4 for high volatility (>= high threshold) """ if len(all_intervals) < lookahead_intervals: # Insufficient data - use default factor (no adjustment) return 1.0 # Extract prices from next N intervals lookahead_prices = [ float(interval["total"]) for interval in all_intervals[:lookahead_intervals] if "total" in interval and interval["total"] is not None ] if not lookahead_prices: # No valid prices - use default factor return 1.0 # Use the same volatility calculation as volatility sensors (coefficient of variation) # This ensures consistent interpretation of volatility across the integration volatility_level = calculate_volatility_level( prices=lookahead_prices, threshold_moderate=volatility_threshold_moderate, threshold_high=volatility_threshold_high, # Note: We don't use VERY_HIGH threshold here, only LOW/MODERATE/HIGH matter for factor ) # Map volatility level to adjustment factor if volatility_level == VOLATILITY_LOW: factor = VOLATILITY_FACTOR_SENSITIVE # 0.6 → More sensitive trend detection elif volatility_level in (VOLATILITY_MODERATE, VOLATILITY_HIGH): # Treat MODERATE and HIGH the same for trend detection # HIGH volatility means noisy data, so we need less sensitive thresholds factor = VOLATILITY_FACTOR_NORMAL if volatility_level == VOLATILITY_MODERATE else VOLATILITY_FACTOR_INSENSITIVE else: # VOLATILITY_VERY_HIGH (should not occur with our thresholds, but handle it) factor = VOLATILITY_FACTOR_INSENSITIVE # 1.4 → Less sensitive (filter noise) return factor def calculate_price_trend( # noqa: PLR0913 - All parameters are necessary for volatility-adaptive calculation current_interval_price: float, future_average: float, threshold_rising: float = 3.0, threshold_falling: float = -3.0, *, volatility_adjustment: bool = True, lookahead_intervals: int | None = None, all_intervals: list[dict[str, Any]] | None = None, volatility_threshold_moderate: float = DEFAULT_VOLATILITY_THRESHOLD_MODERATE, volatility_threshold_high: float = DEFAULT_VOLATILITY_THRESHOLD_HIGH, ) -> tuple[str, float]: """ Calculate price trend by comparing current price with future average. Supports volatility-adaptive thresholds: when enabled, the effective threshold is adjusted based on price volatility in the lookahead period. This makes the trend detection more sensitive during stable periods and less noisy during volatile periods. Uses the same volatility thresholds as configured for volatility sensors, ensuring consistent volatility interpretation across the integration. Args: current_interval_price: Current interval price future_average: Average price of future intervals threshold_rising: Base threshold for rising trend (%, positive, default 3%) threshold_falling: Base threshold for falling trend (%, negative, default -3%) volatility_adjustment: Enable volatility-adaptive thresholds (default True) lookahead_intervals: Number of intervals in trend period for volatility calc all_intervals: Price intervals (today + tomorrow) for volatility calculation volatility_threshold_moderate: User-configured moderate volatility threshold (%) volatility_threshold_high: User-configured high volatility threshold (%) Returns: Tuple of (trend_state, difference_percentage) trend_state: "rising" | "falling" | "stable" difference_percentage: % change from current to future ((future - current) / current * 100) Note: Volatility adjustment factor: - Low volatility (<15%): factor 0.6 → more sensitive (e.g., 3% → 1.8%) - Moderate volatility (15-35%): factor 1.0 → as configured (3%) - High volatility (>35%): factor 1.4 → less sensitive (e.g., 3% → 4.2%) """ if current_interval_price == 0: # Avoid division by zero - return stable trend return "stable", 0.0 # Apply volatility adjustment if enabled and data available effective_rising = threshold_rising effective_falling = threshold_falling volatility_factor = 1.0 if volatility_adjustment and lookahead_intervals and all_intervals: volatility_factor = _calculate_lookahead_volatility_factor( all_intervals, lookahead_intervals, volatility_threshold_moderate, volatility_threshold_high ) effective_rising = threshold_rising * volatility_factor effective_falling = threshold_falling * volatility_factor # Calculate percentage difference from current to future # CRITICAL: Use abs() for negative prices to get correct percentage direction # Example: current=-10, future=-5 → diff=5, pct=5/abs(-10)*100=+50% (correctly shows rising) if current_interval_price == 0: # Edge case: avoid division by zero diff_pct = 0.0 else: diff_pct = ((future_average - current_interval_price) / abs(current_interval_price)) * 100 # Determine trend based on effective thresholds if diff_pct >= effective_rising: trend = "rising" elif diff_pct <= effective_falling: trend = "falling" else: trend = "stable" return trend, diff_pct