goodwe-addon/goodwe_controller/app/price.py

"""TARIFF-02 - price provider plugins, and the one canonical shape they all emit.

WHERE THIS LIVES, AND WHY IT IS NOT WHERE THE SPEC SAYS. specs.md §2.2 puts this
at `ems/ingest/prices/`. That module layout does not exist and is not being
built; the product owner's decision of 2026-08-30 is that everything ships in
`goodwe_controller/app/`, next to arbiter.py, clamp.py, freeze.py and
validate.py. The BEHAVIOUR specs.md §9.3 / §8.3 / §11.4 describes is unchanged
and still authoritative - only the path moved.

WHAT THIS IS FOR, in one sentence: so that a new Belgian supplier is a new class
and one line in `PROVIDERS`, and the optimiser that eventually consumes prices
never learns a supplier's name.

THE CANONICAL FORMAT (§9.3), exactly 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 (energy + markup + taxes), and `provisional` is per point,
not per series (§8.3: unknown hours are flagged, and no irreversible action may
be taken on a provisional price).


THE THREE THINGS THIS MODULE IS SHAPED BY
=========================================

1. DOUBLE-COUNTING MARKUP IS REFUSED, NOT DISCOURAGED - GIVEN A CORRECT
   DECLARATION.
   `generic_dayahead` reads a raw EPEX figure and the user's markup and taxes
   have to be added to it. Engie and TotalEnergies publish a figure that already
   HAS the markup in it. Add the configured markup to that one too and every
   arbitrage decision on that site is made against a price that is 3 c/kWh
   wrong, in a direction nothing reports.

   A comment saying "do not add markup here" is not a mechanism. So:

     * A plugin does not return a price. It returns `RawPoint`s carrying the
       figure ITS SOURCE PUBLISHED, and declares in `capabilities().sources`
       which of {ENERGY, MARKUP, TAXES} that figure already contains.
     * `assemble()` - the framework, never a plugin - adds exactly the
       components the plugin declared it did NOT source, from
       `price_markup_eur_kwh` / `price_taxes_eur_kwh`.
     * ⚠️ A PLUGIN CANNOT REACH THOSE TWO NUMBERS BY ANY ROUTE, and the word
       "any" was earned the hard way. `PluginOptions` raises `FrameworkOption`
       on them, which closes the documented path - but review of PR #16 wrote a
       double-counting plugin THREE more ways, one of them touching nothing but
       `PRICE_OPTION_DEFAULTS`, a public dict in the module every plugin must
       already import for `PriceProvider` and `register`. One closed door in a
       room with three is not a guarantee. So `register()` now also REFUSES ANY
       PLUGIN WHOSE OWN CODE MENTIONS A FRAMEWORK KEY - see `code_leaks()`.

   ⚠️ THE PRECISE SHAPE OF THE GUARANTEE, because an overclaim here is worse
   than no claim: **the framework adds the complement exactly once, GIVEN A
   CORRECT `sources` DECLARATION.** A plugin that declares `{ENERGY}` while its
   feed is really all-in still comes out one markup high, and NO CODE CAN CATCH
   THAT - the declaration is the plugin's unverifiable claim about somebody
   else's tariff. What is mechanised is that a plugin cannot obtain the numbers
   and cannot apply them; what remains human is whether it described its feed
   truthfully. The docstring/declaration check in test_price.py verifies that
   the plugin's two claims AGREE, not that either is true, and it is worth
   knowing which of those two things you are reading.

2. THE PRODUCT TYPE IS DECLARED, NOT INFERRED FROM THE PLUGIN'S NAME.
   `eneco_eddy` is a real-time / imbalance product (§9.2) and §8.3 says those
   are treated as high-volatility with widened arbitrage thresholds. If the
   optimiser has to write `if provider_id == "eneco_eddy"` to know that, then
   adding a sixth supplier means editing the optimiser - which is precisely
   what AC2 forbids. So `capabilities().product` is a `PriceProduct`, it rides
   along on the series, and `volatility_regime()` reads the enum. A plugin id
   appears nowhere in this file except in `PROVIDERS` and in the class that
   answers to it.

3. "NO PRICE DATA" IS A TYPE, NOT AN EMPTY LIST.
   §11.4: no price data -> fall back to the static profile, mark the plan
   provisional, NO ARBITRAGE. That is a different instruction from "prices
   exist and they are all zero", under which arbitrage is merely pointless
   rather than forbidden - and the two are indistinguishable if a failed fetch
   returns `[]`, or `None`, or a dict a caller `.get()`s a default out of.

   ⚠️ THIS REPO SPENT 2026-08-30 REMOVING EXACTLY THAT DEFECT. `sim/riglock.py`
   in the infra repo used to return a partial dict from `holder()`; TEST-04
   rewrote it so a look at the lock is `observed / not_observed / gone` and only
   the observed variant carries fields. `_Outcome` below is that shape,
   deliberately, down to the fatal `.get()`. Read that file's `class Absence`
   block before changing anything here.

   Concretely:

     * `PriceRead.series(...)` REFUSES AN EMPTY POINT LIST. A series with no
       prices in it cannot be constructed, so "I have no prices" cannot arrive
       wearing the shape of an answer.
     * `PriceRead.no_data(reason)` carries `.reason` AND NOTHING ELSE. It has no
       `.points` to iterate, no `.product` to switch on, no `.at()` to call, and
       `.get("points", [])` raises. `min(read.points)` - the line that would
       silently make "no data" mean "everything is free, charge now" - does not
       run against it.
     * `.match(*, series, no_data)` takes both handlers as REQUIRED keyword
       arguments, so a caller who forgets the no-data case gets a TypeError at
       the call site rather than a fall-through.


TIMEZONE, AND WHY THERE IS NO `zoneinfo` HERE
=============================================
§9.3's example is `+01:00`, and Belgium runs CET/CEST. Two constraints:

  * The shipped image is Alpine with `py3-aiohttp` from apk and NO `tzdata`
    package (goodwe_controller/Dockerfile, and .gitea/ci/Dockerfile is the same
    base by design). `ZoneInfo("Europe/Brussels")` raises
    `ZoneInfoNotFoundError` there. A price module that only works on a machine
    with a tz database is a price module that fails on the client's Pi.
  * Fixed offsets have no fold. Every instant here is carried as UTC and
    RENDERED at the offset that applies to it, so the 02:00-03:00 hour that
    happens twice in October is two different instants that render as
    `02:00+02:00` and `02:00+01:00`. A naive-local-hour key collides those two
    and misprices an hour of a 25-hour day; an implementation on a fixed
    24-slot array is wrong on both transition days. test_price.py drives both
    2026 transitions and counts the points rather than asserting 24.

The EU rule is fixed in law and needs no database: CEST from 01:00 UTC on the
last Sunday of March until 01:00 UTC on the last Sunday of October. Review of
PR #16 cross-checked this against `zoneinfo`'s Europe/Brussels over 149,040
hourly instants, 2020-2036 inclusive, and found ZERO mismatches - so the
residual risk here is not arithmetic, it is political: if the EU ever abolishes
the seasonal change, `brussels_offset()` needs editing. That is the same
exposure as a machine whose `tzdata` has gone stale, and it is one edit in one
function rather than a package upgrade.
"""

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

