goodwe-addon/goodwe_controller/app/static_tariff.py

"""TARIFF-03 - static day/night and fully-fixed tariffs, first-class.

WHERE THIS LIVES, AND WHY IT IS NOT WHERE THE SPEC SAYS. specs.md §2.2 puts the
price machinery at `ems/tariff/...`. 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 price.py, arbiter.py and validate.py.
The BEHAVIOUR §9.2 / §9.3 / §11.4 / §10.1 describes is unchanged and still
authoritative - only the path moved.

WHY THIS TICKET EXISTS, in §9.2's own words: "a large share of the installed
base is on [a static profile], and an EMS that only works under dynamic pricing
is an EMS for a minority." The §9.2 product table maps four products onto four
EMS behaviours, and TARIFF-02 shipped the machinery for the top two only:

    Dynamic day-ahead    hourly EPEX BE + markup     full arbitrage
    Real-time/imbalance  e.g. Eneco Eddy Flex        arbitrage, wider thresholds
    Static day/night     two-period tariff           TIME-WINDOW SHIFTING   <- here
    Fully fixed          single price                SELF-CONSUMPTION ONLY  <- here


THE PLACEMENT DECISION, AND THE TWO REASONS IT IS NOT A PROVIDER PLUGIN
=======================================================================
The obvious move is a sixth `@register`ed `PriceProvider` alongside
`generic_dayahead` and `manual_csv`. It was considered and rejected, on two
grounds that are mechanical rather than aesthetic:

  1. A PROVIDER'S PRODUCT IS FIXED AT CLASS-DEFINITION TIME AND THIS ONE IS NOT.
     `PriceProvider.capabilities()` is a `classmethod` taking no options, and
     `load_prices()` reads `caps.product` to stamp the series. But §9.2 makes
     STATIC and FIXED two DIFFERENT products with two different required EMS
     behaviours, and which of the two a site is on is a property of the
     OPERATOR'S DATA - one period or several - not of any class. A plugin would
     have to declare one of them and be wrong for half its users, or the
     framework would have to grow a per-call product override, which is a
     change to TARIFF-02's contract for the benefit of one caller.
     `StaticProfile.product` derives it from the periods instead (see below).

  2. THE PLUGIN FRAMEWORK EXISTS TO SOLVE A PROBLEM THIS PATH DOES NOT HAVE.
     `capabilities().sources` / `assemble()` exist because a FETCHED figure may
     or may not already contain the supplier's markup, and adding it twice is
     invisible (price.py, trap 1). A static profile's numbers are typed by the
     operator as all-in; there is no publisher to disagree with, so the whole
     `RawPoint` -> `assemble()` -> `PricePoint` pipeline would be an identity
     function wrapped in a declaration nobody can check.

  ⚠️ WHAT IS *NOT* A REASON: "it is offline". `manual_csv` is offline and is a
  plugin, correctly. Offline-ness is not what separates these two.

So the static profile sits UNDER the provider framework, as the layer §11.4
already says it is:

    §11.4  No price data -> fall back to the static profile, mark the plan
                            provisional, no arbitrage.

which makes the §9.2 product option and the §11.4 degradation row THE SAME CODE
PATH, differing in exactly one flag. A fully-static site is one that leaves
`price_provider: off` (already the shipped default) and fills in
`static_profile`; a dynamic site whose feed died falls into the same generator
with `provisional=True`. One generator, two call sites, one flag - so the
fallback cannot rot, because it is the path a whole class of installs runs on
every day.

It emits `PricePoint`s in TARIFF-02's canonical `{start, eur_per_kwh,
provisional}` shape (§9.3, "so the optimiser compares like with like"), so
every downstream consumer works unchanged and none of them learns that a
tariff can be static.


HOW "NO ARBITRAGE" IS EXPRESSED SO A FUTURE OPTIMISER CANNOT IGNORE IT
======================================================================
A boolean called `arbitrage_ok` that OPT-04 is asked to remember to check is not
a mechanism; it is a comment with a type. So the numbers an arbitrage strategy
NEEDS - the spread, and which hours are the cheap and dear ends of it - EXIST
ONLY ON THE PERMITTING VARIANT of `Arbitrage`. `refused` carries `.reason` and
nothing else, and reaches for anything else raises `Absence`, the same shape
price.py borrowed from `sim/riglock.py`. An optimiser cannot compute a trade
from a refusal, because it cannot get the inputs out of one.

And the verdict is only obtainable from `TariffPlan.arbitrage(...)`, which takes
`peak_guard_active`, `min_spread_eur_kwh` and `uncertainty` as REQUIRED KEYWORD
ARGUMENTS, none of them defaulted. §9.1 and §10.1 put peak
shaving (priority 2) above price arbitrage (priority 3) - the capacity tariff is
a network cost charged whatever the energy product is - and the way that
ordering is made to survive OPT-04 landing is that OPT-04 cannot obtain a spread
without first stating whether the peak guard is holding. It is not free to
forget the question; the call does not compile without an answer.

⚠️ FIVE REFUSALS, IN THIS ORDER, EACH DERIVED FROM SOMETHING MEASURED:

  1. the peak guard is active           §10.1: priority 2 outranks priority 3
  2. no price data at all               §11.4, and no static_profile to fall to
  3. fewer than TWO FIRM HOURS          §8.3. FCST-03 rewrote this one: it used
                                        to be "any point in the series is
                                        provisional", which would have switched
                                        arbitrage off every morning on every
                                        dynamic site once carry-forward filling
                                        landed (a series is part-firm and
                                        part-carried until the 14-16 h CET
                                        publication). What §8.3 forbids is
                                        acting ON a provisional price, so the
                                        provisional hours are DROPPED and
                                        everything below is computed over the
                                        firm ones. A wholly provisional series -
                                        the §11.4 fallback - still has no firm
                                        pair and is still refused.
  4. the FIRM series is FLAT            max - min == 0, so there is nothing to
                                        arbitrage. THIS is TARIFF-03's AC2: a
                                        fully-fixed tariff disables arbitrage by
                                        ARITHMETIC, not by a special case keyed
                                        on `PriceProduct.FIXED`. A dynamic feed
                                        that happens to print one price all day
                                        gets refused for the same true reason.
  5. the spread does not clear the      §8.3/§9.2/§11.4, new in FCST-03. The
     WIDENED threshold                  caller states its round-trip cost
                                        threshold; this widens it by the widest
                                        active uncertainty source - a real-time
                                        / imbalance PRODUCT (derived from the
                                        declared `PriceProduct`, never from a
                                        plugin id) or the caller's own
                                        `NO_WEATHER`. See price.py's
                                        `required_spread()`; the combination
                                        rule is `max`, not a product.

Nothing here decides a trade. The spread is handed over; OPT-04 owns
`cycle_cost_threshold_eur` (§10.1/§14) and what to do about it. ⚠️ Worth knowing
before that lands: on the Belgian day/night numbers in DOCS.md the whole-tariff
spread is 0.08 EUR/kWh, which is BELOW the shipped 0.10 threshold - so a
correct OPT-04 will decline to cycle for the tariff alone on such a site. That
is a product finding, not a bug here.

WHAT STAYS ON WHEN ARBITRAGE IS OFF (AC2, and §10.1 in full): priorities 1
(self-consumption), 2 (peak shaving), 4 (reserve management) and 5 (load
shifting) are untouched by any refusal above. Only priority 3 goes away. In
0.5.x that is literally true because none of them reads a price: the grid
following loop, the capacity-tariff peak tracker and the freeze/sign watchdogs
all run off P1 and know nothing about this module.


TIME-WINDOW SHIFTING, AND THE ONE SHIFTABLE LOAD THAT EXISTS TODAY
==================================================================
§9.2's handling for a two-period tariff is "time-window shifting", and §10.1
priority 5 is load shifting. The add-on has exactly one shiftable load in
0.5.x: the maintenance cycle's cheap-window grid top-up. `static_status()` is
what main.py reads for it, and `cheap_now()` is the rule:

    cheap  <=>  the price now is STRICTLY BELOW the mean over the horizon

⚠️ STRICTLY BELOW, AND THE STRICTNESS IS AC2. Under a fully-fixed tariff every
hour equals the mean, so `<=` would make EVERY hour a cheap window and a fixed
tariff site would top the battery up from the grid at any hour of the day -
which is the exact opposite of "self-consumption only". With `<`, a flat series
has no cheap window at all, derived rather than special-cased. (The Home
Assistant `price_now_entity`/`price_avg_entity` pair main.py has always used
computes `now <= avg`; the difference is deliberate and is why this is not
simply the same expression.)

⚠️ THE MEAN IS TAKEN OVER SEVEN LOCAL DAYS, not over the 48 h forecast horizon,
and that is not tidiness. Belgian night tariffs usually cover the whole weekend,
so a 48 h mean computed on a Friday afternoon is weighted about 2/3 weekend and
the SAME Tuesday 09:00 comes out cheap or dear depending on the day you asked.
Seven days contains each weekday exactly once, so the mean is a property of the
tariff instead of a property of the hour you called at.


DAYLIGHT SAVING, AND WHY THERE IS NO `zoneinfo` HERE EITHER
============================================================
The shipped Alpine image has no `tzdata` (price.py's docstring, and the
Dockerfile), so `ZoneInfo("Europe/Brussels")` raises there. price.py already
carries `brussels()` / `brussels_offset()` - the EU rule, in arithmetic, checked
in review of PR #16 against `zoneinfo` over 149,040 hourly instants with zero
mismatches - and this module uses it rather than growing a second answer to the
same question.

Generation iterates HOURLY IN UTC and renders each instant to local, so a
spring-forward day yields 23 points and an autumn-back day 25, BY CONSTRUCTION.
An implementation that walked 24 local hours would emit 24 on both and misprice
an hour twice a year; `test_static_tariff.py` drives both 2026 transitions and
counts, rather than asserting 24.

⚠️ A WINDOW BELONGS TO THE DAY THE CLOCK READS, NOT THE DAY IT STARTED. A
`22:00-07:00 mon-fri` night window covers Friday 23:00 (Friday) but NOT
Saturday 03:00 - Saturday 03:00 is Saturday's, and Saturday needs its own
period. That is the only rule under which the tiling check below can be
decidable at all, and it is why the shipped Belgian example carries an
explicit weekend row.


VALIDATION: NEVER INVENT A PRICE FOR AN UNCOVERED HOUR
======================================================
For every day type the periods must tile the 1440-minute clock exactly once -
no gap, no overlap - and `parse_static_profile()` refuses anything else with the
uncovered or doubled range named in local time. `validate.py` calls it at
startup, so a malformed profile is a refusal to start, not a surprise at 02:00.

The alternative - defaulting an uncovered hour to some price - is the defect
price.py's whole `_NoData` type exists to prevent, one level down: an invented
number that arbitrage then acts on. A missing profile is not a zero price, and
a zero price is exactly the value that makes everything look free.
"""

