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` as a REQUIRED KEYWORD ARGUMENT. §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.

⚠️ FOUR 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. the plan is provisional            §11.4 fallback / §8.3 unknown hours
  4. the series is FLAT                 max - min == 0, so there is nothing to
                                        arbitrage. THIS is 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.

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,
    _Outcome, brussels, load_prices, price_option,
)

_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):
        return _Allowed(spread, cheapest, dearest, regime, product)

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


class _Allowed(Arbitrage):
    kind = "allowed"

    def __init__(self, spread, cheapest, dearest, regime, product):
        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
        self.words = (f"spread {spread:.4f} EUR/kWh over "
                      f"{product.value} ({regime} thresholds)")


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 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 cheap_now(self, instant: datetime) -> bool:
        """Is now a cheap window? STRICTLY below the mean - see the docstring.

        False on a provisional plan: §8.3 forbids an irreversible action on a
        provisional price, and a grid top-up of a battery is about as
        irreversible as this add-on gets. 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".
        """
        if not self.has_prices or self.provisional:
            return False
        try:
            return self.price_at(instant) < self.mean_eur_kwh()
        except Absence:
            # An instant outside the series is an hour with no price (§11.4 for
            # that hour), not a cheap one.
            return False

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

        ⚠️ `peak_guard_active` IS REQUIRED AND KEYWORD-ONLY. §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.
        """
        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 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")
        if self.provisional:
            return Arbitrage.refuse(
                f"§11.4: the plan is provisional (origin {self.origin!r}), and "
                f"§8.3 forbids an irreversible action on a provisional price. "
                f"Load shifting and peak shaving continue")
        spread = self.spread_eur_kwh()
        if spread == 0.0:
            return Arbitrage.refuse(
                f"the tariff is flat - every one of the "
                f"{len(self.read.points)} hours costs "
                f"{self.read.points[0].eur_per_kwh:.4f} EUR/kWh, so buying low "
                f"and selling high is arithmetically impossible. §9.2: a fully "
                f"fixed product is self-consumption only")
        pts = self.read.points
        return Arbitrage.allow(
            spread=spread,
            cheapest=min(pts, key=lambda p: p.eur_per_kwh),
            dearest=max(pts, key=lambda p: p.eur_per_kwh),
            regime=self.read.regime,
            product=self.read.product)

    @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) -> 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 (§9.2 rows 1-2)
      provider no_data    + profile  -> §11.4 FALLBACK, provisional, no arbitrage
      provider no_data    + none     -> no prices at all
    """
    stamp = now or datetime.now(timezone.utc)
    pid = str(price_option(opts, "price_provider") or "off").strip()
    if pid in ("", "off"):
        return static_plan(opts, since, until, provisional=False, now=stamp)

    read = await load_prices(opts, since, until, transport)
    return read.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")