_LOG = logging.getLogger("goodwe.price")


# --------------------------------------------------------------------------- #
# Brussels wall time, without a tz database. See the module docstring.
# --------------------------------------------------------------------------- #

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 one. A naive datetime is not a
    moment in time, it is a wall clock reading plus an assumption, and the
    assumption is 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))


# --------------------------------------------------------------------------- #
# Absence, borrowed wholesale 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, and turned into `PriceRead.no_data(...)`.

    ⚠️ It must never escape into the control tick. §11.4's row is a returned
    condition, not a traceback: the loop keeps regulating on P1 whatever the
    price situation is, and a price fetch that could stop it would be a price
    fetch that can stop the battery.
    """


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)` are the other two
        # idioms that turn an absent read into an empty, harmless-looking 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"


# ⚠️ §8.3: "Real-time / imbalance products are treated as high-volatility:
# arbitrage thresholds widen." This maps product -> regime, and it is the ONLY
# place in the shipped code that says which products are volatile. OPT-04 reads
# it; it does not re-derive it, and it certainly does not check plugin ids.
_WIDE = frozenset({PriceProduct.REALTIME})


def volatility_regime(product: PriceProduct) -> str:
    """"wide" | "normal" - the arbitrage threshold regime for a product (§8.3).

    Deliberately a plain string and deliberately tiny: OPT-04 owns what the
    widened numbers ARE. This owns only which products get them, because that
    is a property of the product and belongs with the product enum.
    """
    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.
#
# Two spec rows widen the same number by two different triggers:
#
#   §9.2 / §8.3  real-time / imbalance product   -> "arbitrage thresholds widen"
#   §11.4        no weather data                 -> "widen arbitrage thresholds"
#
# They are one mechanism reached twice, not two mechanisms, so there is one
# function and a set of reasons. ⚠️ THE COMBINATION RULE IS `max`, NOT A
# PRODUCT: multiplying 2.5 by 1.5 would make two independent doubts compound
# into a threshold nothing on a Belgian tariff can ever clear, and an arbitrage
# strategy that can never fire is indistinguishable from one that is switched
# off. The widest single reason wins.
#
# ⚠️ HONEST ABOUT WHERE THE NUMBERS COME FROM. specs.md says thresholds widen
# and does not say by how much; there is no capture to derive a factor from,
# because no arbitrage strategy has ever run here. These are POLICY DEFAULTS,
# chosen so that (a) a real-time product needs a materially larger spread than
# a settled day-ahead one before a cycle is worth its wear, and (b) the two
# factors DIFFER, so `max`, `sum` and `product` are three distinguishable
# answers rather than the same number - a check that grades a combination rule
# where all the candidates agree grades nothing. When OPT-04 lands
# `cycle_cost_threshold_eur` (§10.1/§14) these become the natural second option
# beside it, in PRICE_OPTION_DEFAULTS and nowhere else (CFG-03).
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 - the same rule
    `volatility_regime()` exists for. A sixth supplier declaring
    `PriceProduct.REALTIME` is widened without anyone editing this file.
    """
    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. `widen_factor({"realtime"})` would otherwise be a
    silent 1.0 - a widening that reports itself as applied and is not, which is
    the shape of defect this repo keeps finding.
    """
    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 the caller's own round-trip cost threshold - OPT-04 owns
    that number (§10.1's `cycle_cost_threshold_eur`), this owns only how much
    wider it gets and why.
    """
    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. §9.3 lists three; this has three."""

    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, and the type difference is the guard. A plugin cannot
    hand back something already claiming to be all-in, because the only
    constructor of `PricePoint` that `load_prices()` uses is `assemble()`.
    """

    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
    needs_credentials: bool = False
    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. Two outcomes, and only one answers.

      * `series`  - we have prices. Carries `.points` (NON-EMPTY, by
                    construction), `.product`, `.provider`, `.fetched_at`.
      * `no_data` - we do not. Carries `.reason` and NOTHING ELSE. §11.4: static
                    profile, plan provisional, no arbitrage.

    ⚠️ A SERIES OF ALL-ZERO PRICES IS A `series`, NOT `no_data`, and that
    distinction is the whole point. Zero prices mean arbitrage is pointless;
    no data means arbitrage is FORBIDDEN. An implementation that returns `[]`
    on a failed fetch cannot tell a caller which of those two it is in.
    """

    def match(self, *, series, no_data):
        """Fold to a value. Both handlers required and keyword-only.

        Handlers take one argument: this read. Required-ness is what makes the
        no-data case impossible to forget rather than merely easy to remember.
        """
        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)


class _Series(PriceRead):
    kind = "series"

    def __init__(self, points, product, provider, fetched_at):
        points = list(points)
        # ⚠️ THE REFUSAL THAT MAKES THE TYPE WORTH HAVING. An empty series is
        # how "the fetch failed" gets to wear the shape of "we looked and there
        # is nothing expensive today". It is not constructible.
        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)
        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 covering `instant`. Raises `Absence` if the series does not.

        ⚠️ RETURNS A POINT OR RAISES; it never returns None. `series.at(t) or
        fallback` is the same defaulting bug one level down, so the falsy
        answer does not exist. Ask `covers()` first if you mean to branch.
        """
        u = instant.astimezone(timezone.utc)
        for p in self.points:
            s = p.start.astimezone(timezone.utc)
            if s <= u < s + timedelta(hours=1):
                return p
        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 hour outside the series "
            f"is an hour with no price, which is §11.4 for that hour - not a "
            f"reason to reach for a neighbouring one.")

    def covers(self, instant: datetime) -> bool:
        try:
            self.at(instant)
            return True
        except Absence:
            return False

    def window(self, since: datetime, until: datetime) -> tuple:
        """Points whose start falls in [since, until). May legitimately be empty."""
        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.

    ⚠️ THIS CLASS HAS NO `.points` AND THAT IS THE POINT. `min(read.points)`,
    `for p in read.points`, `read.get("points", [])` and `len(read.points)` all
    raise `Absence` here rather than quietly reporting that the cheapest hour of
    the day costs nothing.

    ⚠️ TWO WAYS PAST IT SURVIVE, BOTH FOUND IN REVIEW OF PR #16, AND NEITHER IS
    WORTH CODE:

      * `vars(read).get("points", [])` returns `[]`. `vars()` bypasses
        `__getattr__` entirely; nothing short of `__slots__` plus a metaclass
        would stop it, and anyone writing `vars()` on a variant type has left
        the road on purpose.
      * `bool(read)` is `True` for BOTH variants, so `if read:` looks like a
        check and is not one. Defining `__bool__` to be False here would make
        `if read:` "work" - and that is exactly the wrong fix: it would turn a
        line that reads like a guard into one that IS a guard by accident,
        which is how `.get(k, default)` got into this codebase in the first
        place. Use `.match(...)`. There is no falsy variant on purpose.
    """

    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, stated once, as a function so no caller has to remember it.

    Not the arbitrage strategy - that is OPT-04. This is only the veto, which
    belongs with the type that expresses the condition.
    """
    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 in
    the series 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