import json
import logging
import math
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone

from .price import (
    MAX_PLAUSIBLE_EUR_KWH, Absence, PricePoint, PriceProduct, PriceRead,
    Uncertainty, _Outcome, brussels, load_prices, price_option,
    provider_configured, required_spread, uncertainty_of, widen_factor,
)

_LOG = logging.getLogger("goodwe.tariff.static")

DAY_NAMES = ("mon", "tue", "wed", "thu", "fri", "sat", "sun")
_DAY_INDEX = {n: i for i, n in enumerate(DAY_NAMES)}      # Mon=0, as weekday()
_DAY_ALIASES = {
    "weekday": "mon-fri", "weekdays": "mon-fri", "week": "mon-fri",
    "weekend": "sat-sun", "weekends": "sat-sun",
    "all": "mon-sun", "daily": "mon-sun", "every": "mon-sun", "*": "mon-sun",
}

MINUTES_PER_DAY = 1440

# The mean that `cheap_now()` grades against is taken over this many local days.
# Seven, so each weekday appears exactly once - see the module docstring. It is
# a MODULE CONSTANT and not an option on purpose: CFG-03 already tracks three
# places option defaults live, and this describes the Belgian week rather than
# the site.
MEAN_WINDOW_DAYS = 7

# The only timezone a `static_profile` may name. See the docstring: there is no
# tz database in the shipped image, so anything else cannot be honoured and is
# refused rather than silently treated as Brussels.
SUPPORTED_TZ = ("europe/brussels", "cet", "cest", "")


class ProfileInvalid(Exception):
    """A `static_profile` that cannot be turned into a tariff.

    Refused at startup by validate.py. Carries a message naming the key and the
    fix, because the operator who typed it is the only one who can repair it.
    """


# --------------------------------------------------------------------------- #
# Parsing
# --------------------------------------------------------------------------- #

