From 15ba84b39cc96295d4466296735310e5870888c0 Mon Sep 17 00:00:00 2001 From: Andreas Date: Thu, 9 Jul 2026 10:38:21 +0200 Subject: [PATCH] fix: add Tibber electricity price extrapolation --- src/akkudoktoreos/prediction/elecprice.py | 11 +- .../prediction/elecpricetibber.py | 223 ++++++++++++---- src/akkudoktoreos/server/eos.py | 27 +- tests/test_elecpricetibber.py | 248 ++++++++---------- 4 files changed, 312 insertions(+), 197 deletions(-) diff --git a/src/akkudoktoreos/prediction/elecprice.py b/src/akkudoktoreos/prediction/elecprice.py index aeaccdd..f350b05 100644 --- a/src/akkudoktoreos/prediction/elecprice.py +++ b/src/akkudoktoreos/prediction/elecprice.py @@ -18,9 +18,14 @@ def elecprice_provider_ids() -> list[str]: try: prediction_eos = get_prediction() except: - # Prediction may not be initialized - # Return at least provider used in example - return ["ElecPriceAkkudoktor"] + # Prediction may not be initialized. Return static built-in provider ids. + return [ + "ElecPriceAkkudoktor", + "ElecPriceEnergyCharts", + "ElecPriceFixed", + "ElecPriceImport", + "ElecPriceTibber", + ] return [ provider.provider_id() diff --git a/src/akkudoktoreos/prediction/elecpricetibber.py b/src/akkudoktoreos/prediction/elecpricetibber.py index fb1bbf7..b2b810b 100644 --- a/src/akkudoktoreos/prediction/elecpricetibber.py +++ b/src/akkudoktoreos/prediction/elecpricetibber.py @@ -1,18 +1,19 @@ -"""Electricity price provider for Tibber.""" +"""Retrieves and processes electricity price forecast data from Tibber.""" -from datetime import datetime -from typing import Any, List, Optional +from typing import Any, List, Optional, Union +import numpy as np import pandas as pd import requests from loguru import logger from pydantic import Field, ValidationError +from statsmodels.tsa.holtwinters import ExponentialSmoothing from akkudoktoreos.config.configabc import SettingsBaseModel from akkudoktoreos.core.cache import cache_in_file from akkudoktoreos.core.pydantic import PydanticBaseModel from akkudoktoreos.prediction.elecpriceabc import ElecPriceProvider -from akkudoktoreos.utils.datetimeutil import to_datetime +from akkudoktoreos.utils.datetimeutil import to_datetime, to_duration TIBBER_GRAPHQL_URL = "https://api.tibber.com/v1-beta/gql" TIBBER_PRICE_QUERY = """ @@ -25,14 +26,16 @@ query TibberPriceInfo { today { startsAt total - energy - tax } tomorrow { startsAt total - energy - tax + } + } + priceInfoRange(resolution: HOURLY, last: 840) { + nodes { + startsAt + total } } } @@ -56,7 +59,9 @@ class ElecPriceTibberCommonSettings(SettingsBaseModel): home_id: Optional[str] = Field( default=None, json_schema_extra={ - "description": "Tibber home id to read prices from.", + "description": ( + "Optional Tibber home id. If omitted, the first home with a subscription is used." + ), "examples": ["00000000-0000-0000-0000-000000000000"], }, ) @@ -65,10 +70,14 @@ class ElecPriceTibberCommonSettings(SettingsBaseModel): class TibberPricePoint(PydanticBaseModel): """Single Tibber price point.""" - startsAt: datetime + startsAt: str total: float - energy: Optional[float] = None - tax: Optional[float] = None + + +class TibberPriceConnection(PydanticBaseModel): + """Tibber connection for historical price nodes.""" + + nodes: List[TibberPricePoint] = Field(default_factory=list) class TibberPriceInfo(PydanticBaseModel): @@ -81,7 +90,8 @@ class TibberPriceInfo(PydanticBaseModel): class TibberSubscription(PydanticBaseModel): """Tibber subscription data.""" - priceInfo: TibberPriceInfo + priceInfo: Optional[TibberPriceInfo] = None + priceInfoRange: Optional[TibberPriceConnection] = None class TibberHome(PydanticBaseModel): @@ -103,23 +113,54 @@ class TibberData(PydanticBaseModel): viewer: TibberViewer +class TibberGraphQLError(PydanticBaseModel): + """Tibber GraphQL error item.""" + + message: str + + class TibberGraphQLResponse(PydanticBaseModel): """Tibber GraphQL response payload.""" data: Optional[TibberData] = None - errors: Optional[list[dict[str, Any]]] = None + errors: Optional[List[TibberGraphQLError]] = None class ElecPriceTibber(ElecPriceProvider): - """Fetch and store Tibber electricity import prices.""" + """Fetch and process electricity price forecast data from Tibber.""" @classmethod def provider_id(cls) -> str: """Return the unique identifier for the Tibber provider.""" return "ElecPriceTibber" - @cache_in_file(with_ttl="5 minutes") - def _request_forecast(self, force_update: Optional[bool] = False) -> TibberGraphQLResponse: + def historic_hours_min(self) -> int: + """Keep enough history for weekly seasonal price extrapolation.""" + return 24 * 35 + + @classmethod + def _validate_data(cls, json_str: Union[bytes, Any]) -> TibberGraphQLResponse: + """Validate Tibber GraphQL response data.""" + try: + tibber_data = TibberGraphQLResponse.model_validate_json(json_str) + except ValidationError as e: + error_msg = "" + for error in e.errors(): + field = " -> ".join(str(x) for x in error["loc"]) + message = error["msg"] + error_type = error["type"] + error_msg += f"Field: {field}\nError: {message}\nType: {error_type}\n" + logger.error(f"Tibber schema change: {error_msg}") + raise ValueError(error_msg) + if tibber_data.errors: + error_msg = "; ".join(error.message for error in tibber_data.errors) + error_msg = f"Tibber GraphQL error: {error_msg}" + logger.error(error_msg) + raise ValueError(error_msg) + return tibber_data + + @cache_in_file(with_ttl="1 hour") + def _request_forecast(self) -> TibberGraphQLResponse: """Fetch electricity price data from the Tibber GraphQL API.""" access_token = self.config.elecprice.tibber.access_token if not access_token: @@ -134,61 +175,139 @@ class ElecPriceTibber(ElecPriceProvider): }, timeout=30, ) + logger.debug(f"Response from Tibber GraphQL API: {response}") response.raise_for_status() - - try: - return TibberGraphQLResponse.model_validate_json(response.content) - except ValidationError as exc: - logger.error("Tibber schema validation failed: {}", exc) - raise ValueError(f"Tibber schema validation failed: {exc}") from exc + tibber_data = self._validate_data(response.content) + self.update_datetime = to_datetime(in_timezone=self.config.general.timezone) + return tibber_data def _select_home(self, response: TibberGraphQLResponse) -> TibberHome: """Select the configured Tibber home from a GraphQL response.""" - home_id = self.config.elecprice.tibber.home_id - if not home_id: - raise ValueError("Tibber home_id is required") - - if response.errors: - raise ValueError(f"Tibber GraphQL error: {response.errors}") - if response.data is None: raise ValueError("Tibber response does not contain data") - for home in response.data.viewer.homes: - if home.id == home_id: - return home + home_id = self.config.elecprice.tibber.home_id + if home_id: + for home in response.data.viewer.homes: + if home.id == home_id: + if home.currentSubscription is None: + raise ValueError(f"Tibber home '{home_id}' has no current subscription") + return home + raise ValueError("Tibber home_id not found") - raise ValueError("Tibber home_id not found") + for home in response.data.viewer.homes: + if home.currentSubscription is not None: + return home + raise ValueError("No Tibber home with a current subscription found") def _parse_data(self, response: TibberGraphQLResponse) -> pd.Series: """Parse Tibber prices into EOS market prices in EUR/Wh.""" home = self._select_home(response) - - if home.currentSubscription is None: + subscription = home.currentSubscription + if subscription is None: raise ValueError("Tibber home has no current subscription") - price_info = home.currentSubscription.priceInfo - points = list(price_info.today) + list(price_info.tomorrow) - - if not price_info.tomorrow: - logger.warning("Tibber tomorrow prices not available yet") + points: list[TibberPricePoint] = [] + if subscription.priceInfoRange is not None: + points.extend(subscription.priceInfoRange.nodes) + if subscription.priceInfo is not None: + points.extend(subscription.priceInfo.today) + points.extend(subscription.priceInfo.tomorrow) + if not subscription.priceInfo.tomorrow: + logger.warning("Tibber tomorrow prices not available yet") if not points: raise ValueError("Tibber response contains no price points") - values: dict[datetime, float] = {} - + series_data = pd.Series(dtype=float) for point in points: - dt = to_datetime(point.startsAt, in_timezone=self.config.general.timezone) - values[dt] = point.total / 1000.0 + orig_datetime = to_datetime(point.startsAt, in_timezone=self.config.general.timezone) + series_data.at[orig_datetime] = point.total / 1000.0 - return pd.Series(values, dtype=float).sort_index() + return series_data.sort_index() + + def _hourly_series(self, series: pd.Series) -> pd.Series: + """Normalize Tibber prices to hourly values for EOS optimization.""" + if series.empty: + return series + series = series.sort_index() + series.index = pd.to_datetime([to_datetime(index).isoformat() for index in series.index]) + return series.resample("1h").mean().dropna() + + def _cap_outliers(self, data: np.ndarray, sigma: int = 2) -> np.ndarray: + mean = data.mean() + std = data.std() + lower_bound = mean - sigma * std + upper_bound = mean + sigma * std + capped_data = data.clip(min=lower_bound, max=upper_bound) + return capped_data + + def _predict_ets(self, history: np.ndarray, seasonal_periods: int, hours: int) -> np.ndarray: + clean_history = self._cap_outliers(history) + model = ExponentialSmoothing( + clean_history, seasonal="add", seasonal_periods=seasonal_periods + ).fit() + return model.forecast(hours) + + def _predict_median(self, history: np.ndarray, hours: int) -> np.ndarray: + clean_history = self._cap_outliers(history) + return np.full(hours, np.median(clean_history)) def _update_data(self, force_update: Optional[bool] = False) -> None: - """Update EOS electricity prices from Tibber price data.""" - response = self._request_forecast(force_update=force_update) - series_data = self._parse_data(response) - self.key_from_series("elecprice_marketprice_wh", series_data) - self.update_datetime = to_datetime(in_timezone=self.config.general.timezone) + """Update Tibber price data and extrapolate missing future prices.""" + tibber_data = self._request_forecast(force_update=force_update) # type: ignore + if not self.ems_start_datetime: + raise ValueError(f"Start DateTime not set: {self.ems_start_datetime}") - logger.info("Updated ElecPriceTibber with {} price points", len(series_data)) + series_data = self._hourly_series(self._parse_data(tibber_data)) + if series_data.empty: + raise ValueError("Tibber response contains no usable hourly price points") + + highest_orig_datetime = to_datetime(series_data.index.max()) + self.key_from_series("elecprice_marketprice_wh", series_data) + + history = self.key_to_array( + key="elecprice_marketprice_wh", + end_datetime=highest_orig_datetime, + fill_method="linear", + ) + + amount_datasets = len(self.records) + if not highest_orig_datetime: + error_msg = f"Highest original datetime not available: {highest_orig_datetime}" + logger.error(error_msg) + raise ValueError(error_msg) + + needed_hours = int( + self.config.prediction.hours + - ((highest_orig_datetime - self.ems_start_datetime).total_seconds() // 3600) + ) + + if needed_hours <= 0: + logger.warning( + "No prediction needed. " + f"needed_hours={needed_hours}, " + f"hours={self.config.prediction.hours}, " + f"highest_orig_datetime={highest_orig_datetime}, " + f"start_datetime={self.ems_start_datetime}" + ) + return + + if amount_datasets > 800: + prediction = self._predict_ets(history, seasonal_periods=168, hours=needed_hours) + elif amount_datasets > 168: + prediction = self._predict_ets(history, seasonal_periods=24, hours=needed_hours) + elif amount_datasets > 0: + prediction = self._predict_median(history, hours=needed_hours) + else: + logger.error("No data available for prediction") + raise ValueError("No data available") + + prediction_series = pd.Series( + data=prediction, + index=[ + highest_orig_datetime + to_duration(f"{i + 1} hours") + for i in range(len(prediction)) + ], + ) + self.key_from_series("elecprice_marketprice_wh", prediction_series) diff --git a/src/akkudoktoreos/server/eos.py b/src/akkudoktoreos/server/eos.py index 1d7066f..ab8343d 100755 --- a/src/akkudoktoreos/server/eos.py +++ b/src/akkudoktoreos/server/eos.py @@ -10,6 +10,7 @@ import traceback from contextlib import asynccontextmanager from typing import Annotated, Any, AsyncGenerator, Dict, List, Optional, Union +import pandas as pd import psutil import uvicorn from fastapi import Body, FastAPI @@ -1130,16 +1131,24 @@ async def fastapi_strompreis() -> list[float]: start_datetime = to_datetime().start_of("day") end_datetime = start_datetime.add(days=2) try: - elecprice = ( - get_prediction() - .key_to_array( - key="elecprice_marketprice_wh", - start_datetime=start_datetime, - end_datetime=end_datetime, - fill_method="ffill" - ) - .tolist() + elecprice_series = get_prediction().key_to_series( + key="elecprice_marketprice_wh", + start_datetime=start_datetime, + end_datetime=end_datetime, ) + elecprice_series.index = pd.to_datetime(elecprice_series.index) + elecprice_series = pd.to_numeric(elecprice_series.sort_index(), errors="coerce") + start_timestamp = pd.Timestamp(start_datetime.isoformat()) + end_timestamp = pd.Timestamp(end_datetime.subtract(seconds=1).isoformat()) + hourly = elecprice_series.resample("1h", origin=start_timestamp).mean() + hourly = hourly.truncate(before=start_timestamp, after=end_timestamp) + hourly_index = pd.date_range( + start=start_timestamp, + end=pd.Timestamp(end_datetime.subtract(hours=1).isoformat()), + freq="1h", + ) + hourly = hourly.reindex(hourly_index) + elecprice = hourly.ffill().bfill().tolist() except Exception as e: raise HTTPException( status_code=404, diff --git a/tests/test_elecpricetibber.py b/tests/test_elecpricetibber.py index b62349e..556d985 100644 --- a/tests/test_elecpricetibber.py +++ b/tests/test_elecpricetibber.py @@ -3,6 +3,8 @@ import json from unittest.mock import Mock, patch +import numpy as np +import pandas as pd import pytest from akkudoktoreos.core.cache import CacheFileStore @@ -15,6 +17,43 @@ from akkudoktoreos.prediction.elecpricetibber import ( from akkudoktoreos.utils.datetimeutil import to_datetime +class _FakeEms: + start_datetime = to_datetime("2026-07-09T00:00:00+00:00") + + +def _price(starts_at: str, total: float) -> dict[str, object]: + return {"startsAt": starts_at, "total": total} + + +def _tibber_payload( + prices: list[dict[str, object]], + *, + home_id: str = "home-1", + include_other_home: bool = False, +) -> dict[str, object]: + homes: list[dict[str, object]] = [] + if include_other_home: + homes.append( + { + "id": "other-home", + "currentSubscription": { + "priceInfo": {"today": [_price("2026-07-09T00:00:00+00:00", 0.999)]} + }, + } + ) + + homes.append( + { + "id": home_id, + "currentSubscription": { + "priceInfo": {"today": prices[:2], "tomorrow": prices[2:]}, + "priceInfoRange": {"nodes": prices}, + }, + } + ) + return {"data": {"viewer": {"homes": homes}}} + + @pytest.fixture def provider(config_eos): """Create a fresh Tibber electricity price provider.""" @@ -23,7 +62,17 @@ def provider(config_eos): provider="ElecPriceTibber", tibber=ElecPriceTibberCommonSettings(access_token="token-123", home_id="home-1"), ) - return ElecPriceTibber() + config_eos.prediction.hours = 6 + provider = ElecPriceTibber() + provider.records.clear() + return provider + + +@pytest.fixture +def tibber_provider(provider, monkeypatch): + """Create a Tibber provider with a deterministic EMS start time.""" + monkeypatch.setattr("akkudoktoreos.core.coreabc.get_ems", lambda: _FakeEms()) + return provider @pytest.fixture @@ -35,59 +84,14 @@ def cache_store(): @pytest.fixture def tibber_response_dict(): """Sample Tibber GraphQL response.""" - return { - "data": { - "viewer": { - "homes": [ - { - "id": "other-home", - "currentSubscription": { - "priceInfo": { - "today": [ - { - "startsAt": "2026-07-07T00:00:00.000+02:00", - "total": 0.999, - "energy": 0.111, - "tax": 0.888, - } - ], - "tomorrow": [], - } - }, - }, - { - "id": "home-1", - "currentSubscription": { - "priceInfo": { - "today": [ - { - "startsAt": "2026-07-07T01:00:00.000+02:00", - "total": 0.2970716, - "energy": 0.10922, - "tax": 0.1878516, - }, - { - "startsAt": "2026-07-07T00:00:00.000+02:00", - "total": 0.3109662, - "energy": 0.12098, - "tax": 0.1899862, - }, - ], - "tomorrow": [ - { - "startsAt": "2026-07-08T00:00:00.000+02:00", - "total": 0.30468, - "energy": 0.1162, - "tax": 0.18848, - } - ], - } - }, - }, - ] - } - } - } + return _tibber_payload( + [ + _price("2026-07-07T01:00:00.000+02:00", 0.2970716), + _price("2026-07-07T00:00:00.000+02:00", 0.3109662), + _price("2026-07-08T00:00:00.000+02:00", 0.30468), + ], + include_other_home=True, + ) @pytest.fixture @@ -135,22 +139,21 @@ def test_missing_access_token_raises(provider, config_eos): provider._request_forecast(force_update=True) -def test_missing_home_id_raises(provider, config_eos, tibber_response): - """A Tibber home id is required for selecting prices.""" +def test_select_home_uses_first_subscription_when_home_id_is_omitted( + provider, config_eos, tibber_response +): + """If no home id is configured, the first subscribed Tibber home is used.""" config_eos.elecprice.tibber.home_id = None - with pytest.raises(ValueError, match="Tibber home_id is required"): - provider._select_home(tibber_response) + home = provider._select_home(tibber_response) + + assert home.id == "other-home" def test_graphql_errors_raise(provider): """GraphQL errors are surfaced as ValueError.""" - response = TibberGraphQLResponse.model_validate( - {"errors": [{"message": "Authentication failed"}]} - ) - with pytest.raises(ValueError, match="Tibber GraphQL error"): - provider._select_home(response) + provider._validate_data(json.dumps({"errors": [{"message": "Authentication failed"}]})) def test_unknown_home_id_raises(provider, config_eos, tibber_response): @@ -162,7 +165,7 @@ def test_unknown_home_id_raises(provider, config_eos, tibber_response): def test_parse_data_combines_sorts_and_converts_total(provider, tibber_response): - """Today and tomorrow prices are sorted and converted from EUR/kWh to EUR/Wh.""" + """Today, tomorrow, and history prices are sorted and converted to EUR/Wh.""" series = provider._parse_data(tibber_response) assert list(series.index) == [ @@ -175,86 +178,26 @@ def test_parse_data_combines_sorts_and_converts_total(provider, tibber_response) assert series.iloc[2] == pytest.approx(0.00030468) -def test_update_data_stores_elecprice_marketprice_wh(provider, tibber_response): - """Parsed Tibber totals are stored in EOS records.""" - with patch.object(provider, "_request_forecast", return_value=tibber_response): - provider.update_data(force_enable=True, force_update=True) +def test_tibber_hourly_series_averages_quarter_hour_prices(provider): + """Quarter-hour Tibber prices are averaged to hourly EOS prices.""" + index = pd.date_range("2026-07-09T00:00:00+00:00", periods=8, freq="15min") + series = pd.Series([0.10, 0.30, 0.50, 0.70, 1.0, 1.4, 1.8, 2.2], index=index) - series = provider.key_to_series("elecprice_marketprice_wh") + hourly = provider._hourly_series(series) - assert len(series) == 3 - assert series.iloc[0] == pytest.approx(0.0003109662) - assert series.iloc[1] == pytest.approx(0.0002970716) - assert series.iloc[2] == pytest.approx(0.00030468) - - -def test_total_conversion_exact_example(provider): - """Tibber total 0.311 EUR/kWh is stored as 0.000311 EUR/Wh.""" - response = TibberGraphQLResponse.model_validate( - { - "data": { - "viewer": { - "homes": [ - { - "id": "home-1", - "currentSubscription": { - "priceInfo": { - "today": [ - { - "startsAt": "2026-07-07T00:00:00.000+02:00", - "total": 0.311, - } - ], - "tomorrow": [], - } - }, - } - ] - } - } - } - ) - - series = provider._parse_data(response) - - assert series.iloc[0] == pytest.approx(0.000311) + assert hourly.tolist() == pytest.approx([0.40, 1.60]) def test_empty_tomorrow_stores_only_today_and_warns(provider): - """An empty tomorrow list does not create fake values.""" + """An empty tomorrow list does not create fake values before forecasting.""" response = TibberGraphQLResponse.model_validate( - { - "data": { - "viewer": { - "homes": [ - { - "id": "home-1", - "currentSubscription": { - "priceInfo": { - "today": [ - { - "startsAt": "2026-07-07T00:00:00.000+02:00", - "total": 0.3109662, - }, - { - "startsAt": "2026-07-07T01:00:00.000+02:00", - "total": 0.2970716, - }, - ], - "tomorrow": [], - } - }, - } - ] - } - } - } + _tibber_payload([_price("2026-07-07T00:00:00.000+02:00", 0.3109662)]) ) with patch("akkudoktoreos.prediction.elecpricetibber.logger.warning") as mock_warning: series = provider._parse_data(response) - assert len(series) == 2 + assert len(series) == 1 mock_warning.assert_called_once_with("Tibber tomorrow prices not available yet") @@ -282,5 +225,44 @@ def test_request_forecast_uses_tibber_graphql_api( assert kwargs["headers"]["Content-Type"] == "application/json" assert "query" in kwargs["json"] assert "TibberPriceInfo" in kwargs["json"]["query"] + assert "priceInfoRange" in kwargs["json"]["query"] assert "total" in kwargs["json"]["query"] assert kwargs["timeout"] == 30 + + +def test_tibber_update_extrapolates_missing_hours_with_seasonal_history( + tibber_provider, monkeypatch +): + """Missing Tibber future hours are forecast from seasonal price history.""" + data = TibberGraphQLResponse.model_validate( + _tibber_payload( + [ + _price("2026-07-09T00:00:00+00:00", 0.30), + _price("2026-07-09T01:00:00+00:00", 0.42), + _price("2026-07-09T02:00:00+00:00", 0.36), + ] + ) + ) + monkeypatch.setattr(tibber_provider, "_request_forecast", lambda **_: data) + monkeypatch.setattr( + tibber_provider, + "_predict_ets", + lambda history, seasonal_periods, hours: np.full(hours, 0.0005), + ) + + history = pd.Series( + data=np.linspace(0.0002, 0.0004, 169), + index=pd.date_range("2026-07-01T23:00:00+00:00", periods=169, freq="1h"), + ) + tibber_provider.key_from_series("elecprice_marketprice_wh", history) + + tibber_provider._update_data(force_update=True) + + prices = tibber_provider.key_to_array( + key="elecprice_marketprice_wh", + start_datetime=to_datetime("2026-07-09T00:00:00+00:00"), + end_datetime=to_datetime("2026-07-09T06:00:00+00:00"), + fill_method="ffill", + ) + + assert prices.tolist() == pytest.approx([0.0003, 0.00042, 0.00036, 0.0005, 0.0005, 0.0005])