# --------------------------------------------------------------------------- #

# ⚠️ CFG-03, AND HOW THIS TICKET AVOIDS WIDENING IT. Option defaults in this
# add-on already live in three places - config.yaml, main.py's `o.get(k, X)`
# fallbacks, and a dataclass default - held apart only by a tripwire in
# test_validate.py. This module adds ONE more location, not another three:
#
#   * every price option's fallback is here, and only here;
#   * validate.py imports this dict rather than restating any number;
#   * main.py gets no `o.get("price_...", X)` fallbacks at all;
#   * there is no dataclass carrying a default copy of any of these;
#   * test_price.py READS config.yaml and asserts the shipped default equals
#     this dict, key for key, so the two cannot drift silently.
#
# config.yaml remains the source of truth. This is the single in-code mirror,
# pinned to it by a check, which is the `validate.py` pattern rather than a
# fourth freehand copy.
PRICE_OPTION_DEFAULTS = {
    "price_provider": "off",
    "price_markup_eur_kwh": 0.03,      # §14 tariffs.energy_price.markup_eur_kwh
    "price_taxes_eur_kwh": 0.06,       # §14 tariffs.energy_price.taxes_eur_kwh
    "export_price_eur_kwh": 0.04,      # §14 tariffs.export_price_eur_kwh (AC4)
    "price_csv_path": "/share/goodwe_prices.csv",
    "price_feed_url": "",
    "price_horizon_h": 48,
    "price_timeout_s": 10,
    # TARIFF-03. §14 `tariffs.static_profile: null`. A JSON object describing a
    # static day/night or fully-fixed tariff; "" is §14's `null`. Parsed and
    # validated by app/static_tariff.py, refused at startup by validate.py.
    #
    # ⚠️ IT LIVES HERE RATHER THAN IN static_tariff.py FOR CFG-03's SAKE. This
    # dict is the ONE in-code mirror of config.yaml, pinned to it by
    # test_price.py section 1. A second defaults dict in the new module would be
    # a fourth place option defaults live, which is exactly the open ticket.
    "static_profile": "",
}