def _minutes(value, field: str) -> int:
    """"HH:MM" -> minutes past local midnight. `to: "24:00"` means end-of-day."""
    text = str(value).strip()
    if not text:
        raise ProfileInvalid(f"{field} is empty; write it as \"HH:MM\", e.g. "
                             f"\"07:00\"")
    parts = text.split(":")
    if len(parts) != 2:
        raise ProfileInvalid(
            f"{field} is {value!r}; it must be \"HH:MM\" local Belgian time, "
            f"e.g. \"22:00\". A bare hour or a decimal is refused rather than "
            f"guessed at - 7.5 could be 07:30 or 07:50.")
    try:
        h, m = int(parts[0]), int(parts[1])
    except ValueError:
        raise ProfileInvalid(f"{field} is {value!r}; \"HH:MM\" needs two "
                             f"integers, e.g. \"07:00\"")
    if not (0 <= m < 60) or not (0 <= h <= 24) or (h == 24 and m != 0):
        raise ProfileInvalid(
            f"{field} is {value!r}, which is not a time of day. Hours run 0-23, "
            f"minutes 0-59; \"24:00\" is accepted only as an END meaning "
            f"midnight at the end of the day.")
    return h * 60 + m


def _days(value, field: str) -> frozenset:
    """A day selector -> a frozenset of weekday() indices. Mon=0 .. Sun=6.

    Accepts a list (`["mon", "tue"]`), a comma list (`"sat,sun"`), a range
    (`"mon-fri"`), the aliases in `_DAY_ALIASES`, and any mixture.
    """
    if value is None:
        raise ProfileInvalid(
            f"{field} is missing. Name the days this period applies to - "
            f"\"mon-fri\", \"sat,sun\" or \"all\". It is required rather than "
            f"defaulted to every day, because a Belgian night tariff that "
            f"silently covered the weekend too would misprice 48 h a week with "
            f"nothing reporting it.")
    tokens = []
    if isinstance(value, (list, tuple)):
        for item in value:
            tokens.extend(str(item).strip().lower().split(","))
    else:
        tokens.extend(str(value).strip().lower().split(","))

    out = set()
    for token in tokens:
        token = _DAY_ALIASES.get(token.strip(), token.strip())
        if not token:
            continue
        if "-" in token:
            a, _, b = token.partition("-")
            a, b = a.strip(), b.strip()
            if a not in _DAY_INDEX or b not in _DAY_INDEX:
                raise ProfileInvalid(
                    f"{field}: {token!r} is not a day range. Days are "
                    f"{', '.join(DAY_NAMES)}; a range is written \"mon-fri\".")
            i, j = _DAY_INDEX[a], _DAY_INDEX[b]
            # A range may wrap (`fri-mon`), because a Belgian weekend tariff is
            # often written that way.
            out.update((i + k) % 7 for k in range((j - i) % 7 + 1))
        else:
            if token not in _DAY_INDEX:
                raise ProfileInvalid(
                    f"{field}: {token!r} is not a day. Use "
                    f"{', '.join(DAY_NAMES)}, a range like \"mon-fri\", or one "
                    f"of {', '.join(sorted(_DAY_ALIASES))}.")
            out.add(_DAY_INDEX[token])
    if not out:
        raise ProfileInvalid(f"{field} selected no days at all")
    return frozenset(out)


def _price(value, field: str) -> float:
    """The all-in EUR/kWh for a period, bounded exactly as price.py bounds a feed."""
    if value is None:
        raise ProfileInvalid(
            f"{field} is missing. Every period needs an all-in eur_per_kwh "
            f"(energy + supplier markup + taxes, §9.3) - the same basis the "
            f"dynamic providers are assembled onto, so the two are comparable.")
    if isinstance(value, bool) or not isinstance(value, (int, float, str)):
        raise ProfileInvalid(f"{field} is {value!r}, which is not a number")
    try:
        out = float(value)
    except (TypeError, ValueError):
        raise ProfileInvalid(f"{field} is {value!r}, which is not a number")
    if not math.isfinite(out):
        raise ProfileInvalid(f"{field} is {value!r}, which is not finite")
    if abs(out) > MAX_PLAUSIBLE_EUR_KWH:
        raise ProfileInvalid(
            f"{field} is {out} EUR/kWh, beyond +-{MAX_PLAUSIBLE_EUR_KWH}. The "
            f"usual cause is a figure in cents or in EUR/MWh - 34 instead of "
            f"0.34 - which would make every hour look like an emergency. Same "
            f"bound the fetched feeds are held to (price.py `_validate_raw`).")
    return out


@dataclass(frozen=True)
class Period:
    """One priced window of the local week. Half-open [start, end) on the clock.

    `end <= start` means the window wraps past local midnight (22:00-07:00), and
    `days` selects the day THE CLOCK READS, not the day the window opened - see
    the module docstring.
    """

    name: str
    start_min: int
    end_min: int
    days: frozenset
    eur_per_kwh: float

    @property
    def wraps(self) -> bool:
        return self.end_min <= self.start_min

    def covers(self, weekday: int, minute: int) -> bool:
        if weekday not in self.days:
            return False
        if not self.wraps:
            return self.start_min <= minute < self.end_min
        return minute >= self.start_min or minute < self.end_min

    @property
    def words(self) -> str:
        def hhmm(m):
            return f"{m // 60:02d}:{m % 60:02d}"
        days = ",".join(DAY_NAMES[d] for d in sorted(self.days))
        return (f"{self.name} {hhmm(self.start_min)}-"
                f"{hhmm(self.end_min if self.end_min else MINUTES_PER_DAY)} "
                f"{days} @ {self.eur_per_kwh:.4f}")


