1382 lines
59 KiB
Python
1382 lines
59 KiB
Python
"""TARIFF-02 - price provider plugins, and the one canonical shape they all emit.
|
|
|
|
A new Belgian supplier is a new class and one line in `PROVIDERS`; the
|
|
optimiser never learns a supplier's name. Lives in `goodwe_controller/app/`,
|
|
not `ems/ingest/prices/` (owner, 2026-08-30).
|
|
|
|
Canonical format (specs.md 9.3), three fields and no more:
|
|
|
|
{ "prices": [ { "start": "2026-01-15T00:00:00+01:00",
|
|
"eur_per_kwh": 0.1834,
|
|
"provisional": false } ] }
|
|
|
|
`start` is tz-aware at the Brussels offset for that instant, the price is EUR
|
|
per kWh and ALL-IN, `provisional` is per point.
|
|
|
|
Three things shape the module (trap 1 markup is added once by the framework,
|
|
2 the product is declared, 3 "no data" is a type), and there is no `zoneinfo`
|
|
(the image has no tzdata). The reasoning and the later additions:
|
|
DECISIONS/TARIFF-02-price-provider-plugins.md."""
|
|
|
|
import ast
|
|
import csv
|
|
import inspect
|
|
import io
|
|
import json
|
|
import logging
|
|
import math
|
|
from abc import ABC, abstractmethod
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timedelta, timezone
|
|
from enum import Enum
|
|
from textwrap import dedent
|
|
from urllib.parse import urlencode
|
|
|
|
from . import messages
|
|
from .custom_formula import (
|
|
EXPORT_FORMULA_OPTION, FormulaInvalid, PREVIEW_HOURS, parse_export_formula,
|
|
parse_price_formula, preview_lines, price_eur_kwh, to_resolution)
|
|
|
|
_LOG = logging.getLogger("goodwe.price")
|
|
|
|
|
|
# Brussels wall time, without a tz database (DECISIONS/TARIFF-02-price-
|
|
# provider-plugins.md).
|
|
|
|
CET = timezone(timedelta(hours=1))
|
|
CEST = timezone(timedelta(hours=2))
|
|
|
|
|
|
def _last_sunday(year: int, month: int) -> int:
|
|
"""Day-of-month of the last Sunday. March and October both have 31 days."""
|
|
day = 31
|
|
while datetime(year, month, day).weekday() != 6: # 6 == Sunday
|
|
day -= 1
|
|
return day
|
|
|
|
|
|
def brussels_offset(instant: datetime) -> timezone:
|
|
"""The offset Belgium is on at this UTC instant. CET or CEST, never else."""
|
|
u = instant.astimezone(timezone.utc)
|
|
y = u.year
|
|
start = datetime(y, 3, _last_sunday(y, 3), 1, tzinfo=timezone.utc)
|
|
end = datetime(y, 10, _last_sunday(y, 10), 1, tzinfo=timezone.utc)
|
|
return CEST if start <= u < end else CET
|
|
|
|
|
|
def brussels(instant: datetime) -> datetime:
|
|
"""Render a UTC instant at the Brussels offset for that instant.
|
|
Takes an INSTANT and refuses a naive datetime (wrong for one hour twice a year)."""
|
|
if instant.tzinfo is None or instant.utcoffset() is None:
|
|
raise ValueError(
|
|
f"brussels({instant!r}): a naive datetime is not an instant. Parse "
|
|
f"the source's offset, or say UTC explicitly - guessing here is how "
|
|
f"one hour of a transition day gets priced twice.")
|
|
return instant.astimezone(brussels_offset(instant))
|
|
|
|
|
|
def brussels_date_range(since: datetime, until: datetime) -> tuple:
|
|
"""The Brussels calendar days a half-open window [since, until) touches.
|
|
|
|
Returns `(first, last)`, both inclusive: the shape `api.energy-charts.info`
|
|
takes. TARIFF-09: a one-day request left everything past today provisional.
|
|
The calendar is Brussels, never UTC; `until` is exclusive (local midnight
|
|
belongs to the day before); an empty or inverted window is one day.
|
|
DECISIONS/TARIFF-02-price-provider-plugins.md."""
|
|
first = brussels(since).date()
|
|
edge = brussels(until)
|
|
last = edge.date()
|
|
if (edge.hour, edge.minute, edge.second, edge.microsecond) == (0, 0, 0, 0):
|
|
last = last - timedelta(days=1)
|
|
if last < first:
|
|
last = first
|
|
return first, last
|
|
|
|
|
|
def dated_feed_url(base: str, **params) -> str:
|
|
"""`base` plus a query string. Refuses a `base` that already carries one
|
|
(appending would build a malformed or duplicated query). Not reached by an
|
|
operator's `price_feed_url`, which is used exactly as given."""
|
|
if "?" in base or "#" in base:
|
|
raise SourceUnavailable(
|
|
f"the feed base {base!r} already carries a query string or "
|
|
f"fragment, and this builder appends parameters. Adding a second "
|
|
f"'?' produces a URL the server parses differently from the one "
|
|
f"anybody intended; put the fixed parameters in the call instead")
|
|
return f"{base}?{urlencode(params)}" if params else base
|
|
|
|
|
|
# Absence, borrowed from sim/riglock.py (TEST-04).
|
|
|
|
class Absence(Exception):
|
|
"""A value was asked of a read that never observed one."""
|
|
|
|
|
|
class FrameworkOption(Exception):
|
|
"""A plugin reached for an option the FRAMEWORK owns. See trap 1."""
|
|
|
|
|
|
class SourceUnavailable(Exception):
|
|
"""A plugin could not get its data. Raised by plugins, caught by the framework,
|
|
turned into `PriceRead.no_data(...)`. It must never escape into the control
|
|
tick: the loop keeps regulating on P1 whatever the price situation is."""
|
|
|
|
|
|
class _Outcome:
|
|
"""One outcome of one read. The absent variant CARRIES NO FIELDS."""
|
|
|
|
kind = "?"
|
|
words = "?" # human phrasing. Cosmetic, never an input to a decision.
|
|
|
|
def __getattr__(self, name):
|
|
if name.startswith("__") and name.endswith("__"):
|
|
raise AttributeError(name)
|
|
raise Absence(
|
|
f"{type(self).__name__} has no {name!r}. This read did not observe "
|
|
f"a price series, and a value invented here is the bug this "
|
|
f"representation exists to prevent - 'no data' becoming 'free'. "
|
|
f"Handle the {self.kind!r} case explicitly; see match().")
|
|
|
|
def get(self, *_a, **_k):
|
|
raise Absence(
|
|
f"{type(self).__name__}.get() does not exist: a default is how a "
|
|
f"failed fetch becomes an observation. Use .match(...) and handle "
|
|
f"both outcomes.")
|
|
|
|
def __iter__(self):
|
|
# DELIBERATE: `for p in read:` and `list(read)` would also turn an
|
|
# absent read into an empty loop.
|
|
raise Absence(
|
|
f"{type(self).__name__} is not iterable: an empty loop body is how "
|
|
f"'we have no prices' becomes 'we checked every hour and none were "
|
|
f"expensive'. Use .match(series=..., no_data=...).")
|
|
|
|
def __repr__(self):
|
|
return f"{type(self).__name__}({self.words!r})"
|
|
|
|
|
|
# The vocabulary
|
|
|
|
class Component(Enum):
|
|
"""The three parts of an all-in Belgian consumption price (§9.3)."""
|
|
ENERGY = "energy"
|
|
MARKUP = "markup"
|
|
TAXES = "taxes"
|
|
|
|
|
|
ALL_IN = frozenset(Component)
|
|
|
|
|
|
class PriceProduct(Enum):
|
|
"""The §9.2 product table. The optimiser switches on THIS, never on a name."""
|
|
DAY_AHEAD = "dynamic_dayahead"
|
|
REALTIME = "realtime_imbalance"
|
|
STATIC = "static_profile"
|
|
FIXED = "fixed"
|
|
|
|
|
|
# specs.md 8.3: real-time / imbalance products are high-volatility and widen
|
|
# the
|
|
# arbitrage thresholds. The ONLY place that says which products are volatile.
|
|
# OPT-04 reads it; nothing checks plugin ids.
|
|
_WIDE = frozenset({PriceProduct.REALTIME})
|
|
|
|
|
|
def volatility_regime(product: PriceProduct) -> str:
|
|
""""wide" | "normal" - the arbitrage threshold regime for a product (8.3).
|
|
OPT-04 owns what the widened numbers are; this owns which products get them."""
|
|
if not isinstance(product, PriceProduct):
|
|
raise TypeError(f"volatility_regime() takes a PriceProduct, not "
|
|
f"{product!r} - a name is exactly what AC2 forbids.")
|
|
return "wide" if product in _WIDE else "normal"
|
|
|
|
|
|
# FCST-03: ONE widening path, driven by a SET of active uncertainty sources
|
|
# (real-time product 8.3, no weather data 11.4). The combination rule is `max`,
|
|
# not a product. The factors are policy defaults, not derived, and they differ
|
|
# so the combination rules stay distinguishable (DECISIONS/TARIFF-02-price-
|
|
# provider-plugins.md).
|
|
class Uncertainty(Enum):
|
|
"""Why an arbitrage threshold is currently wider than the tariff alone says."""
|
|
REALTIME_PRODUCT = "realtime_product" # §9.2 row 2, §8.3
|
|
NO_WEATHER = "no_weather" # §11.4 "no weather data"
|
|
|
|
|
|
WIDEN_FACTOR = {
|
|
Uncertainty.REALTIME_PRODUCT: 2.5,
|
|
Uncertainty.NO_WEATHER: 1.5,
|
|
}
|
|
|
|
|
|
def uncertainty_of(product: PriceProduct) -> frozenset:
|
|
"""The uncertainty a PRODUCT contributes on its own (9.2 row 2), derived from the
|
|
declared product, never from a plugin id."""
|
|
if not isinstance(product, PriceProduct):
|
|
raise TypeError(f"uncertainty_of() takes a PriceProduct, not "
|
|
f"{product!r} - a plugin name is what §9.3 forbids.")
|
|
return (frozenset({Uncertainty.REALTIME_PRODUCT})
|
|
if product in _WIDE else frozenset())
|
|
|
|
|
|
def widen_factor(sources) -> float:
|
|
"""The multiplier the widest active source asks for. 1.0 when there are none.
|
|
A string is refused: it would be a silent 1.0."""
|
|
srcs = frozenset(sources)
|
|
bad = [s for s in srcs if not isinstance(s, Uncertainty)]
|
|
if bad:
|
|
raise TypeError(f"widen_factor() takes Uncertainty members, not {bad!r}. "
|
|
f"A string here would widen by 1.0 and say it widened.")
|
|
return max((WIDEN_FACTOR[s] for s in srcs), default=1.0)
|
|
|
|
|
|
def required_spread(base_eur_kwh: float, sources) -> float:
|
|
"""The spread an arbitrage cycle must clear, given what is uncertain (8.3).
|
|
`base_eur_kwh` is OPT-04's own round-trip cost threshold."""
|
|
if isinstance(base_eur_kwh, bool) or not isinstance(base_eur_kwh, (int, float)) or not math.isfinite(base_eur_kwh) or base_eur_kwh < 0:
|
|
raise ValueError(
|
|
f"required_spread() takes a non-negative, finite base threshold in "
|
|
f"EUR/kWh, not {base_eur_kwh!r}. A negative or NaN base makes every "
|
|
f"spread clear it, which is arbitrage with the brake removed.")
|
|
return round(float(base_eur_kwh) * widen_factor(sources), 6)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class PricePoint:
|
|
"""One canonical point. THREE FIELDS, as 9.3 lists them."""
|
|
|
|
start: datetime # tz-aware, rendered at the Brussels offset
|
|
eur_per_kwh: float # ALL-IN: energy + markup + taxes
|
|
provisional: bool
|
|
|
|
def canonical(self) -> dict:
|
|
return {"start": self.start.isoformat(),
|
|
"eur_per_kwh": self.eur_per_kwh,
|
|
"provisional": self.provisional}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class RawPoint:
|
|
"""What a plugin returns: the figure its source published, unmodified. Not a
|
|
PricePoint: only `assemble()` builds those, so a plugin cannot claim all-in."""
|
|
|
|
start: datetime # tz-aware instant
|
|
eur_per_kwh: float # as published, containing exactly caps.sources
|
|
provisional: bool
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ProviderCapabilities:
|
|
"""What a plugin declares about itself, before anyone asks it for prices."""
|
|
|
|
product: PriceProduct
|
|
sources: frozenset # which Components the published figure contains
|
|
horizon_h: int = 48 # aligns with the §8.1/§8.2 forecast horizon
|
|
egress_hosts: tuple = () # §16 allow-list, declared not discovered
|
|
|
|
def __post_init__(self):
|
|
if not isinstance(self.product, PriceProduct):
|
|
raise ValueError(f"product must be a PriceProduct, not "
|
|
f"{self.product!r}")
|
|
if Component.ENERGY not in self.sources:
|
|
raise ValueError(
|
|
"capabilities().sources must contain Component.ENERGY - a feed "
|
|
"that does not publish an energy price is not a price feed.")
|
|
extra = set(self.sources) - set(ALL_IN)
|
|
if extra:
|
|
raise ValueError(f"unknown price components declared: {extra}")
|
|
|
|
@property
|
|
def adds(self) -> frozenset:
|
|
"""The components the FRAMEWORK must add. The complement of `sources`."""
|
|
return frozenset(ALL_IN - self.sources)
|
|
|
|
|
|
# The read: a series, or no data. Nothing in between.
|
|
|
|
class PriceRead(_Outcome):
|
|
"""What one attempt to get prices said: `series` (prices, NON-EMPTY by
|
|
construction) or `no_data` (a `.reason` and nothing else; 11.4: static profile,
|
|
plan provisional, no arbitrage). A series of all-zero prices is still a
|
|
`series`: zero prices make arbitrage pointless, no data forbids it."""
|
|
|
|
def match(self, *, series, no_data):
|
|
"""Fold to a value. Both handlers are required keywords, so the no-data case cannot be forgotten."""
|
|
return {"series": series, "no_data": no_data}[self.kind](self)
|
|
|
|
@staticmethod
|
|
def series(points, product, provider, fetched_at):
|
|
return _Series(points, product, provider, fetched_at)
|
|
|
|
@staticmethod
|
|
def no_data(reason):
|
|
if not reason or not isinstance(reason, str):
|
|
raise ValueError(
|
|
"PriceRead.no_data() takes a reason a human can act on. It is "
|
|
"the only field this variant has; an empty one makes an "
|
|
"operator's only evidence a blank status line.")
|
|
return _NoData(reason)
|
|
|
|
|
|
# How long ONE point covers - TARIFF-14
|
|
|
|
# A point does not carry its own duration (9.3 has three fields): the span is
|
|
# derived from the neighbours once, at construction.
|
|
MAX_POINT_SPAN = timedelta(hours=1)
|
|
|
|
|
|
def _cover_table(points) -> tuple:
|
|
"""`(start_utc, end_utc, point)` per point, ascending. The `at()` index.
|
|
|
|
TARIFF-14: a point covers from its own start until the next point's start, and
|
|
never more than one hour. Each point takes its own gap (firm quarter-hours and
|
|
carried hours are mixed in steady state). The cap keeps FCST-04's deliberate
|
|
holes from being filled by stretching the point before them. The last point
|
|
gets the span of the one before it (one hour if alone), so on a 96-point day
|
|
`at(00:00)` raises `Absence` rather than answering with today's last price.
|
|
Sorted here, so no caller's list order matters. Nothing here raises on a
|
|
malformed series: a duplicated start (refused by `load_prices()`) gives the
|
|
earlier entry a zero-width window. DECISIONS/TARIFF-14-the-window-a-point-covers.md."""
|
|
rows = sorted(((p.start.astimezone(timezone.utc), i, p)
|
|
for i, p in enumerate(points)),
|
|
key=lambda r: (r[0], r[1]))
|
|
out = []
|
|
for n, (start, _i, point) in enumerate(rows):
|
|
if n + 1 < len(rows):
|
|
span = rows[n + 1][0] - start # -> the next point's start
|
|
elif n:
|
|
span = start - rows[n - 1][0] # -> the tail, see above
|
|
else:
|
|
span = MAX_POINT_SPAN # -> a series of one point
|
|
if span > MAX_POINT_SPAN:
|
|
span = MAX_POINT_SPAN
|
|
out.append((start, start + span, point))
|
|
return tuple(out)
|
|
|
|
|
|
class _Series(PriceRead):
|
|
kind = "series"
|
|
|
|
def __init__(self, points, product, provider, fetched_at):
|
|
points = list(points)
|
|
# An empty series is not constructible: "the fetch failed" must not
|
|
# look like "nothing is dear today".
|
|
if not points:
|
|
raise ValueError(
|
|
"PriceRead.series() refuses an empty point list. A fetch that "
|
|
"produced no points is PriceRead.no_data(reason) - a different "
|
|
"variant with different fields, on purpose. See §11.4.")
|
|
if not isinstance(product, PriceProduct):
|
|
raise ValueError(
|
|
f"product must be a PriceProduct, not {product!r}. The "
|
|
f"optimiser switches on this enum instead of on a plugin name; "
|
|
f"a string here reopens exactly the hardcoding AC2 forbids.")
|
|
self.points = tuple(points)
|
|
self.product = product
|
|
self.provider = provider
|
|
self.fetched_at = fetched_at
|
|
self.regime = volatility_regime(product)
|
|
# TARIFF-14. Built once here so `at()` scans an index whose order it
|
|
# established itself.
|
|
self._cover = _cover_table(self.points)
|
|
self.words = (f"{len(self.points)} points from {provider} "
|
|
f"({product.value}, {self.regime} thresholds)")
|
|
|
|
def canonical(self) -> dict:
|
|
"""The §9.3 wire format. Exactly `{"prices": [...]}`."""
|
|
return {"prices": [p.canonical() for p in self.points]}
|
|
|
|
def at(self, instant: datetime) -> PricePoint:
|
|
"""The point whose OWN interval covers `instant`, at ANY resolution. Returns a
|
|
point or raises; there is no falsy answer, so `at(t) or fallback` cannot happen.
|
|
Ask `covers()` first to branch. Linear scan (about 192 entries, 30 s path).
|
|
The window rule is in `_cover_table()`."""
|
|
u = instant.astimezone(timezone.utc)
|
|
for start, end, point in self._cover:
|
|
if start <= u < end:
|
|
return point
|
|
raise Absence(
|
|
f"no price point covers {u.isoformat()}; the series runs "
|
|
f"{self.points[0].start.isoformat()} .. "
|
|
f"{self.points[-1].start.isoformat()}. An instant outside the "
|
|
f"series - or inside a hole FCST-04 left because nothing "
|
|
f"published that hour - is an instant with no price, which is "
|
|
f"§11.4 for it, not a reason to reach for a neighbouring one.")
|
|
|
|
def covers(self, instant: datetime) -> bool:
|
|
"""Whether `at(instant)` would answer. Delegates to `at()`: a second copy of the
|
|
window rule is a second place for it to be wrong."""
|
|
try:
|
|
self.at(instant)
|
|
return True
|
|
except Absence:
|
|
return False
|
|
|
|
def last_end_before(self, instant: datetime):
|
|
"""When the last price that has run out by `instant` stopped covering, or None
|
|
(PRICE-01: "no prices since 00:00"). A hole is reported from where it starts."""
|
|
u = instant.astimezone(timezone.utc)
|
|
ends = [end for _start, end, _p in self._cover if end <= u]
|
|
return max(ends) if ends else None
|
|
|
|
def window(self, since: datetime, until: datetime) -> tuple:
|
|
"""Points whose START falls in [since, until). May be empty. Not a coverage
|
|
query (it never asks how long a point lasts); `at()`/`covers()` are."""
|
|
a = since.astimezone(timezone.utc)
|
|
b = until.astimezone(timezone.utc)
|
|
return tuple(p for p in self.points
|
|
if a <= p.start.astimezone(timezone.utc) < b)
|
|
|
|
|
|
class _NoData(PriceRead):
|
|
"""We have no prices, and the 11.4 row applies. There is no `.points`, so
|
|
`min(read.points)` and `.get("points", [])` raise `Absence` rather than report
|
|
the cheapest hour as free. `bool(read)` is True for both variants and has no
|
|
falsy form on purpose: use `.match(...)`."""
|
|
|
|
kind = "no_data"
|
|
|
|
def __init__(self, reason):
|
|
self.reason = reason
|
|
self.words = f"no price data - {reason}"
|
|
|
|
|
|
def arbitrage_allowed(read: PriceRead) -> bool:
|
|
"""11.4 as a function, so no caller has to remember it. Only the veto; the strategy is OPT-04."""
|
|
return read.match(series=lambda _r: True, no_data=lambda _r: False)
|
|
|
|
|
|
def plan_is_provisional(read: PriceRead) -> bool:
|
|
"""8.3 / 11.4: provisional if there is no data at all, or if ANY point is itself
|
|
provisional. Derived from the points, never asserted."""
|
|
return read.match(
|
|
series=lambda r: any(p.provisional for p in r.points),
|
|
no_data=lambda _r: True)
|
|
|
|
|
|
# Options: what a plugin may see, and what it may not
|
|
|
|
# TARIFF-18: what `price_taxes_eur_kwh` IS MADE OF
|
|
|
|
# Each row is one line of the Eneco residential card for Flanders, September
|
|
# 2026, including 6 % VAT, in EUR/kWh. The default of `price_taxes_eur_kwh` is
|
|
# their sum (test_price.py section 1b re-adds them). Per DSO, and set near the
|
|
# low
|
|
# end of the band on purpose; `energiebijdrage` is a zero row since 2026-08-01.
|
|
# Why: DECISIONS/TARIFF-02-price-provider-plugins.md, "Taxes components".
|
|
FLEMISH_TAXES_COMPONENTS_EUR_KWH = (
|
|
("Fluvius distribution, Antwerpen", 0.0535),
|
|
("Bijzondere accijns, 'Alle verbruik' - no volume bands", 0.04876),
|
|
("Bijdrage groene stroom en WKK Vlaanderen", 0.0151),
|
|
("Energiebijdrage - 0 EUR/MWh since 2026-08-01, inside the accijns", 0.0),
|
|
)
|
|
|
|
# TARIFF-18: moving this default changes which configs start. validate.py
|
|
# (TARIFF-17 rung) compares `price_taxes_eur_kwh` to the default, so a
|
|
# superseded
|
|
# default (0.06) must still read as untouched. Do not change a default without
|
|
# that check. DECISIONS/TARIFF-02-price-provider-plugins.md.
|
|
SUPERSEDED_TAXES_DEFAULTS_EUR_KWH = (0.06,) # TARIFF-02 .. TARIFF-17
|
|
|
|
|
|
# CFG-03. The one in-code mirror of config.yaml for the price options:
|
|
# validate.py
|
|
# imports it, main.py has no price `.get()` fallbacks, and test_price.py
|
|
# section 1
|
|
# pins it to config.yaml key for key.
|
|
PRICE_OPTION_DEFAULTS = {
|
|
"price_provider": "off",
|
|
"price_markup_eur_kwh": 0.03, # §14 tariffs.energy_price.markup_eur_kwh
|
|
# TARIFF-18: the sum of FLEMISH_TAXES_COMPONENTS_EUR_KWH, to the card's
|
|
# last digit.
|
|
"price_taxes_eur_kwh": 0.11736,
|
|
"export_price_eur_kwh": 0.04, # §14 tariffs.export_price_eur_kwh (AC4)
|
|
"price_csv_path": "/share/goodwe_prices.csv",
|
|
"price_feed_url": "",
|
|
# PRICE-01. The bearer token (a secret: in SECRET_OPTIONS, so `scrub()`
|
|
# keeps it out of text).
|
|
"price_service_token": "",
|
|
"price_horizon_h": 48,
|
|
"price_timeout_s": 10,
|
|
# TARIFF-03. A JSON static day/night or fixed tariff; "" is unset. Parsed
|
|
# by
|
|
# app/static_tariff.py, refused at startup by validate.py. Here, not in
|
|
# static_tariff.py, so there is one defaults mirror (CFG-03).
|
|
"static_profile": "",
|
|
# TARIFF-17. A period table in `static_profile`'s grammar whose
|
|
# `eur_per_kwh`
|
|
# is the TAXES component per band; "" is unset and the scalar decides. In
|
|
# FRAMEWORK_KEYS, so a plugin cannot add taxes a second time per hour.
|
|
"price_taxes_profile": "",
|
|
# TARIFF-15. One contract's two coefficients, units, VAT rate and the card
|
|
# figures they are graded against; "" is unset. Parsed by
|
|
# app/custom_formula.py.
|
|
"price_formula": "",
|
|
# TARIFF-16. The INJECTION leg in `price_formula`'s own grammar, read by
|
|
# `custom_formula.parse_export_formula()`; "" is unset and
|
|
# `export_price_eur_kwh`
|
|
# decides. In FRAMEWORK_KEYS, so a consumption plugin cannot fold it in.
|
|
"price_export_formula": "",
|
|
# OPT-04, 10.1 priority 3, and `optimizer.cycle_cost_threshold_eur`. The
|
|
# two
|
|
# SoC keys are not the maintenance keys. 0.10 is the 10.1 name, not a value
|
|
# that
|
|
# makes anything trade (static spread is about 0.08): an owner decision.
|
|
# DECISIONS/TARIFF-02-price-provider-plugins.md.
|
|
"arbitrage_enabled": False,
|
|
"cycle_cost_threshold_eur": 0.10,
|
|
"arbitrage_soc_floor": 20,
|
|
"arbitrage_soc_ceiling": 95,
|
|
}
|
|
|
|
# The keys the FRAMEWORK owns. A plugin that reads one can double-count, so
|
|
# `PluginOptions` refuses them by name (trap 1).
|
|
FRAMEWORK_KEYS = frozenset({
|
|
"price_markup_eur_kwh", "price_taxes_eur_kwh", "export_price_eur_kwh",
|
|
# TARIFF-17. The hour-varying `price_taxes_eur_kwh`, added once by
|
|
# `caps.adds`. `code_leaks()` derives from this set.
|
|
"price_taxes_profile",
|
|
# TARIFF-16. The hour-varying `export_price_eur_kwh`; same reason, same
|
|
# derivation.
|
|
EXPORT_FORMULA_OPTION,
|
|
})
|
|
|
|
|
|
# PRICE-01. Options whose VALUE must never be printed; `scrub()` removes them
|
|
# from error text.
|
|
SECRET_OPTIONS = frozenset({"price_service_token"})
|
|
|
|
|
|
def scrub(text, opts) -> str:
|
|
"""`text` with every secret option's value replaced by `***`. The last line of
|
|
defence: a library may put a request header into its error text."""
|
|
out = str(text)
|
|
for key in SECRET_OPTIONS:
|
|
secret = str((opts or {}).get(key) or "").strip()
|
|
if len(secret) >= 4:
|
|
out = out.replace(secret, "***")
|
|
return out
|
|
|
|
|
|
class PluginOptions:
|
|
"""The options dict, minus the keys only the framework may read: reading
|
|
`price_markup_eur_kwh` raises `FrameworkOption` rather than returning 0.03. One of
|
|
two doors; `register()`'s `code_leaks()` scan is the other. Do not weaken either."""
|
|
|
|
def __init__(self, opts: dict):
|
|
self._o = dict(opts or {})
|
|
|
|
def get(self, key, default=None):
|
|
if key in FRAMEWORK_KEYS:
|
|
raise FrameworkOption(
|
|
f"{key!r} belongs to the framework, not to a plugin. A plugin "
|
|
f"returns RawPoints carrying what its SOURCE published and "
|
|
f"declares capabilities().sources; assemble() adds the missing "
|
|
f"components exactly once. If a plugin could read this number "
|
|
f"it could add it a second time, and the resulting price would "
|
|
f"be wrong by one markup with nothing anywhere reporting it.")
|
|
v = self._o.get(key, None)
|
|
if v is None or v == "":
|
|
return PRICE_OPTION_DEFAULTS.get(key, default)
|
|
return v
|
|
|
|
def __getitem__(self, key):
|
|
return self.get(key)
|
|
|
|
def __contains__(self, key):
|
|
return key not in FRAMEWORK_KEYS and key in self._o
|
|
|
|
|
|
def price_option(opts: dict, key: str):
|
|
"""A framework-side option read. One place, defaults from one dict."""
|
|
if key not in PRICE_OPTION_DEFAULTS:
|
|
raise KeyError(f"{key!r} is not a price option; add it to "
|
|
f"PRICE_OPTION_DEFAULTS and to config.yaml together.")
|
|
v = (opts or {}).get(key, None)
|
|
if v is None or v == "":
|
|
return PRICE_OPTION_DEFAULTS[key]
|
|
return v
|
|
|
|
|
|
def provider_id(opts: dict) -> str:
|
|
"""The selected plugin id, normalised. `"off"` when none is selected.
|
|
One spelling for three call sites (TARIFF-10). validate.py keeps its own reader
|
|
(it validates the raw blob). It does not decide whether the id is shipped."""
|
|
return str(price_option(opts, "price_provider") or "off").strip()
|
|
|
|
|
|
def provider_configured(opts: dict) -> bool:
|
|
"""Has the operator selected a price feed at all? (TARIFF-10) Starts the
|
|
background refresh, which used to wait for `arbitrage_enabled` (ships false).
|
|
NOT a trading switch: `arbitrage.enabled()` answers that (`main.tariff_refresh_wanted()`)."""
|
|
return provider_id(opts) not in ("", "off")
|
|
|
|
|
|
def export_price_eur_kwh(opts: dict) -> float:
|
|
"""AC4 / 9.3: the injection price is a SEPARATE SCALAR, not a second series.
|
|
|
|
Not reachable from a plugin (`export_price_eur_kwh` is in FRAMEWORK_KEYS). The
|
|
formula leg is `export_formula()`; this stays the flat default. Nothing under
|
|
`app/` takes a decision on it. DECISIONS/TARIFF-02-price-provider-plugins.md."""
|
|
return float(price_option(opts, "export_price_eur_kwh"))
|
|
|
|
|
|
def export_formula(opts: dict):
|
|
"""The configured `price_export_formula` as a `FormulaSpec`, or None (TARIFF-16).
|
|
`custom_formula.FormulaInvalid` is refused at startup by validate.py, so it
|
|
should not reach a running site. Same parser as the consumption leg."""
|
|
return parse_export_formula(price_option(opts, EXPORT_FORMULA_OPTION))
|
|
|
|
|
|
def export_price_at(opts: dict, index_eur_per_mwh: float) -> float:
|
|
"""What the site is PAID for one exported kWh at one index value, EUR/kWh.
|
|
No caller under `app/` yet. A flat price ignores the index (correct); a
|
|
negative answer is real and is never floored."""
|
|
spec = export_formula(opts)
|
|
if spec is None:
|
|
return export_price_eur_kwh(opts)
|
|
return price_eur_kwh(spec, index_eur_per_mwh)
|
|
|
|
|
|
def taxes_profile(opts: dict):
|
|
"""The configured `price_taxes_profile` as a period table, or None (TARIFF-17).
|
|
`static_tariff.ProfileInvalid` is refused at startup by validate.py;
|
|
`load_prices()` still handles it loudly. The import is lazy: static_tariff.py
|
|
imports this module at module scope."""
|
|
from .static_tariff import parse_taxes_profile
|
|
return parse_taxes_profile(price_option(opts, "price_taxes_profile"))
|
|
|
|
|
|
# The plugin interface
|
|
|
|
MAX_PLAUSIBLE_EUR_KWH = 10.0 # see _validate_raw
|
|
|
|
|
|
class PriceProvider(ABC):
|
|
"""One supplier. Declares what it is, then returns what its source said.
|
|
`fetch()` returns `RawPoint`s, not prices; `assemble()` does the arithmetic."""
|
|
|
|
id = "?"
|
|
|
|
# PRICE-01. Ids to try in order when this source cannot answer, and whether
|
|
# the
|
|
# last answer is kept for a day. Empty / False for every earlier plugin.
|
|
FALLBACKS: tuple = ()
|
|
KEEPS_LAST_GOOD: bool = False
|
|
|
|
@classmethod
|
|
@abstractmethod
|
|
def capabilities(cls) -> ProviderCapabilities:
|
|
...
|
|
|
|
@abstractmethod
|
|
async def fetch(self, opts, since: datetime, until: datetime,
|
|
transport) -> list:
|
|
"""Return `RawPoint`s, or raise `SourceUnavailable(reason)`.
|
|
|
|
`transport(url, timeout_s) -> bytes` is injected. `opts` is a `PluginOptions`.
|
|
`since`/`until` are the range the caller wants; a plugin that ignores them gets
|
|
its feed's default (TARIFF-09). Not mechanically enforced: test_price.py pins
|
|
which plugins consume the range as a literal roster."""
|
|
|
|
def startup_preview(self, points) -> list:
|
|
"""Lines to log ONCE, the first time this provider answers (TARIFF-15). `points`
|
|
are the assembled all-in `PricePoint`s; the default says nothing. A hook, since
|
|
only the plugin knows what went in. Must not raise into the control path
|
|
(`_emit_startup_preview()` catches everything)."""
|
|
return []
|
|
|
|
|
|
# Names a plugin's own CODE may not mention: FRAMEWORK_KEYS plus `_o`,
|
|
# `price_option` and `PRICE_OPTION_DEFAULTS`, the other routes PR #16 review
|
|
# used
|
|
# to double-count.
|
|
BANNED_IN_PLUGIN_CODE = tuple(sorted(FRAMEWORK_KEYS)) + (
|
|
"PRICE_OPTION_DEFAULTS", "price_option", "_o")
|
|
|
|
|
|
def code_leaks(cls) -> list:
|
|
"""Framework-owned names appearing in a plugin's own CODE. Empty is clean.
|
|
Docstrings are stripped first (they must name the components). Fails open on
|
|
missing source; test_price.py asserts source is found for every shipped plugin."""
|
|
try:
|
|
src = dedent(inspect.getsource(cls))
|
|
tree = ast.parse(src)
|
|
except (OSError, TypeError, IndentationError, SyntaxError) as e:
|
|
_LOG.warning("Could not read the code of price source %s (%s), so its "
|
|
"double-counting check is skipped.",
|
|
getattr(cls, "id", cls), e)
|
|
return []
|
|
for node in ast.walk(tree):
|
|
body = getattr(node, "body", None)
|
|
if isinstance(body, list) and body and isinstance(body[0], ast.Expr) and isinstance(body[0].value, ast.Constant) and isinstance(body[0].value.value, str):
|
|
body.pop(0)
|
|
code = ast.unparse(tree)
|
|
return sorted({name for name in BANNED_IN_PLUGIN_CODE if name in code})
|
|
|
|
|
|
# TARIFF-07: the ids that USED to resolve here. A retired id is not a typo and
|
|
# has no silent alias (it published retail prices, `generic_dayahead` reads
|
|
# wholesale). Each value is the id-specific half of the refusal; validate.py
|
|
# owns
|
|
# the framing. DECISIONS/TARIFF-07-withdrawing-three-plugins.md.
|
|
RETIRED_PROVIDERS: dict = {
|
|
"engie_dynamic": (
|
|
"its endpoint https://api.engie.be/dynamic/be does not exist - the "
|
|
"gateway there answers 404 with Kong's \"no Route matched with those "
|
|
"values\". No public endpoint could replace it: ENGIE's dynamic price "
|
|
"is (B x eSpot_15) - A with PER-CONTRACT coefficients that live behind "
|
|
"your klantenzone login, so no unauthenticated URL can return YOUR "
|
|
"price"),
|
|
"totalenergies_dynamic": (
|
|
"its endpoint https://api.totalenergies.be/prices/dynamic does not "
|
|
"exist - the TLS handshake fails outright because that hostname is "
|
|
"absent from the certificate the server presents. The coefficients are "
|
|
"per-contract there too"),
|
|
"eneco_eddy": (
|
|
"its endpoint https://api.eneco.be/eddy/prices does not exist - it "
|
|
"answers 404 with an empty body - and \"Eddy\" is not an Eneco "
|
|
"product at all; their dynamic product is \"Zon & Wind Dynamisch\", "
|
|
"which is indexed to Belpex/EPEX DAY-AHEAD and whose coefficients are "
|
|
"likewise per-contract"),
|
|
}
|
|
|
|
|
|
PROVIDERS: dict = {}
|
|
|
|
|
|
def register(cls):
|
|
"""Register a provider by its own declared id. Decorator. The only name->class
|
|
mapping in the system (AC2: adding a plugin needs no optimiser changes)."""
|
|
if cls.id in PROVIDERS:
|
|
raise ValueError(f"duplicate provider id {cls.id!r}")
|
|
if cls.id in RETIRED_PROVIDERS:
|
|
# At import: re-registering a retired id silences validate.py's
|
|
# refusal. Take
|
|
# the id out of RETIRED_PROVIDERS in the same commit.
|
|
raise ValueError(
|
|
f"{cls.id!r} is in RETIRED_PROVIDERS and cannot also be a shipped "
|
|
f"plugin. A site configured on it is refused at startup with a "
|
|
f"message naming the id; registering it silently takes that "
|
|
f"message away. If this id really ships again, remove it from "
|
|
f"RETIRED_PROVIDERS in the same commit - and say in the CHANGELOG "
|
|
f"what changed about the endpoint.")
|
|
cls.capabilities() # fail at import, not at 15:00 on fetch day
|
|
leaks = code_leaks(cls)
|
|
if leaks:
|
|
# At import, not at fetch: a plugin that can reach the markup must not
|
|
# exist.
|
|
raise FrameworkOption(
|
|
f"{cls.id!r} mentions {', '.join(leaks)} in its own code. A plugin "
|
|
f"returns what its SOURCE published and declares "
|
|
f"capabilities().sources; the framework adds the complement, once. "
|
|
f"Reaching any of {BANNED_IN_PLUGIN_CODE} lets it be added twice, "
|
|
f"and a price that is one markup high is invisible in every log "
|
|
f"this add-on writes. Remove the reference; if the plugin genuinely "
|
|
f"needs a per-site number, add a NON-framework option key for it.")
|
|
PROVIDERS[cls.id] = cls
|
|
return cls
|
|
|
|
|
|
def _validate_raw(points, caps):
|
|
"""Reject at the boundary, so a malformed feed cannot become a bad decision.
|
|
Negative prices are valid and not clamped (9.3, 17). Rejected: NaN, infinity,
|
|
an implausible magnitude, a naive datetime, duplicate or out-of-order starts."""
|
|
if not isinstance(caps, ProviderCapabilities):
|
|
raise SourceUnavailable("the plugin did not declare capabilities")
|
|
if not points:
|
|
raise SourceUnavailable("the source returned no points")
|
|
seen = None
|
|
for p in points:
|
|
if not isinstance(p, RawPoint):
|
|
raise SourceUnavailable(f"a plugin returned {type(p).__name__}, not "
|
|
f"a RawPoint - see PriceProvider.fetch()")
|
|
if p.start.tzinfo is None or p.start.utcoffset() is None:
|
|
raise SourceUnavailable(f"point at {p.start!r} has no offset")
|
|
if isinstance(p.eur_per_kwh, bool) or \
|
|
not isinstance(p.eur_per_kwh, (int, float)) or \
|
|
not math.isfinite(p.eur_per_kwh):
|
|
raise SourceUnavailable(f"point at {p.start.isoformat()} has a "
|
|
f"non-finite price {p.eur_per_kwh!r}")
|
|
if abs(p.eur_per_kwh) > MAX_PLAUSIBLE_EUR_KWH:
|
|
raise SourceUnavailable(
|
|
f"point at {p.start.isoformat()} is {p.eur_per_kwh} EUR/kWh, "
|
|
f"beyond +-{MAX_PLAUSIBLE_EUR_KWH}, so the feed is probably in "
|
|
f"EUR/MWh")
|
|
u = p.start.astimezone(timezone.utc)
|
|
if seen is not None and u <= seen:
|
|
raise SourceUnavailable(
|
|
f"points are not strictly increasing at {u.isoformat()} - a "
|
|
f"duplicated hour on a DST day is the usual cause, and keying "
|
|
f"prices by a naive local hour is how it happens")
|
|
seen = u
|
|
|
|
|
|
def taxes_at(taxes, instant: datetime) -> float:
|
|
"""The TAXES component at ONE instant. A scalar, or a period table (TARIFF-17).
|
|
Duck-typed on `price_at` (price.py must not import static_tariff.py). A linear
|
|
scan over two to four periods on the 30 s path; validate.py refuses a malformed
|
|
table at startup. DECISIONS/TARIFF-02-price-provider-plugins.md."""
|
|
at = getattr(taxes, "price_at", None)
|
|
if at is None:
|
|
return float(taxes)
|
|
return float(at(instant))
|
|
|
|
|
|
def assemble(caps: ProviderCapabilities, raw, markup: float,
|
|
taxes) -> list:
|
|
"""RawPoints -> canonical all-in PricePoints. THE ONLY PLACE MARKUP IS ADDED.
|
|
|
|
Adds exactly `caps.adds`, the complement of what the plugin declared. `taxes`
|
|
is evaluated at EACH point's own instant: hoisting it out of the loop collapses
|
|
the feature (test_price.py section 24). On a scalar the result is byte-identical
|
|
to before TARIFF-17."""
|
|
add_markup = float(markup) if Component.MARKUP in caps.adds else 0.0
|
|
wants_taxes = Component.TAXES in caps.adds
|
|
out = []
|
|
for p in raw:
|
|
local = brussels(p.start)
|
|
add = add_markup + (taxes_at(taxes, local) if wants_taxes else 0.0)
|
|
out.append(PricePoint(start=local,
|
|
eur_per_kwh=round(p.eur_per_kwh + add, 6),
|
|
provisional=bool(p.provisional)))
|
|
return out
|
|
|
|
|
|
# Transport - one place, HTTPS only, host checked against the declaration
|
|
|
|
async def https_get(url: str, timeout_s: float, headers=None) -> bytes:
|
|
"""The default transport. 16: outbound is allow-listed and TLS'd. aiohttp is
|
|
imported lazily so `manual_csv` works offline. PRICE-01: with `headers` it
|
|
answers in the safe words of `status_words()` and never lets a library error
|
|
through (it could print a token)."""
|
|
import aiohttp
|
|
t = aiohttp.ClientTimeout(total=float(timeout_s))
|
|
if headers is None:
|
|
async with aiohttp.ClientSession(timeout=t) as s:
|
|
async with s.get(url) as r:
|
|
r.raise_for_status()
|
|
return await r.read()
|
|
try:
|
|
async with aiohttp.ClientSession(timeout=t) as s:
|
|
# No redirects: the host is fixed in code and a redirect would take
|
|
# the token past the allow-list.
|
|
async with s.get(url, headers=headers,
|
|
allow_redirects=False) as r:
|
|
if r.status != 200:
|
|
raise SourceUnavailable(status_words(r.status))
|
|
return await r.read()
|
|
except SourceUnavailable:
|
|
raise
|
|
except Exception as e: # noqa: BLE001
|
|
raise SourceUnavailable(
|
|
f"the price service did not answer ({type(e).__name__})")
|
|
|
|
|
|
# PRICE-01. The phrase a refused token is reported with; the fallback chain
|
|
# reads it.
|
|
AUTH_REFUSED = "the price service refused the token"
|
|
|
|
|
|
def status_words(status: int) -> str:
|
|
"""What an HTTP status from the price service is called. No header text."""
|
|
if status in (401, 403):
|
|
return f"{AUTH_REFUSED} ({status})"
|
|
if status == 429:
|
|
return "the price service says too many requests (429)"
|
|
if status >= 500:
|
|
return f"the price service has an error ({status})"
|
|
return f"the price service answered {status}"
|
|
|
|
|
|
def check_egress(url: str, caps: ProviderCapabilities) -> str:
|
|
"""16: a plugin may only reach a host it DECLARED. Returns the host. The scheme
|
|
is split off before the host is parsed (a fixed-length strip once made the TLS
|
|
rule untestable; see DECISIONS/TARIFF-02-price-provider-plugins.md)."""
|
|
if not url:
|
|
raise SourceUnavailable("no feed URL configured (price_feed_url)")
|
|
scheme, sep, rest = url.partition("://")
|
|
if not sep:
|
|
raise SourceUnavailable(
|
|
f"not an absolute URL, so no host can be checked against the "
|
|
f"§16 allow-list: {url!r}")
|
|
if scheme.lower() != "https":
|
|
raise SourceUnavailable(
|
|
f"refusing a non-TLS price feed: {url!r} uses {scheme!r}. §16 "
|
|
f"allow-lists outbound destinations and SEC-01 requires TLS on "
|
|
f"every one of them.")
|
|
host = rest.split("/", 1)[0].split("?", 1)[0].split("@")[-1]
|
|
host = host.split(":", 1)[0].lower()
|
|
if host not in {h.lower() for h in caps.egress_hosts}:
|
|
raise SourceUnavailable(
|
|
f"{host!r} is not in this provider's declared egress_hosts "
|
|
f"{caps.egress_hosts} - §16 allow-lists outbound destinations, and "
|
|
f"a URL from the options file is not a declaration")
|
|
return host
|
|
|
|
|
|
# The shipped plugins. Each docstring has the two fixed lines `SOURCES: ...` /
|
|
# `The framework ADDS: ...`; test_price.py parses them and compares with
|
|
# `capabilities().sources`.
|
|
|
|
# TARIFF-08: the COLUMNAR document the public EPEX feed publishes (parallel
|
|
# `unix_seconds` / `price` arrays plus `unit`). The only parser; the row parser
|
|
# it
|
|
# replaced read a shape no feed published. Two shapes, two functions, no
|
|
# sniffing.
|
|
# DECISIONS/TARIFF-02-price-provider-plugins.md.
|
|
|
|
COLUMNAR_TIME_KEY = "unix_seconds"
|
|
COLUMNAR_PRICE_KEY = "price"
|
|
COLUMNAR_UNIT_KEY = "unit"
|
|
|
|
|
|
def _columnar_unit(text) -> str:
|
|
"""`"EUR / MWh"` -> `"eur/mwh"`. Spacing and case are not information."""
|
|
return "".join(str(text).split()).lower()
|
|
|
|
|
|
def _points_from_columnar_json(body, expect_unit, scale,
|
|
provisional_default):
|
|
"""Parse a COLUMNAR price document: parallel `unix_seconds` / `price`.
|
|
|
|
Refuses what could become a number: unequal arrays (a truncated evening peak),
|
|
a unit other than the declared one (a factor of 1000), and anything but UTC
|
|
instants. A `null` price is dropped and logged. The Brussels rendering is
|
|
`assemble()`'s. DECISIONS/TARIFF-02-price-provider-plugins.md."""
|
|
try:
|
|
text = body.decode("utf-8") if isinstance(body, (bytes, bytearray)) \
|
|
else str(body)
|
|
doc = json.loads(text)
|
|
except (ValueError, UnicodeDecodeError) as e:
|
|
raise SourceUnavailable(f"the feed did not parse as JSON: {e}")
|
|
if not isinstance(doc, dict):
|
|
raise SourceUnavailable(
|
|
f"the feed is a {type(doc).__name__}, not a columnar object with "
|
|
f"{COLUMNAR_TIME_KEY!r} and {COLUMNAR_PRICE_KEY!r} arrays")
|
|
times = doc.get(COLUMNAR_TIME_KEY)
|
|
prices = doc.get(COLUMNAR_PRICE_KEY)
|
|
for key, col in ((COLUMNAR_TIME_KEY, times), (COLUMNAR_PRICE_KEY, prices)):
|
|
if not isinstance(col, list):
|
|
raise SourceUnavailable(
|
|
f"the feed carried no {key!r} array; this provider reads the "
|
|
f"columnar shape (parallel {COLUMNAR_TIME_KEY}/"
|
|
f"{COLUMNAR_PRICE_KEY} arrays), and a row-per-point document "
|
|
f"is a different feed, not this one with a field missing")
|
|
if len(times) != len(prices):
|
|
raise SourceUnavailable(
|
|
f"{COLUMNAR_TIME_KEY} has {len(times)} entries and "
|
|
f"{COLUMNAR_PRICE_KEY} has {len(prices)}, so the feed is cut short "
|
|
f"and none of it is used")
|
|
unit = _columnar_unit(doc.get(COLUMNAR_UNIT_KEY, ""))
|
|
if unit != _columnar_unit(expect_unit):
|
|
raise SourceUnavailable(
|
|
f"the feed reports {doc.get(COLUMNAR_UNIT_KEY)!r}, not "
|
|
f"{expect_unit!r}, so its prices cannot be converted")
|
|
if doc.get("deprecated"):
|
|
# Not a refusal, but 11.4 makes a dead feed look quiet, so warn while
|
|
# the source still works.
|
|
_LOG.warning("The price feed says it will stop (deprecated=%r), "
|
|
"though it still answers for now.", doc.get("deprecated"))
|
|
out = []
|
|
dropped = []
|
|
for i, (t, v) in enumerate(zip(times, prices)):
|
|
if v is None:
|
|
dropped.append(i)
|
|
continue
|
|
if isinstance(t, bool) or not isinstance(t, (int, float)) or \
|
|
not math.isfinite(t):
|
|
raise SourceUnavailable(
|
|
f"{COLUMNAR_TIME_KEY}[{i}] is {t!r}, not epoch seconds")
|
|
if isinstance(v, bool) or not isinstance(v, (int, float)) or \
|
|
not math.isfinite(v):
|
|
raise SourceUnavailable(
|
|
f"{COLUMNAR_PRICE_KEY}[{i}] is {v!r}, which is not a price")
|
|
start = datetime.fromtimestamp(float(t), timezone.utc)
|
|
out.append(RawPoint(start=start, eur_per_kwh=float(v) * scale,
|
|
provisional=bool(provisional_default)))
|
|
if dropped:
|
|
_LOG.warning("The price feed gave no price for %d of %s, at index "
|
|
"%s. Those are left out, not counted as free.",
|
|
len(dropped), messages.count(len(times), "time"),
|
|
", ".join(str(i) for i in dropped[:5])
|
|
+ (" and more" if len(dropped) > 5 else ""))
|
|
if not out:
|
|
raise SourceUnavailable(
|
|
f"the feed carried {len(times)} instants and a price for none of "
|
|
f"them")
|
|
return out
|
|
|
|
|
|
def dayahead_feed_url(plugin_id, opts, since, until, base, bzn) -> str:
|
|
"""The day-ahead URL for a window, or the operator's override, verbatim.
|
|
|
|
Shared by `generic_dayahead` and `custom_formula` (TARIFF-15); takes
|
|
`plugin_id` as an argument (AC2). An override is used exactly as given and so
|
|
loses the range; that is logged at fetch time with the horizon consequence.
|
|
DECISIONS/TARIFF-02-price-provider-plugins.md."""
|
|
override = str(opts.get("price_feed_url") or "").strip()
|
|
if override:
|
|
_LOG.warning(
|
|
"price_feed_url is set, so %s asks for %s as given, and "
|
|
"tomorrow's prices may stay provisional. Clear it to ask for %s "
|
|
"to %s.",
|
|
plugin_id, override, *(d.isoformat()
|
|
for d in brussels_date_range(since, until)))
|
|
_LOG.debug(
|
|
"The public feed answers one Brussels day to a request that names "
|
|
"no range. Every later hour is then copied from yesterday and "
|
|
"marked provisional, and §8.3 leaves provisional hours out of "
|
|
"trading.")
|
|
return override
|
|
first, last = brussels_date_range(since, until)
|
|
return dated_feed_url(base, bzn=bzn, start=first.isoformat(),
|
|
end=last.isoformat())
|
|
|
|
|
|
@register
|
|
class GenericDayAhead(PriceProvider):
|
|
"""Public EPEX BE day-ahead feed, COLUMNAR, in EUR/MWh.
|
|
|
|
SOURCES: energy
|
|
The framework ADDS: markup, taxes
|
|
|
|
The published figure is the wholesale clearing price, so both user-entered
|
|
components are added by the framework. Quarter-hourly since 2025-10-01; nothing
|
|
here assumes a granularity."""
|
|
|
|
id = "generic_dayahead"
|
|
|
|
@classmethod
|
|
def capabilities(cls):
|
|
return ProviderCapabilities(
|
|
product=PriceProduct.DAY_AHEAD,
|
|
sources=frozenset({Component.ENERGY}),
|
|
egress_hosts=("api.energy-charts.info", "web-api.tp.entsoe.eu"))
|
|
|
|
FEED_BASE = "https://api.energy-charts.info/price"
|
|
BZN = "BE"
|
|
|
|
async def fetch(self, opts, since, until, transport):
|
|
"""Request the range the CALLER asked for, on the Brussels calendar
|
|
(TARIFF-09; `dayahead_feed_url()` builds the URL)."""
|
|
url = dayahead_feed_url(self.id, opts, since, until,
|
|
self.FEED_BASE, self.BZN)
|
|
check_egress(url, self.capabilities())
|
|
body = await transport(url, float(opts.get("price_timeout_s")))
|
|
return _points_from_columnar_json(body, expect_unit="EUR / MWh",
|
|
scale=0.001,
|
|
provisional_default=False)
|
|
|
|
|
|
@register
|
|
class PriceService(PriceProvider):
|
|
"""The owner's own price server (PRICE-01), COLUMNAR, in EUR/MWh.
|
|
|
|
SOURCES: energy
|
|
The framework ADDS: markup, taxes
|
|
|
|
Answers the energy-charts layout plus `covers`, the complete Brussels days it
|
|
holds. The host is fixed here, HTTPS only; the bearer token (option
|
|
`price_service_token`) goes in the `Authorization` header and into no message.
|
|
`FALLBACKS` names the next source; `app/price_chain.py` walks it."""
|
|
|
|
id = "price_service"
|
|
|
|
HOST = "prices.kammenstraatha.duckdns.org"
|
|
URL = "https://prices.kammenstraatha.duckdns.org/v1/dayahead/BE"
|
|
FALLBACKS = ("generic_dayahead",)
|
|
KEEPS_LAST_GOOD = True
|
|
|
|
@classmethod
|
|
def capabilities(cls):
|
|
return ProviderCapabilities(
|
|
product=PriceProduct.DAY_AHEAD,
|
|
sources=frozenset({Component.ENERGY}),
|
|
egress_hosts=(cls.HOST,))
|
|
|
|
async def fetch(self, opts, since, until, transport):
|
|
token = str(opts.get("price_service_token") or "").strip()
|
|
if not token:
|
|
raise SourceUnavailable("no token is set for the price service")
|
|
if any(c.isspace() or not c.isprintable() for c in token):
|
|
raise SourceUnavailable(
|
|
"the price service token has a space or a line break in it")
|
|
check_egress(self.URL, self.capabilities())
|
|
try:
|
|
body = await transport(
|
|
self.URL, float(opts.get("price_timeout_s")),
|
|
headers={"Authorization": "Bearer " + token})
|
|
except SourceUnavailable:
|
|
raise
|
|
except Exception as e: # noqa: BLE001
|
|
raise SourceUnavailable(
|
|
f"the price service did not answer ({type(e).__name__})")
|
|
points = _points_from_columnar_json(body, expect_unit="EUR / MWh",
|
|
scale=0.001,
|
|
provisional_default=False)
|
|
covers = _covered_days(body)
|
|
if covers is None:
|
|
return points
|
|
whole = [p for p in points
|
|
if brussels(p.start).date().isoformat() in covers]
|
|
if not whole:
|
|
raise SourceUnavailable(
|
|
"the price service sent no complete day")
|
|
return whole
|
|
|
|
|
|
def _covered_days(body):
|
|
"""The `covers` list of a price service answer as a set, or None. A day not listed as complete is not used."""
|
|
try:
|
|
text = body.decode("utf-8") if isinstance(body, (bytes, bytearray)) \
|
|
else str(body)
|
|
covers = json.loads(text).get("covers")
|
|
except (ValueError, UnicodeDecodeError, AttributeError):
|
|
return None
|
|
if covers is None:
|
|
return None
|
|
if not isinstance(covers, list) or \
|
|
not all(isinstance(d, str) for d in covers):
|
|
raise SourceUnavailable("the price service sent a bad list of days")
|
|
return set(covers)
|
|
|
|
|
|
@register
|
|
class ManualCsv(PriceProvider):
|
|
"""A file on disk. AC3: the documented fallback, fully offline.
|
|
|
|
SOURCES: energy, markup, taxes
|
|
The framework ADDS: nothing
|
|
|
|
The CSV is all-in (the plugin cannot know what a typed number contains), and
|
|
DOCS.md says so. No transport, aiohttp, network or tz database.
|
|
|
|
Format - header required, two columns or three:
|
|
|
|
start,eur_per_kwh,provisional
|
|
2026-01-15T00:00:00+01:00,0.1834,false"""
|
|
|
|
id = "manual_csv"
|
|
|
|
@classmethod
|
|
def capabilities(cls):
|
|
return ProviderCapabilities(
|
|
product=PriceProduct.DAY_AHEAD,
|
|
sources=ALL_IN,
|
|
egress_hosts=()) # declared empty: this one reaches nothing
|
|
|
|
async def fetch(self, opts, since, until, transport):
|
|
path = opts.get("price_csv_path")
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as fh:
|
|
text = fh.read()
|
|
except OSError as e:
|
|
raise SourceUnavailable(f"cannot read {path}: {e}")
|
|
return self.parse(text)
|
|
|
|
@staticmethod
|
|
def parse(text: str) -> list:
|
|
"""Split out so the offline check can drive it without a filesystem."""
|
|
rdr = csv.DictReader(io.StringIO(text))
|
|
if not rdr.fieldnames or "start" not in rdr.fieldnames or \
|
|
"eur_per_kwh" not in rdr.fieldnames:
|
|
raise SourceUnavailable(
|
|
f"the CSV needs a header with at least 'start' and "
|
|
f"'eur_per_kwh'; found {rdr.fieldnames!r}")
|
|
out = []
|
|
for n, row in enumerate(rdr, start=2):
|
|
try:
|
|
start = datetime.fromisoformat((row["start"] or "").strip())
|
|
value = float((row["eur_per_kwh"] or "").strip())
|
|
except (TypeError, ValueError) as e:
|
|
raise SourceUnavailable(f"CSV line {n}: {e}")
|
|
if start.tzinfo is None:
|
|
raise SourceUnavailable(
|
|
f"CSV line {n}: {row['start']!r} has no UTC offset. Write "
|
|
f"'2026-01-15T00:00:00+01:00', not '2026-01-15 00:00' - a "
|
|
f"local wall clock is ambiguous for one hour every October.")
|
|
flag = (row.get("provisional") or "false").strip().lower()
|
|
out.append(RawPoint(start=start, eur_per_kwh=value,
|
|
provisional=flag in ("1", "true", "yes")))
|
|
return out
|
|
|
|
|
|
@register
|
|
class CustomFormula(PriceProvider):
|
|
"""The operator's OWN tariff card, as two fixed coefficients. TARIFF-15.
|
|
|
|
SOURCES: energy, markup
|
|
The framework ADDS: taxes
|
|
|
|
It declares `{ENERGY, MARKUP}`, not `ALL_IN`: a card's formula is the energy
|
|
component only. No expression evaluation: the coefficients are numbers because
|
|
the parser refuses anything else (the string is settable over HTTP). The index
|
|
path is `generic_dayahead`'s. DECISIONS/TARIFF-02-price-provider-plugins.md."""
|
|
|
|
id = "custom_formula"
|
|
|
|
FEED_BASE = "https://api.energy-charts.info/price"
|
|
BZN = "BE"
|
|
|
|
# Written by fetch(), read by startup_preview(), and by nothing else.
|
|
spec = None
|
|
index_by_instant: dict = {}
|
|
|
|
@classmethod
|
|
def capabilities(cls):
|
|
return ProviderCapabilities(
|
|
product=PriceProduct.DAY_AHEAD,
|
|
sources=frozenset({Component.ENERGY, Component.MARKUP}),
|
|
egress_hosts=("api.energy-charts.info", "web-api.tp.entsoe.eu"))
|
|
|
|
async def fetch(self, opts, since, until, transport):
|
|
"""Fetch the index, apply the card's arithmetic, return RawPoints. `since`/`until`
|
|
are used through the shared URL builder (TARIFF-09)."""
|
|
try:
|
|
spec = parse_price_formula(opts.get("price_formula"))
|
|
except FormulaInvalid as e:
|
|
raise SourceUnavailable(f"price_formula: {e}")
|
|
if spec is None:
|
|
raise SourceUnavailable(
|
|
"“Dynamic price source” (price_provider) is 'custom_formula', "
|
|
"but “Your own tariff formula” (price_formula) is empty. "
|
|
"Restart the add-on to check the settings again")
|
|
url = dayahead_feed_url(self.id, opts, since, until,
|
|
self.FEED_BASE, self.BZN)
|
|
check_egress(url, self.capabilities())
|
|
body = await transport(url, float(opts.get("price_timeout_s")))
|
|
# `scale=1.0`, so these RawPoints carry EUR/**MWh** in a field spelled
|
|
# `eur_per_kwh`. Contained: destructured on the next line and never
|
|
# leaves this
|
|
# method (`_validate_raw()` would refuse an escape).
|
|
index = _points_from_columnar_json(body, expect_unit="EUR / MWh",
|
|
scale=1.0,
|
|
provisional_default=False)
|
|
pairs = to_resolution([(p.start, p.eur_per_kwh) for p in index],
|
|
spec.resolution)
|
|
self.spec = spec
|
|
self.index_by_instant = {t.astimezone(timezone.utc): v
|
|
for t, v in pairs}
|
|
# `provisional=False` as for `generic_dayahead`: the auction is settled
|
|
# once published.
|
|
return [RawPoint(start=t, eur_per_kwh=price_eur_kwh(spec, v),
|
|
provisional=False)
|
|
for t, v in pairs]
|
|
|
|
def startup_preview(self, points) -> list:
|
|
"""`index EUR/MWh -> all-in EUR/kWh` for the first 24 h, once. The index beside the price makes a factor of 1000 obvious."""
|
|
if self.spec is None or not points:
|
|
return []
|
|
cut = points[0].start + timedelta(hours=PREVIEW_HOURS)
|
|
rows = []
|
|
for point in points:
|
|
if point.start >= cut:
|
|
break
|
|
index = self.index_by_instant.get(
|
|
point.start.astimezone(timezone.utc))
|
|
if index is not None:
|
|
rows.append((point.start, index, point.eur_per_kwh))
|
|
return preview_lines(self.spec, rows)
|
|
|
|
|
|
# The one entry point
|
|
|
|
# Provider ids whose startup preview was already emitted in this process;
|
|
# filled at runtime.
|
|
_PREVIEWED: set = set()
|
|
|
|
|
|
def _emit_startup_preview(plugin, pid: str, points) -> None:
|
|
"""Log a plugin's preview once per process, and never fail because of it. The id
|
|
is recorded before the hook runs, so a raising preview is not retried every 30 minutes."""
|
|
if pid in _PREVIEWED:
|
|
return
|
|
_PREVIEWED.add(pid)
|
|
try:
|
|
lines = list(plugin.startup_preview(points))
|
|
except Exception as e: # noqa: BLE001
|
|
_LOG.warning("%s could not show its start-up preview (%s: %s). Prices "
|
|
"are not affected.",
|
|
pid, type(e).__name__, e)
|
|
return
|
|
for line in lines:
|
|
_LOG.info("%s", line)
|
|
|
|
|
|
|
|
async def load_prices(opts: dict, since: datetime, until: datetime,
|
|
transport=None) -> PriceRead:
|
|
"""Resolve the configured provider, fetch, assemble all-in, return a read.
|
|
|
|
RETURNS, NEVER RAISES, for any failure a site can have (11.4 is a degradation
|
|
row, not a traceback into the 30 s tick). `FrameworkOption` is re-raised: it is
|
|
a plugin reading the markup, not a site condition."""
|
|
pid = provider_id(opts)
|
|
if not provider_configured(opts):
|
|
return PriceRead.no_data(
|
|
"no price feed is set (price_provider is 'off'), so the fixed "
|
|
"tariff is used, the plan is estimated and there is no trading")
|
|
cls = PROVIDERS.get(pid)
|
|
if cls is None:
|
|
return PriceRead.no_data(
|
|
f"unknown price provider {pid!r}. The known ones are "
|
|
f"{', '.join(sorted(PROVIDERS))}")
|
|
|
|
caps = cls.capabilities()
|
|
markup = float(price_option(opts, "price_markup_eur_kwh"))
|
|
|
|
# TARIFF-17: the table wins over the scalar; validate.py refuses configs
|
|
# where both speak.
|
|
taxes_key = "price_taxes_eur_kwh"
|
|
try:
|
|
table = taxes_profile(opts)
|
|
except Exception as e: # noqa: BLE001
|
|
# Not silent: validate.py should have refused this at startup, so say
|
|
# which key is wrong.
|
|
_LOG.error("price_taxes_profile cannot be used (%s: %s), so no taxes "
|
|
"are added to this price update. Fix the table or clear it.",
|
|
type(e).__name__, e)
|
|
_LOG.debug("price_taxes_eur_kwh is not used instead, because a "
|
|
"silently different tax basis is the fault this setting "
|
|
"exists to remove.")
|
|
return PriceRead.no_data(
|
|
f"price_taxes_profile cannot be used: {e}")
|
|
if table is not None:
|
|
taxes = table
|
|
taxes_key = "price_taxes_profile"
|
|
else:
|
|
taxes = float(price_option(opts, "price_taxes_eur_kwh"))
|
|
|
|
# Not a refusal, but not silent: a plugin whose feed is already all-in
|
|
# ignores these two numbers.
|
|
ignored = [n for c, n, v in ((Component.MARKUP, "price_markup_eur_kwh", markup),
|
|
(Component.TAXES, taxes_key, taxes))
|
|
if c not in caps.adds and v]
|
|
if ignored:
|
|
_LOG.warning("%s publishes a figure that already contains %s, so the "
|
|
"configured %s ignored - adding them would double-count",
|
|
pid,
|
|
"/".join(sorted(c.value for c in
|
|
caps.sources - {Component.ENERGY})),
|
|
" and ".join(ignored) +
|
|
(" is" if len(ignored) == 1 else " are"))
|
|
|
|
if transport is None:
|
|
transport = https_get
|
|
plugin = cls()
|
|
try:
|
|
raw = await plugin.fetch(PluginOptions(opts), since, until, transport)
|
|
_validate_raw(raw, caps)
|
|
points = assemble(caps, raw, markup, taxes)
|
|
except FrameworkOption:
|
|
raise # a plugin bug, not a site condition
|
|
except SourceUnavailable as e:
|
|
return PriceRead.no_data(scrub(f"{pid}: {e}", opts))
|
|
except Exception as e: # noqa: BLE001 - see the docstring
|
|
return PriceRead.no_data(
|
|
scrub(f"{pid}: {type(e).__name__}: {e}", opts))
|
|
|
|
_emit_startup_preview(plugin, pid, points)
|
|
return PriceRead.series(points, caps.product, pid,
|
|
datetime.now(timezone.utc))
|