# ⚠️ The keys the FRAMEWORK owns. A plugin that reads one of these is a plugin
# that 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",
})


class PluginOptions:
    """The options dict, minus the keys only the framework may read.

    Reading `price_markup_eur_kwh` here raises `FrameworkOption` rather than
    returning 0.03.

    ⚠️ THIS IS ONE OF THE TWO DOORS, NOT THE ONLY ONE. `_o` is reachable, and
    `price_option()` and `PRICE_OPTION_DEFAULTS` are public names in this
    module. This class alone is a convention; it is `register()`'s
    `code_leaks()` scan that makes it an inability. Do not weaken either half
    on the strength of the other.
    """

    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 export_price_eur_kwh(opts: dict) -> float:
    """AC4 / §9.3: the injection price is a SEPARATE SCALAR, not a second series.

    In Belgium it is often near zero, which by itself favours self-consumption
    over export. It is deliberately not routed through the plugin framework: no
    shipped supplier publishes a dynamic injection curve, and inventing the
    shape for one would be a second canonical format nobody emits. When one
    appears it gets a `PriceProvider` of its own and this scalar becomes its
    FIXED-product default - the seam is already the right shape for that.

    ⚠️ It is NOT reachable from a plugin: `export_price_eur_kwh` is in
    FRAMEWORK_KEYS, so a consumption plugin cannot quietly fold an injection
    price into a consumption price.
    """
    return float(price_option(opts, "export_price_eur_kwh"))