class StaticProfile:
    """A validated two-or-more-period tariff. Immutable, pure, I/O-free."""

    def __init__(self, periods):
        self.periods = tuple(periods)
        if not self.periods:
            raise ProfileInvalid(
                "static_profile has no periods. A profile with no periods is "
                "not a fixed tariff, it is an absent one - leave the option "
                "empty to say that on purpose (§11.4 then applies).")
        _check_tiling(self.periods)

    # -- what kind of tariff this is, DERIVED --------------------------------
    @property
    def prices(self) -> tuple:
        return tuple(p.eur_per_kwh for p in self.periods)

    @property
    def product(self) -> PriceProduct:
        """§9.2's product, derived from the DATA rather than declared.

        One distinct price across every period is a fully-fixed tariff whatever
        the operator called the rows; two or more is a static profile. Nobody
        has to tick a "this is fixed" box and nobody can tick it wrongly.
        """
        return (PriceProduct.FIXED if len(set(self.prices)) == 1
                else PriceProduct.STATIC)

    # -- lookups -------------------------------------------------------------
    def at(self, instant: datetime) -> Period:
        """The period covering an INSTANT. Tiling makes this total."""
        local = brussels(instant)
        minute = local.hour * 60 + local.minute
        for p in self.periods:
            if p.covers(local.weekday(), minute):
                return p
        # Unreachable given _check_tiling, and it stays an exception rather than
        # a default for the reason the module docstring gives.
        raise Absence(
            f"no period covers {local.isoformat()} - the tiling check that "
            f"makes this impossible has been weakened")

    def price_at(self, instant: datetime) -> float:
        return self.at(instant).eur_per_kwh

    def next_change(self, instant: datetime, *, limit_days: int = 8):
        """The next instant at which the period NAME changes, or None.

        Coarse 15-minute scan to bracket the change, then a 1-minute refine, so
        this is ~800 lookups rather than ~11,500. Returns UTC.
        """
        u = instant.astimezone(timezone.utc).replace(second=0, microsecond=0)
        # ⚠️ KEYED ON (name, price), NOT ON THE NAME. Two rows may legitimately
        # share a name - a Belgian night tariff is often written as a weekday
        # row and a weekend row both called "night" - and the tariff has not
        # changed when the clock crosses between them. But two rows with one
        # name and DIFFERENT prices is a change an operator must see, and a
        # name-only comparison would report the next change as never.
        here = self._key(u)
        coarse = timedelta(minutes=15)
        end = u + timedelta(days=limit_days)
        probe = u
        while probe < end:
            probe += coarse
            if self._key(probe) != here:
                fine = probe - coarse
                while fine < probe:
                    fine += timedelta(minutes=1)
                    if self._key(fine) != here:
                        return fine
                return probe
        return None

    def _key(self, instant):
        p = self.at(instant)
        return (p.name, p.eur_per_kwh)

    # -- the canonical series ------------------------------------------------
    def series(self, since: datetime, until: datetime, *,
               provisional: bool) -> list:
        """Hourly `PricePoint`s over [since, until), in §9.3's canonical shape.

        ⚠️ ITERATES IN UTC. That is what makes a spring-forward day 23 points and
        an autumn-back day 25 without anything here knowing the dates.
        """
        a = since.astimezone(timezone.utc).replace(minute=0, second=0,
                                                   microsecond=0)
        b = until.astimezone(timezone.utc)
        out = []
        u = a
        while u < b:
            out.append(PricePoint(start=brussels(u),
                                  eur_per_kwh=round(self.price_at(u), 6),
                                  provisional=bool(provisional)))
            u += timedelta(hours=1)
        if not out:
            raise ProfileInvalid(
                f"static_profile.series({since.isoformat()}, "
                f"{until.isoformat()}) spans no whole hour. An empty series is "
                f"not constructible downstream on purpose (price.py trap 3).")
        return out

    @property
    def words(self) -> str:
        kind = ("fully fixed" if self.product is PriceProduct.FIXED
                else f"static, {len(set(self.prices))} price levels")
        return f"{kind}: " + "; ".join(p.words for p in self.periods)


def _check_tiling(periods) -> None:
    """Every day type must be covered by EXACTLY ONE period, every minute.

    ⚠️ REFUSES BOTH DIRECTIONS. A gap means an hour with no price, and the only
    way to serve it is to invent one. An overlap means two prices for one hour
    and a silent winner decided by list order - the "two writers, one of which
    quietly wins" shape validate.py exists to refuse.
    """
    for weekday in range(7):
        gap_from = over_from = None
        for minute in range(MINUTES_PER_DAY):
            n = sum(1 for p in periods if p.covers(weekday, minute))
            if n == 0 and gap_from is None:
                gap_from = minute
            if n > 1 and over_from is None:
                over_from = (minute, n)
            if n != 0 and gap_from is not None:
                _raise_gap(weekday, gap_from, minute)
            if n <= 1 and over_from is not None:
                _raise_overlap(weekday, over_from, minute, periods)
        if gap_from is not None:
            _raise_gap(weekday, gap_from, MINUTES_PER_DAY)
        if over_from is not None:
            _raise_overlap(weekday, over_from, MINUTES_PER_DAY, periods)


def _hhmm(m):
    return f"{m // 60:02d}:{m % 60:02d}"


def _raise_gap(weekday, start, end):
    raise ProfileInvalid(
        f"static_profile leaves {DAY_NAMES[weekday]} "
        f"{_hhmm(start)}-{_hhmm(end)} with no period, so those hours have no "
        f"price. Refused rather than defaulted: an invented price is one an "
        f"optimiser would trade on, and §11.4 already says what a missing price "
        f"means. Add a period covering it - remember a window written "
        f"\"22:00-07:00 mon-fri\" does NOT cover Saturday morning.")


def _raise_overlap(weekday, over, end, periods):
    start, n = over
    names = [p.name for p in periods if p.covers(weekday, start)]
    raise ProfileInvalid(
        f"static_profile gives {DAY_NAMES[weekday]} {_hhmm(start)}-"
        f"{_hhmm(end)} {n} prices at once ({', '.join(names)}). Two prices for "
        f"one hour means one of them silently wins on list order. Narrow one of "
        f"the windows or split the day selector.")