# --------------------------------------------------------------------------- #
# 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()` DOES NOT RETURN A PRICE. It returns `RawPoint`s carrying the
    published figure. The all-in arithmetic happens in `assemble()`, once, in
    the framework, using `capabilities().sources`. See trap 1.
    """

    id = "?"

    @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 so the checks never
        touch a network and `manual_csv` never needs one. `opts` is a
        `PluginOptions`, so the markup and taxes are not reachable from here.
        """


# ⚠️ THE NAMES A PLUGIN'S OWN CODE MAY NOT MENTION. FRAMEWORK_KEYS is the
# obvious half; the other three are the routes review of PR #16 actually used to
# write a double-counting plugin after `PluginOptions` was in place:
#
#   `_o`                     - PluginOptions' private dict, reachable as
#                              `opts._o["price_markup_eur_kwh"]`.
#   `price_option`           - a PUBLIC framework helper that takes the raw
#                              options dict and answers with the markup.
#   `PRICE_OPTION_DEFAULTS`  - a PUBLIC module constant. This is the one that
#                              decided the review: no private access, no options
#                              dict, no helper - just a dict in the module every
#                              plugin already imports for `register`.
#
# Any of them turns +0.088 into +0.119 on a wholesale feed with nothing
# anywhere reporting it.
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 BEFORE THE SEARCH, AND THAT IS NOT COSMETIC.
    Every shipped plugin's docstring is REQUIRED to name the components it
    sources, so `engie_dynamic` says "markup" in prose and would false-positive
    against `price_markup_eur_kwh` on a raw text scan. The check that fires on
    the documentation it also mandates is the check that gets deleted. So the
    class is parsed, every docstring node is dropped, and the search runs on
    `ast.unparse` of what is left - i.e. on code, which is the only thing that
    can actually add a number twice.

    ⚠️ FAIL-OPEN ON MISSING SOURCE, DELIBERATELY. `inspect.getsource` needs the
    .py file, which the add-on image has (Dockerfile COPYs `app/`) but a frozen
    or zipapp build would not. Refusing to start a battery controller because a
    STATIC ANALYSIS could not run would trade a real outage for a hypothetical
    accounting error, so an unreadable source returns `[]` and logs. CI proves
    the scan really runs: test_price.py asserts source was found for all five
    shipped plugins, so a build that silently lost it is a red check, not a
    quietly disarmed guard.
    """
    try:
        src = dedent(inspect.getsource(cls))
        tree = ast.parse(src)
    except (OSError, TypeError, IndentationError, SyntaxError) as e:
        _LOG.warning("could not read %s's source for the framework-key scan "
                     "(%s); the double-counting guard is NOT enforced for it",
                     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})


PROVIDERS: dict = {}


def register(cls):
    """Register a provider by its own declared id. Decorator.

    ⚠️ THE ONLY NAME->CLASS MAPPING IN THE SYSTEM. AC2 is "adding a plugin needs
    no optimiser changes", and this is where that is bought: config resolves an
    id here, the optimiser is handed a `_Series` carrying a `PriceProduct`, and
    nothing downstream ever sees the id except to print it.
    """
    if cls.id in PROVIDERS:
        raise ValueError(f"duplicate provider id {cls.id!r}")
    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 is a
        # plugin that can bill wrong for a month before anyone notices, so it
        # does not get to exist - the add-on fails to start and names the line.
        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 ARE NOT CLAMPED (§17 keeps a negative-price
    regression case; Belgian day-ahead goes negative on windy Sundays and those
    hours are the single best charging opportunity of the year). What is
    rejected is what cannot be a price at all: NaN, infinity, an implausible
    magnitude, a naive datetime, and 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} - the usual cause is a feed "
                f"in EUR/MWh being read as EUR/kWh, which is a factor of 1000 "
                f"and would make every hour look like an emergency")
        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 assemble(caps: ProviderCapabilities, raw, markup: float,
             taxes: float) -> list:
    """RawPoints -> canonical all-in PricePoints. THE ONLY PLACE MARKUP IS ADDED.

    Adds exactly `caps.adds` - the complement of what the plugin declared its
    source already contains. A plugin that publishes an all-in figure gets
    nothing added; a raw EPEX figure gets both.
    """
    add = 0.0
    if Component.MARKUP in caps.adds:
        add += float(markup)
    if Component.TAXES in caps.adds:
        add += float(taxes)
    return [PricePoint(start=brussels(p.start),
                       eur_per_kwh=round(p.eur_per_kwh + add, 6),
                       provisional=bool(p.provisional))
            for p in raw]


# --------------------------------------------------------------------------- #
# Transport - one place, HTTPS only, host checked against the declaration
# --------------------------------------------------------------------------- #

async def https_get(url: str, timeout_s: float) -> bytes:
    """The default transport. §16: outbound is allow-listed and TLS'd.

    Imported lazily so `manual_csv` - the offline fallback - never needs aiohttp
    to be importable at all. AC3 says it works fully offline; that includes a
    machine where the HTTP stack is missing.
    """
    import aiohttp
    t = aiohttp.ClientTimeout(total=float(timeout_s))
    async with aiohttp.ClientSession(timeout=t) as s:
        async with s.get(url) as r:
            r.raise_for_status()
            return await r.read()


def check_egress(url: str, caps: ProviderCapabilities) -> str:
    """§16: a plugin may only reach a host it DECLARED. Enforced, not documented.

    Returns the host, so a caller that forgets to call this is not the same
    shape as one that did.

    ⚠️ THE SCHEME IS SPLIT OFF BEFORE THE HOST IS PARSED, AND THE ORDER MATTERS
    FOR THE TEST MORE THAN FOR THE CODE. The first version stripped a fixed
    `len("https://")` from the front, which on an `http://` URL ate two
    characters of the hostname instead - `http://api.energy-charts.info/p`
    became the host `pi.energy-charts.info`, which then failed the ALLOW-LIST.
    So a non-TLS feed was refused, the check named "plain http is refused"
    passed, and deleting the TLS rule entirely left it passing. Review of
    PR #16 found that by mutation: an assertion that does not test what it
    names, inside a security check - this project's own recurring defect. Now
    the scheme is parsed as a scheme, the host of an `http://` URL comes out
    correct and DECLARED, and the TLS rule is the only thing that can refuse
    it - which is what makes the check able to fail.
    """
    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 five shipped plugins.
#
# ⚠️ EACH DOCSTRING STATES WHICH COMPONENTS IT SOURCES AND WHICH THE FRAMEWORK
# ADDS, in the two fixed lines `SOURCES: ...` / `The framework ADDS: ...`. That
# is not decoration - it is the human-readable half of `capabilities().sources`,
# and test_price.py PARSES those lines out of the docstring and compares them
# with the declaration, so a plugin whose prose and whose code disagree about
# what its feed contains is a red check rather than a 3 c/kWh error nobody sees.
# --------------------------------------------------------------------------- #

def _points_from_json(body, price_key, scale, provisional_default):
    """Shared parse for the three JSON feeds. Shape differences are the args."""
    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}")
    rows = doc.get("data") if isinstance(doc, dict) else doc
    if not isinstance(rows, list):
        raise SourceUnavailable("the feed carried no 'data' list")
    out = []
    for row in rows:
        try:
            start = datetime.fromisoformat(str(row["start"]))
            value = float(row[price_key]) * scale
        except (KeyError, TypeError, ValueError) as e:
            raise SourceUnavailable(f"malformed row {row!r}: {e}")
        if start.tzinfo is None:
            raise SourceUnavailable(
                f"row {row['start']!r} carries no UTC offset; this feed is "
                f"required to publish absolute instants")
        out.append(RawPoint(start=start, eur_per_kwh=value,
                            provisional=bool(row.get("provisional",
                                                     provisional_default))))
    return out


@register
class GenericDayAhead(PriceProvider):
    """Public EPEX BE day-ahead feed, in EUR/MWh.

    SOURCES: energy
    The framework ADDS: markup, taxes

    This is the plugin the double-counting trap is about. The published figure
    is the wholesale clearing price and nothing else, so both user-entered
    components have to be added to it - and because this plugin cannot read
    either number, it cannot be the one that adds them twice.
    """

    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"))

    async def fetch(self, opts, since, until, transport):
        url = opts.get("price_feed_url") or \
            "https://api.energy-charts.info/price?bzn=BE"
        check_egress(url, self.capabilities())
        body = await transport(url, float(opts.get("price_timeout_s")))
        return _points_from_json(body, "price_eur_mwh", 0.001, False)


@register
class EngieDynamic(PriceProvider):
    """Engie's dynamic day-ahead product, published in EUR/kWh.

    SOURCES: energy, markup
    The framework ADDS: taxes

    ⚠️ Engie's published figure ALREADY CONTAINS THEIR MARKUP - it is a retail
    price, not a wholesale one. Adding `price_markup_eur_kwh` on top of it is
    the defect this ticket's design exists to make unreachable; the declaration
    above is what stops it, and nothing here needs to remember the rule.
    """

    id = "engie_dynamic"

    @classmethod
    def capabilities(cls):
        return ProviderCapabilities(
            product=PriceProduct.DAY_AHEAD,
            sources=frozenset({Component.ENERGY, Component.MARKUP}),
            needs_credentials=True,
            egress_hosts=("api.engie.be",))

    async def fetch(self, opts, since, until, transport):
        url = opts.get("price_feed_url") or "https://api.engie.be/dynamic/be"
        check_egress(url, self.capabilities())
        body = await transport(url, float(opts.get("price_timeout_s")))
        return _points_from_json(body, "price_eur_kwh", 1.0, False)


@register
class TotalEnergiesDynamic(PriceProvider):
    """TotalEnergies' dynamic day-ahead product, published in EUR/kWh.

    SOURCES: energy, markup
    The framework ADDS: taxes

    Same retail shape as Engie: the supplier's own margin is inside the number.
    """

    id = "totalenergies_dynamic"

    @classmethod
    def capabilities(cls):
        return ProviderCapabilities(
            product=PriceProduct.DAY_AHEAD,
            sources=frozenset({Component.ENERGY, Component.MARKUP}),
            needs_credentials=True,
            egress_hosts=("api.totalenergies.be",))

    async def fetch(self, opts, since, until, transport):
        url = opts.get("price_feed_url") or \
            "https://api.totalenergies.be/prices/dynamic"
        check_egress(url, self.capabilities())
        body = await transport(url, float(opts.get("price_timeout_s")))
        return _points_from_json(body, "price_eur_kwh", 1.0, False)


@register
class EnecoEddy(PriceProvider):
    """Eneco Eddy / Eddy Flex - a REAL-TIME / IMBALANCE product, all-in EUR/MWh.

    SOURCES: energy, markup, taxes
    The framework ADDS: nothing

    ⚠️ THE PRODUCT TYPE IS THE INTERESTING FIELD HERE, NOT THE PRICE. §9.2
    classes Eddy Flex as real-time / imbalance and §8.3 says those get widened
    arbitrage thresholds. That regime reaches the optimiser through
    `capabilities().product` -> `_Series.product` -> `volatility_regime()`. If
    it reached the optimiser through the string "eneco_eddy" instead, then the
    sixth supplier would need an optimiser edit and AC2 would be dead. This
    class is the reason the enum exists.

    Imbalance prices are settled after the fact, so the current quarter's figure
    is genuinely provisional and the feed says so per point.
    """

    id = "eneco_eddy"

    @classmethod
    def capabilities(cls):
        return ProviderCapabilities(
            product=PriceProduct.REALTIME,
            sources=ALL_IN,
            horizon_h=24,
            needs_credentials=True,
            egress_hosts=("api.eneco.be",))

    async def fetch(self, opts, since, until, transport):
        url = opts.get("price_feed_url") or "https://api.eneco.be/eddy/prices"
        check_egress(url, self.capabilities())
        body = await transport(url, float(opts.get("price_timeout_s")))
        return _points_from_json(body, "price_eur_mwh", 0.001, True)


@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, AND DOCS.md says so. It has to be: this plugin has no
    way to know what the numbers a human typed already contain, and guessing
    would be the double-counting defect with a friendlier face. `load_prices()`
    warns when markup/taxes are configured non-zero under a plugin that sources
    them, so the ignored numbers are visible rather than silently discarded.

    ⚠️ NO TRANSPORT, NO aiohttp, NO NETWORK, NO tz DATABASE. It is the escape
    hatch for a site with no internet and the ingest path the TEST-01 replay
    harness will share, so it must not acquire a dependency on anything that
    needs one. `transport` is accepted and deliberately unused.

    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


# --------------------------------------------------------------------------- #
# The one entry point
# --------------------------------------------------------------------------- #

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 every failure a site can actually have. §11.4
    makes "no price data" a degradation row, and a degradation row that arrives
    as a traceback into the 30 s planning tick is a controller that stops
    regulating a battery because a supplier's TLS certificate expired.

    The one exception is `FrameworkOption`, which is re-raised: that is not a
    site condition, it is a plugin trying to read the markup, and turning it
    into "no price data" would hide the exact defect trap 1 exists to catch.
    """
    pid = str(price_option(opts, "price_provider") or "off").strip()
    if pid in ("", "off"):
        return PriceRead.no_data(
            "no price provider configured (price_provider is 'off'); "
            "§11.4: static profile, plan provisional, no arbitrage")
    cls = PROVIDERS.get(pid)
    if cls is None:
        return PriceRead.no_data(
            f"unknown price provider {pid!r}; shipped: "
            f"{', '.join(sorted(PROVIDERS))}")

    caps = cls.capabilities()
    markup = float(price_option(opts, "price_markup_eur_kwh"))
    taxes = float(price_option(opts, "price_taxes_eur_kwh"))

    # ⚠️ Not a refusal, but not silent either. A plugin whose feed is already
    # all-in ignores these two numbers, and an operator who typed them has every
    # reason to believe they took effect.
    ignored = [n for c, n, v in ((Component.MARKUP, "price_markup_eur_kwh", markup),
                                 (Component.TAXES, "price_taxes_eur_kwh", 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
    try:
        raw = await cls().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(f"{pid}: {e}")
    except Exception as e:          # noqa: BLE001 - see the docstring
        return PriceRead.no_data(f"{pid}: {type(e).__name__}: {e}")

    return PriceRead.series(points, caps.product, pid,
                            datetime.now(timezone.utc))