def parse_static_profile(value):
    """`static_profile` -> a `StaticProfile`, or None when unset.

    Accepts what the option can actually arrive as:
      * `None` / `""` / `"null"`      - unset. §14 ships `static_profile: null`.
      * a JSON string                 - what config.yaml's flat `str?` option
                                        carries on a Supervisor install.
      * a dict `{"periods": [...]}`   - already-parsed, from /data/options.json
                                        under a future nested schema, from
                                        sim/inject.py, or from a check.
      * a bare list of period dicts   - the same thing without the wrapper.

    Raises `ProfileInvalid`; never returns a half-built profile.
    """
    if value is None:
        return None
    if isinstance(value, str):
        text = value.strip()
        if not text or text.lower() in ("null", "none", "off"):
            return None
        try:
            value = json.loads(text)
        except ValueError as e:
            raise ProfileInvalid(
                f"static_profile is not valid JSON ({e}). It is a JSON object "
                f"with a \"periods\" list, on one line, e.g.\n"
                f'  {{"periods": [{{"name":"day","from":"07:00","to":"22:00",'
                f'"days":"mon-fri","eur_per_kwh":0.34}}, ...]}}\n'
                f"Leave it empty for a dynamic-only or price-free site.")

    if isinstance(value, dict):
        tz = str(value.get("timezone", "") or "").strip().lower()
        if tz not in SUPPORTED_TZ:
            raise ProfileInvalid(
                f"static_profile.timezone is {value.get('timezone')!r}. Only "
                f"Europe/Brussels is supported: the shipped image carries no tz "
                f"database (see app/price.py), so any other zone could not be "
                f"honoured and would be silently treated as Brussels - which is "
                f"wrong by an hour for part of the year.")
        rows = value.get("periods")
        if rows is None:
            raise ProfileInvalid(
                "static_profile has no \"periods\" list. Even a fully-fixed "
                "tariff is written as one period covering 00:00-24:00 all days "
                "- one shape for both products, so nothing downstream branches "
                "on which kind of site this is.")
    else:
        rows = value

    if not isinstance(rows, (list, tuple)):
        raise ProfileInvalid(
            f"static_profile.periods is {type(rows).__name__}, not a list")

    periods = []
    for i, row in enumerate(rows, start=1):
        if not isinstance(row, dict):
            raise ProfileInvalid(
                f"static_profile.periods[{i}] is {type(row).__name__}, not an "
                f"object with from/to/days/eur_per_kwh")
        where = f"static_profile.periods[{i}]"
        name = str(row.get("name", "") or f"period{i}").strip() or f"period{i}"
        start = _minutes(row.get("from", row.get("start")), f"{where}.from")
        end = _minutes(row.get("to", row.get("end")), f"{where}.to")
        if end == 0:
            end = MINUTES_PER_DAY          # "00:00" as an END is end-of-day
        if start == end:
            raise ProfileInvalid(
                f"{where} runs {_hhmm(start)}-{_hhmm(end)}, which is either "
                f"zero minutes or the whole day and there is no way to tell "
                f"which was meant. Write a full day as 00:00-24:00.")
        if start == MINUTES_PER_DAY:
            raise ProfileInvalid(
                f"{where}.from is \"24:00\"; a period cannot START at "
                f"end-of-day. Did you mean \"00:00\"?")
        periods.append(Period(name=name, start_min=start,
                              end_min=end,
                              days=_days(row.get("days"), f"{where}.days"),
                              eur_per_kwh=_price(row.get("eur_per_kwh"),
                                                 f"{where}.eur_per_kwh")))
    return StaticProfile(periods)


# --------------------------------------------------------------------------- #
# The arbitrage verdict. A variant type, because a boolean is a suggestion.
# --------------------------------------------------------------------------- #

class Arbitrage(_Outcome):
    """May an optimiser trade on this tariff, and with what numbers.

      * `allowed` - carries `.spread_eur_kwh`, `.cheapest`, `.dearest`,
                    `.regime`, `.product`. These are the ONLY inputs an
                    arbitrage strategy needs, and they exist here and nowhere
                    else.
      * `refused` - carries `.reason`. Nothing else. `.spread_eur_kwh` on a
                    refusal raises `Absence`.

    ⚠️ NO `__bool__`, for price.py's reason: `if verdict:` must not look like a
    guard while being none. Use `.match(allowed=..., refused=...)`, both
    handlers required, so the refusal cannot be fallen through.
    """

    def match(self, *, allowed, refused):
        return {"allowed": allowed, "refused": refused}[self.kind](self)

    @staticmethod
    def allow(spread, cheapest, dearest, regime, product, required, sources):
        return _Allowed(spread, cheapest, dearest, regime, product, required,
                        sources)

    @staticmethod
    def refuse(reason):
        if not reason or not isinstance(reason, str):
            raise ValueError(
                "Arbitrage.refuse() takes a reason a human can act on - §10.2 "
                "requires every plan change to carry one, and 'arbitrage did "
                "nothing' with no reason is indistinguishable from a bug.")
        return _Refused(reason)


# ⚠️ FIRM BY CONSTRUCTION, ASSERTED RATHER THAN TRUSTED (FCST-03). §8.3 forbids
# an irreversible action on a provisional price, and the two points on the
# permitting variant ARE the action: they are the hours a strategy buys and
# sells in. A provisional point reaching this far is a defect in the refusal
# ladder above, and it fails loudly at the moment of construction rather than
# quietly at the meter a day later.
def _firm_or_die(cheapest, dearest):
    for role, pt in (("cheapest", cheapest), ("dearest", dearest)):
        if getattr(pt, "provisional", True):
            raise ValueError(
                f"the {role} hour handed to an arbitrage strategy is "
                f"PROVISIONAL ({getattr(pt, 'start', '?')}). §8.3: no "
                f"irreversible action may be taken on a provisional price - "
                f"an ALLOWED verdict cannot be built out of one.")


class _Allowed(Arbitrage):
    kind = "allowed"

    def __init__(self, spread, cheapest, dearest, regime, product, required,
                 sources):
        _firm_or_die(cheapest, dearest)
        self.spread_eur_kwh = spread
        self.cheapest = cheapest          # the PricePoint at the low end
        self.dearest = dearest            # the PricePoint at the high end
        self.regime = regime              # §8.3 "wide" | "normal"
        self.product = product
        # FCST-03. The threshold this spread actually cleared, and why it was
        # the size it was. An optimiser that wants to know how much margin it
        # has does not get to re-derive it from a product enum.
        self.required_spread_eur_kwh = required
        self.uncertainty = frozenset(sources)
        self.words = (f"spread {spread:.4f} EUR/kWh over "
                      f"{product.value} ({regime} thresholds, cleared "
                      f"{required:.4f} EUR/kWh)")




class _Refused(Arbitrage):
    kind = "refused"

    def __init__(self, reason):
        self.reason = reason
        self.words = f"no arbitrage - {reason}"


# --------------------------------------------------------------------------- #
# The plan: one tariff, whatever produced it
# --------------------------------------------------------------------------- #

class TariffPlan:
    """What this site's tariff looks like right now, and what may be done on it.

    `origin` is how the series was obtained, and it is NOT the same question as
    `product`:

        "provider"  a live feed (§9.2 rows 1-2)
        "static"    the CONFIGURED static_profile (§9.2 rows 3-4)
        "fallback"  the static_profile standing in for a dead feed (§11.4) -
                    identical data, `provisional=True`, arbitrage refused
        "none"      no feed and no profile. §6: the loop keeps regulating.
    """

    def __init__(self, read: PriceRead, profile, origin: str):
        self.read = read
        self.profile = profile
        self.origin = origin

    # -- derived facts -------------------------------------------------------
    @property
    def from_feed(self) -> bool:
        """Is this series an OBSERVATION of the configured provider's feed?

        TARIFF-12. `origin` names four ways a plan is obtained; this collapses
        them to the one question a caller holding a PREVIOUS plan has to ask
        about a new one - did this resolution see the feed, or is it standing
        in for one that did not? Only `"provider"` saw it. `"fallback"` and
        `"none"` are the two §11.4 rows, and `"static"` is a site with no
        feed to observe at all, where the question does not arise.

        ⚠️ IT IS NOT `has_prices`, AND THE DIFFERENCE IS THE WHOLE POINT. A
        §11.4 fallback HAS prices - the operator's own static profile - but
        every point `_degrade()` builds carries `provisional=True`, so
        `firm_points()` is empty, `arbitrage()` refuses and `cheap_now()` is
        False. It is a real tariff and it is not an observation, and a caller
        deciding whether to REPLACE an observation with it needs the second
        fact, not the first.

        ⚠️ AND IT IS A PROPERTY HERE RATHER THAN AN `origin == "provider"` AT
        THE CALL SITE, for `main.fresh_tariff_plan()`'s stated reason: a check
        has to be able to assert on the shipped expression rather than on a
        transcription that drifts, and `origin`'s vocabulary belongs to this
        module.
        """
        return self.origin == "provider"

    @property
    def has_prices(self) -> bool:
        return self.read.match(series=lambda _r: True, no_data=lambda _r: False)

    @property
    def product(self):
        return self.read.match(series=lambda r: r.product,
                               no_data=lambda _r: None)

    @property
    def provisional(self) -> bool:
        """§8.3/§11.4, derived from the points rather than asserted."""
        return self.read.match(
            series=lambda r: any(p.provisional for p in r.points),
            no_data=lambda _r: True)

    def price_at(self, instant: datetime) -> float:
        """EUR/kWh now. Raises `Absence` when there is no price - never 0.0."""
        return self.read.match(
            series=lambda r: r.at(instant).eur_per_kwh,
            no_data=lambda r: _absent(r.reason))

    def window_at(self, instant: datetime) -> str:
        """The name of the tariff window now (`day`, `night`, ...).

        Only a static profile has named windows; a day-ahead series does not,
        and inventing one would be a label with no referent.
        """
        if self.profile is None:
            raise Absence(
                "this plan has no static_profile, so it has no named tariff "
                "window. Time-window shifting (§9.2) is a static-profile "
                "behaviour; a day-ahead series is shifted against prices, not "
                "against window names.")
        return self.profile.at(instant).name

    def mean_eur_kwh(self) -> float:
        return self.read.match(
            series=lambda r: sum(p.eur_per_kwh for p in r.points) / len(r.points),
            no_data=lambda r: _absent(r.reason))

    def spread_eur_kwh(self) -> float:
        return self.read.match(
            series=lambda r: (max(p.eur_per_kwh for p in r.points)
                              - min(p.eur_per_kwh for p in r.points)),
            no_data=lambda r: _absent(r.reason))

    # -- the two things a consumer is allowed to act on ----------------------
    def firm_points(self) -> tuple:
        """The points that carry a PUBLISHED price for their own hour (§8.3).

        The complement of the carried-forward fill FCST-03 adds. Empty on a
        §11.4 fallback plan and on a `no_data` read - and empty is honest here,
        because it is a count of observations, not a value invented in place of
        one. Never used to answer "what does this hour cost"; only to answer
        "may something irreversible rest on it".
        """
        return self.read.match(
            series=lambda r: tuple(p for p in r.points if not p.provisional),
            no_data=lambda _r: ())

    def cheap_now(self, instant: datetime) -> bool:
        """Is now a cheap window? STRICTLY below the mean - see the docstring.

        ⚠️ FALSE ON A PROVISIONAL HOUR, AND IT IS THIS HOUR THAT DECIDES
        (FCST-03). §8.3 forbids an irreversible action on a provisional price
        and a paid grid top-up of a battery is about as irreversible as this
        add-on gets - but the price that would be paid is THIS hour's, so a
        firm hour does not stop being cheap because tomorrow's hours have not
        been published yet. Before FCST-03 the whole plan had to be firm, which
        was the same answer for the only two series that existed then (all firm,
        or all provisional) and would have switched the cheap window off every
        morning once carry-forward filling landed.

        The mean is taken over the FIRM hours for the same reason: yesterday's
        carried values are not observations of what today costs, and a
        comparison basis half-built out of them grades this hour against a
        number nobody published. On an all-firm series - every series that
        existed before FCST-03 - it is the same mean to the last digit.

        False when there is no price at all, which is the §11.4 row and also
        main.py's long-standing "fixed-tariff site: never force a paid grid
        top-up, wait for sun".
        """
        firm = self.firm_points()
        if not firm:
            return False
        try:
            here = self.read.match(series=lambda r: r.at(instant),
                                   no_data=lambda r: _absent(r.reason))
        except Absence:
            # An instant outside the series is an hour with no price (§11.4 for
            # that hour), not a cheap one.
            return False
        if here.provisional:
            return False
        return here.eur_per_kwh < (sum(p.eur_per_kwh for p in firm) / len(firm))

    def arbitrage(self, *, peak_guard_active: bool, min_spread_eur_kwh,
                  uncertainty) -> Arbitrage:
        """May OPT-04 trade, and on what numbers. See the module docstring.

        ⚠️ ALL THREE ARGUMENTS ARE REQUIRED AND KEYWORD-ONLY, for one reason
        each, and none of them has a default:

          `peak_guard_active`   §9.1/§10.1 put peak shaving above arbitrage,
                                and this is how that ordering survives a
                                strategy that has not been written yet: the
                                spread is not obtainable without answering the
                                question.
          `min_spread_eur_kwh`  the caller's own round-trip cost threshold
                                (§10.1's `cycle_cost_threshold_eur`). A verdict
                                that says "allowed" without being told what
                                would make a cycle worth doing is a verdict
                                about arithmetic, not about money. Pass 0.0 to
                                mean "any positive spread", explicitly.
          `uncertainty`         the §11.4 doubts the CALLER can see and this
                                object cannot - today that is
                                `Uncertainty.NO_WEATHER`. Defaulting it to the
                                empty set would be this module asserting the
                                weather forecast is healthy, which is a fact it
                                has no way to observe. The product's own
                                real-time uncertainty (§9.2 row 2) is derived
                                here and unioned in, so AC3 cannot be forgotten
                                by a caller who passes an empty set.
        """
        if not isinstance(peak_guard_active, bool):
            raise TypeError(
                f"peak_guard_active must be a bool, not {peak_guard_active!r}. "
                f"A truthy 'unknown' is how §10.1's priority 2 gets skipped.")
        if isinstance(min_spread_eur_kwh, bool) or                 not isinstance(min_spread_eur_kwh, (int, float)) or                 not math.isfinite(min_spread_eur_kwh) or                 min_spread_eur_kwh < 0:
            raise TypeError(
                f"min_spread_eur_kwh must be a finite, non-negative number of "
                f"EUR/kWh, not {min_spread_eur_kwh!r}. A negative threshold is "
                f"one every spread clears.")
        caller_sources = frozenset(uncertainty)
        if any(not isinstance(u, Uncertainty) for u in caller_sources):
            raise TypeError(
                f"uncertainty must be Uncertainty members, not "
                f"{sorted(map(repr, caller_sources))} - a string widens by "
                f"nothing while reporting that it widened.")

        if peak_guard_active:
            return Arbitrage.refuse(
                "the capacity-tariff peak guard is holding, and §10.1 puts peak "
                "shaving (priority 2) above price arbitrage (priority 3) - the "
                "capaciteitstarief is a network cost charged whatever the "
                "energy product is (§9.1)")
        if not self.has_prices:
            return Arbitrage.refuse(
                f"§11.4: {self.read.reason}. No series exists, so no spread "
                f"exists; self-consumption, peak shaving and reserve management "
                f"(§10.1 priorities 1, 2 and 4) are unaffected")

        # ⚠️ REFUSAL 3, REWRITTEN BY FCST-03 FROM "ANY POINT IS PROVISIONAL" TO
        # "THE FIRM POINTS ARE NOT ENOUGH TO TRADE ON", and the difference is
        # the whole ticket. §8.3 fills the hours a day-ahead feed has not
        # published yet with yesterday's profile and flags them - so from
        # midnight until the 14-16 h CET publication, a normal healthy site has
        # a series that is PART firm and PART carried. Refusing on "any point
        # provisional" would switch arbitrage off every morning of every day on
        # every dynamic site, which is not what §8.3 says and is a behaviour
        # nothing would ever report. What §8.3 forbids is acting ON a
        # provisional price - so the provisional hours are DROPPED from the
        # trade set and everything below is computed over the firm ones only.
        # A fully provisional series (the §11.4 fallback, or a pre-publication
        # day with no fetch at all) has no firm pair left and is refused, which
        # is the previous behaviour arrived at by a truer route.
        pts = self.read.points
        firm = tuple(p for p in pts if not p.provisional)
        if len(firm) < 2:
            return Arbitrage.refuse(
                f"§8.3: only {len(firm)} of {len(pts)} hours carry a published "
                f"price - the rest are provisional (origin {self.origin!r}), "
                f"and no irreversible action may be taken on a provisional "
                f"price. A buy hour and a sell hour must BOTH be firm before a "
                f"cycle can be committed. Load shifting, self-consumption and "
                f"peak shaving continue")

        spread = round(max(p.eur_per_kwh for p in firm)
                       - min(p.eur_per_kwh for p in firm), 6)
        if spread == 0.0:
            return Arbitrage.refuse(
                f"the tariff is flat - every one of the {len(firm)} firm hours "
                f"costs {firm[0].eur_per_kwh:.4f} EUR/kWh, so buying low and "
                f"selling high is arithmetically impossible. §9.2: a fully "
                f"fixed product is self-consumption only")

        # REFUSAL 5, NEW IN FCST-03 (AC3). One widening path, driven by the set
        # of active uncertainty sources: the product's own (§9.2 row 2, derived
        # from the DECLARED product and never from a plugin id) unioned with
        # whatever the caller can see (§11.4's no-weather row).
        sources = uncertainty_of(self.read.product) | caller_sources
        need = required_spread(min_spread_eur_kwh, sources)
        if spread < need:
            return Arbitrage.refuse(
                f"the firm spread is {spread:.4f} EUR/kWh and this cycle needs "
                f"{need:.4f} - {min_spread_eur_kwh:.4f} widened by "
                f"{widen_factor(sources):.2f}x because "
                f"{', '.join(sorted(u.value for u in sources)) or 'nothing'} "
                f"is uncertain (§8.3, §9.2, §11.4). A cycle whose spread does "
                f"not cover its own round-trip cost costs money and pack life")

        return Arbitrage.allow(
            spread=spread,
            cheapest=min(firm, key=lambda p: p.eur_per_kwh),
            dearest=max(firm, key=lambda p: p.eur_per_kwh),
            regime=self.read.regime,
            product=self.read.product,
            required=need,
            sources=sources)

    @property
    def words(self) -> str:
        if not self.has_prices:
            return f"no tariff - {self.read.reason}"
        tag = {"provider": self.read.provider, "static": "static_profile",
               "fallback": "static_profile (feed down)"}.get(self.origin,
                                                             self.origin)
        return (f"{self.product.value} from {tag}, {len(self.read.points)} h"
                + (", provisional" if self.provisional else ""))


def _absent(reason):
    raise Absence(
        f"this plan observed no prices ({reason}). A number invented here is "
        f"the bug price.py's _NoData exists to prevent - 'no data' becoming "
        f"'free'. Ask has_prices, or use read.match(...).")


# --------------------------------------------------------------------------- #
# The two entry points
# --------------------------------------------------------------------------- #

def static_plan(opts: dict, since: datetime, until: datetime, *,
                provisional: bool = False, now=None) -> TariffPlan:
    """The static-profile plan. PURE, I/O-FREE, no network, no tz database.

    This is what main.py calls on the planning tick. It is synchronous on
    purpose: `load_prices()` may perform an HTTP fetch, and a fetch on the tick
    that feeds the ESP32 heartbeat is a supplier's TLS timeout able to stall the
    heartbeat towards layer 1's 30 s veto. Generation is pure, so §2.3 says
    regenerate every tick rather than cache, and there is nothing here that can
    block.
    """
    profile = parse_static_profile(price_option(opts, "static_profile"))
    if profile is None:
        return TariffPlan(
            PriceRead.no_data(
                "no static_profile configured; §11.4: plan provisional, no "
                "arbitrage. Self-consumption, peak shaving and reserve "
                "management are unaffected"),
            None, "none")
    points = profile.series(since, until, provisional=provisional)
    return TariffPlan(
        PriceRead.series(points, profile.product, "static_profile",
                         now or datetime.now(timezone.utc)),
        profile, "fallback" if provisional else "static")


def static_status(opts: dict, now: datetime):
    """(cheap, words) for the status page and for main.py's cheap-window flag.

    Never raises: a `static_profile` that does not parse is refused at STARTUP
    by validate.py, so reaching this with a broken one means somebody handed the
    controller a dict the validator never saw. That is worth a warning and a
    "no cheap window", not a controller that stops regulating a battery (§6:
    absent machinery must not disable the loop).
    """
    try:
        plan = static_plan(opts, now, now + timedelta(days=MEAN_WINDOW_DAYS),
                           now=now)
    except ProfileInvalid as e:
        _LOG.warning("static_profile is unusable (%s); no cheap window. This "
                     "should have been refused at startup - see validate.py", e)
        return False, f"static_profile unusable: {e}"
    if not plan.has_prices:
        return False, "no static profile"
    cheap = plan.cheap_now(now)
    window = plan.window_at(now)
    price = plan.price_at(now)
    when = plan.profile.next_change(now)
    until = "" if when is None else f" until {brussels(when):%H:%M}"
    kind = ("fixed" if plan.product is PriceProduct.FIXED else "day/night")
    return cheap, (f"{kind}: {window}{until} at {price:.4f} EUR/kWh, "
                   f"mean {plan.mean_eur_kwh():.4f} over "
                   f"{MEAN_WINDOW_DAYS} d - "
                   f"{'CHEAP window' if cheap else 'not a cheap window'}")


async def resolve_tariff(opts: dict, since: datetime, until: datetime,
                         transport=None, now=None, *, history) -> TariffPlan:
    """The full §9.2 + §11.4 resolution. MAY PERFORM I/O - see `static_plan`.

    ⚠️ NOT ON THE 30 s TICK, and that is a deliberate boundary rather than an
    omission. Wiring a provider-backed plan into the control loop needs a
    BACKGROUND refresh (a day-ahead feed publishes once a day; polling it every
    30 s would be rude and would put a network timeout on the heartbeat path).
    That task belongs with the first strategy that actually trades - OPT-04 -
    and this function is the seam it consumes. Everything below it is exercised
    by test_static_tariff.py today.

      price_provider off  + profile  -> the CONFIGURED product (§9.2 rows 3-4)
      price_provider off  + none     -> no prices at all (§11.4, nothing to
                                        fall back to)
      provider series                -> the feed, §8.3-FILLED (§9.2 rows 1-2)
      provider no_data    + profile  -> §11.4 FALLBACK, provisional, no arbitrage
      provider no_data    + none     -> no prices at all

    ⚠️ `history` IS REQUIRED, KEYWORD-ONLY, AND HAS NO DEFAULT (FCST-04). It is
    the §8.3 carry-forward store - `price_history.PriceHistory`, or the explicit
    `price_history.NoHistory()` for a caller that keeps none. A default of
    `None` meaning "skip the fill" would give a caller who forgot the argument
    the TRUNCATED HORIZON this ticket exists to remove, silently, while the code
    read as though the fill were happening. There is nothing to forget now.

    ⚠️ AND THE FILL RUNS ONLY ON A `series` READ. A feed that published nothing
    still falls to `_degrade()` and §11.4's own row - the static profile the
    operator configured FOR that case - rather than to yesterday's feed. The two
    are indistinguishable to the optimiser (both are 100 % provisional, so
    `arbitrage()` finds fewer than two firm hours and refuses, and `cheap_now()`
    is False), so preferring yesterday would change a documented degradation for
    a difference only the status page can see. See price_history.py's closing
    note and DECISIONS/FCST-04-the-day-that-is-carried.md.

    ⚠️ THE FILL IS PER WHOLE HOUR AND ITS COVERAGE IS CONDITIONAL. Every point
    the provider published inside the window passes through BY IDENTITY - four
    of them in one hour on a quarter-hourly product like `eneco_eddy` - and a
    carried point is added only for a whole hour the feed did not print at all
    AND yesterday did. An hour neither source can price is left out, which is
    §11.4 for that hour: `plan.price_at()` raises rather than answering, and an
    hour with no price is not a free one.

    ⚠️ THE FILLED SERIES IS WINDOWED TO [since, until), AND THAT IS A CHANGE.
    `load_prices()` returns whatever the feed sent, past hours included, and
    `TariffPlan.arbitrage()` took its spread over all of it - so an hour that
    had already gone could be the cheapest hour of a cycle nothing can now buy
    in. `fill_from_yesterday()` walks the requested horizon and emits exactly
    that, so the spread is over hours that are still ahead. Both directions of
    the difference are real: the window gains the unpublished hours it was
    missing and loses the elapsed ones it should never have had.
    """
    stamp = now or datetime.now(timezone.utc)
    if not provider_configured(opts):
        return static_plan(opts, since, until, provisional=False, now=stamp)

    read = await load_prices(opts, since, until, transport)
    return read.match(
        series=lambda r: _provider_plan(opts, r, since, until, stamp, history),
        no_data=lambda r: _degrade(opts, r, since, until, stamp))


def _provider_plan(opts, read, since, until, stamp, history) -> TariffPlan:
    """§8.3: record what the feed published, carry yesterday into the gaps.

    The fill can itself return `no_data` - a feed whose points all fall outside
    the horizon, with nothing in the history to fill it from - and that is the
    §11.4 row arrived at one step later, so it degrades exactly as a failed
    fetch does rather than being special-cased here.
    """
    filled = history.carry(read, since=since, until=until, stamp=stamp)
    return filled.match(
        series=lambda r: TariffPlan(r, _quiet_profile(opts), "provider"),
        no_data=lambda r: _degrade(opts, r, since, until, stamp))


def _quiet_profile(opts):
    """The profile, for window naming, without letting a bad one break a good feed."""
    try:
        return parse_static_profile(price_option(opts, "static_profile"))
    except ProfileInvalid:
        return None


def _degrade(opts, read, since, until, stamp) -> TariffPlan:
    """§11.4: fall back to the static profile, mark provisional, no arbitrage."""
    try:
        profile = parse_static_profile(price_option(opts, "static_profile"))
    except ProfileInvalid as e:
        _LOG.warning("the price feed is down AND static_profile is unusable "
                     "(%s), so there is no tariff at all", e)
        profile = None
    if profile is None:
        _LOG.warning("§11.4: %s, and no static_profile is configured to fall "
                     "back to. The plan is provisional and arbitrage is off; "
                     "self-consumption, peak shaving and reserve management "
                     "continue (§6, §10.1).", read.reason)
        return TariffPlan(read, None, "none")
    _LOG.warning("§11.4: %s - falling back to the static profile, plan marked "
                 "PROVISIONAL, arbitrage off.", read.reason)
    points = profile.series(since, until, provisional=True)
    return TariffPlan(
        PriceRead.series(points, profile.product, "static_profile", stamp),
        profile, "fallback")