goodwe-addon/goodwe_controller/app/price_forecast.py

"""FCST-03 - the hours the feed has not published yet, and what may not rest on them.

WHERE THIS LIVES. specs.md §2.2 puts price ingest under `ems/ingest/prices/`.
That 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/`. The
BEHAVIOUR §8.3 describes is unchanged and still authoritative - only the path
moved. Read `ems/...` as `app/...`.

§8.3, in full, because every line below is one clause of it:

    "Day-ahead prices arrive once daily. Until they do, unknown hours use
     yesterday's profile and are flagged `provisional: true` - and NO
     IRREVERSIBLE ACTION MAY BE TAKEN ON A PROVISIONAL PRICE. Real-time /
     imbalance products are treated as high-volatility: arbitrage thresholds
     widen."

Three separate things, and this module owns the first two. The third - the
widening arithmetic - lives in price.py beside the product enum it is derived
from (`Uncertainty`, `widen_factor`, `required_spread`), and is spent by
`static_tariff.TariffPlan.arbitrage()`, because that is where the verdict this
repo already refuses to hand out casually is produced. FCST-03 added a refusal
to that existing ladder rather than a second mechanism beside it.


THE DEFECT THIS MODULE IS SHAPED BY: A COPY THAT KEEPS THE WRONG FLAG
=====================================================================
A point filled from yesterday and a point published this morning are the SAME
SHAPE. Three floats and a bool; only the bool tells them apart. And the natural
way to write a carry-forward fill is:

    PricePoint(start=t, eur_per_kwh=y.eur_per_kwh, provisional=y.provisional)
                                                                ^^^^^^^^^^^^^

which copies yesterday's `False` onto today's guess and produces exactly the
thing the ticket exists to prevent: STALE DATA PRESENTED AS CURRENT. It reads
fine, it type-checks, it round-trips through §9.3's JSON, and every price on
the status page looks like a price. This is the meter-side SAFETY-18 defect one
layer up: the reading arrives on time, looks fresh, and is wrong - and a
freshness check that only asks "did something arrive" cannot see it.

So the flag is not passed through anywhere in this file:

  * A CARRIED POINT IS BUILT BY `_carried()`, WHICH HAS NO PARAMETER FOR THE
    FLAG. `provisional=True` is a literal in one place. There is no argument a
    caller can pass, no keyword to forget, and no variable that could hold
    yesterday's value at the moment the point is constructed.
  * A PUBLISHED POINT IS NEVER COPIED AT ALL. It is passed through BY IDENTITY -
    the object the provider produced is the object in the output series - so
    there is no copy that could lose or gain a flag. `filled.points[i] is
    fetched.points[j]` is a checkable property, and test_price_forecast.py
    checks it. EVERY published point in the window is passed through, not one
    per hour: see the sub-hourly note below.
  * `_audit()` RE-DERIVES the expected flag for every output point from the set
    of hours the feed actually published, and RAISES if any point disagrees.
    It is a post-condition, not a filter: it never fixes a flag, because a
    module that quietly corrects its own provenance is one whose provenance
    cannot be trusted.
  * and test_price_forecast.py walks THIS FILE'S AST and fails if any
    `PricePoint(...)` in it passes anything but a literal for `provisional`, so
    the pass-through idiom cannot come back in a later edit and pass review by
    looking ordinary.

⚠️ "PUBLISHED" DOES NOT MEAN "FIRM". `eneco_eddy` is an imbalance product whose
current-quarter print is genuinely provisional and says so per point (§9.2,
TARIFF-02). Such a point is published AND provisional, and the audit's rule for
a published point is "the flag is whatever the feed said", never "False". A
provenance check that forced published points to False would be a bug that
looks like a guarantee.


THE FILL, AND WHY IT KEYS ON THE WALL CLOCK
===========================================
For every whole hour in the requested horizon that the feed has not published,
take yesterday's value for THE SAME BRUSSELS WALL-CLOCK HOUR - not `t - 86400
s`. On the two days a year that matter those are different instants, and the
difference is not academic: a 23-hour day has no 02:00 at all and a 25-hour day
has two, so a fill that subtracts a fixed 24 h shifts every carried hour by one
across a transition and misprices the whole day in a way nothing reports.

On the 25-hour day the two 02:00 hours - `02:00+02:00` and `02:00+01:00` - both
key to local hour 2 and are BOTH filled from yesterday's single 02:00 value.
That is deliberate: they are both guesses, they are both flagged, and nothing
irreversible may rest on either. If yesterday itself was a 25-hour day, the
first of its two 02:00 points is the one used, deterministically.

An hour with no published price AND no yesterday value is LEFT OUT of the
series. It is not filled with a neighbour, a mean, or a zero. §11.4 applies to
that hour - `_Series.at()` raises `Absence` for it - and the caller that wanted
a tariff for the whole horizon falls back to the static profile (TARIFF-03),
which is the row the spec actually writes for "no price data". If NOTHING can
be produced, the result is `PriceRead.no_data(reason)` and not an empty series,
because an empty series is not constructible on purpose (price.py trap 3).

⚠️ A SUB-HOURLY FEED IS NOT DECIMATED TO ONE POINT AN HOUR, and the hourly walk
below is about WHICH HOURS NEED FILLING - never about what a published hour is
allowed to contain. `eneco_eddy` is the Belgian imbalance product and settles
per QUARTER, `_validate_raw()` accepts any strictly-increasing spacing, and a
published hour may therefore hold four points. So EVERY published point inside
the window is passed through, in order, and a carried point is added only for a
whole hour with NO published coverage at all.

The alternative was measured rather than reasoned about, because FCST-04's
first caller shipped it: keying the published set as `{_utc_hour(p.start): p
for p in r.points}` reads perfectly ordinary and keeps the LAST point of each
hour, dropping the other three. On a 96-point quarter-hourly feed - all firm,
fully published, nothing to carry - the series came out 15 points long, the
arbitrage spread collapsed from 0.143 to 0.014 EUR/kWh, and the cheapest
"hour" became a `:45` point whose `at()` window runs into the next hour. A feed
losing 84 % of its own prints is not a windowing effect and is not something a
provisional flag would ever have reported.


AC2: THE IRREVERSIBLE-ACTION GATE
=================================
"The distinction should be an explicit property of the action, not a judgement
call at each call site" - so an irreversible action is a value, `Irreversible`,
carrying the number of hours it COMMITS. `price_permits()` looks at exactly
those hours and answers with a `Gate` variant that carries the evidence on the
permitting arm and a reason on the blocking one, the `_Outcome` shape price.py
borrowed from `sim/riglock.py`.

The shipped example is the maintenance drain (§12): starting one commits hours
of forced discharge that cannot be undone if the price turns out to have been
yesterday's. Its commitment window is DERIVED from `maintenance.DRAIN_TIMEOUT_MIN`
rather than typed here, so the gate and the phase machine cannot drift apart.

⚠️ WHAT DOES NOT GO THROUGH THIS GATE, AND WHY THAT MATTERS MORE THAN WHAT
DOES. The maintenance cycle is a CALENDAR obligation (§12.2: a cycle that
silently never fires is indistinguishable from one that is not due yet), and a
site with `price_provider: off` - the shipped default - has no price series at
all. Routing every maintenance start through a gate that blocks on `no_data`
would stop the monthly cycle on every default install, permanently and
silently: the exact failure §12 opens by naming. So `begin_phase()` is the
entry point for a start that is being JUSTIFIED BY A PRICE, main.py's existing
calendar path is untouched by this ticket, and OPT-04/MAINT-05 own the decision
of which one a site runs. Nothing here disables anything that runs today.

Reversible actions - ordinary self-consumption modulation, the loop that
actually holds the grid at zero - are not gated at all and cannot be: §6 says
absent machinery must not disable the loop.
"""

import logging
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from enum import Enum

from .maintenance import CHARGE_TIMEOUT_MIN, DRAIN_TIMEOUT_MIN
from .price import PricePoint, PriceRead, _Outcome, brussels

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


class Provenance(Enum):
    """Where a point in a filled series came from. INTERNAL - §9.3's wire format
    carries the boolean and nothing else, so this never leaves the module.

    It exists so `_audit()` has a vocabulary to state its post-condition in, and
    so the two cases have names in the log rather than being "the ones with the
    flag" and "the others".
    """
    PUBLISHED = "published"    # the provider printed this hour. Flag: the
                               # feed's own - an imbalance print may be
                               # published AND provisional (§9.2).
    CARRIED = "carried"        # yesterday's same-wall-clock hour (§8.3).
                               # Flag: ALWAYS True, by construction.


# --------------------------------------------------------------------------- #
# The fill
# --------------------------------------------------------------------------- #

def _utc_hour(instant: datetime) -> datetime:
    u = instant.astimezone(timezone.utc)
    return u.replace(minute=0, second=0, microsecond=0)


def _published_by_hour(read: PriceRead) -> dict:
    """UTC hour -> the provider's OWN points that start inside it, in order.

    ⚠️ A TUPLE PER HOUR, NOT A POINT, AND THAT IS THE WHOLE OF THE SUB-HOURLY
    RULE. `{_utc_hour(p.start): p for p in r.points}` reads identically, is one
    character shorter, and silently keeps one point per hour - the last one
    written. The mapping answers two DIFFERENT questions here: "does this hour
    need filling at all?" is `u in published`, and "what does the output carry
    for it?" is all of `published[u]`. Collapsing them is how a quarter-hourly
    feed loses three quarters of its prints.
    """
    out: dict = {}
    read.match(
        series=lambda r: [out.setdefault(_utc_hour(p.start), []).append(p)
                          for p in r.points],
        no_data=lambda _r: [])
    return {u: tuple(ps) for u, ps in out.items()}


def _carried(start_utc: datetime, eur_per_kwh: float) -> PricePoint:
    """A point filled from yesterday's profile. §8.3: ALWAYS provisional.

    ⚠️ THERE IS NO `provisional` PARAMETER, AND THAT IS THE MECHANISM. This is
    the only constructor of a carried point, the flag is a literal, and no
    caller - including a future one editing in a hurry - has anything to pass,
    default, or copy from the point being carried.
    """
    return PricePoint(start=brussels(start_utc),
                      eur_per_kwh=round(float(eur_per_kwh), 6),
                      provisional=True)


def _audit(points, published) -> None:
    """Post-condition: every flag matches the point's TRUE provenance (AC1).

    `published` maps a UTC hour to the TUPLE of points the provider produced
    for it (`_published_by_hour()`). An hour ABSENT from that mapping is an hour
    the feed did not print, and a carried point is the only thing that may
    stand in it.

    The rule, in the two forms an output point can take:

      * PUBLISHED - the point must BE one of `published[u]`, by identity, and
        may be emitted only once. Its flag is whatever the feed said, never
        forced to False: an imbalance print is published AND provisional (§9.2,
        TARIFF-02), and a check that "corrected" it would be a bug wearing a
        guarantee.
      * CARRIED - the hour must not appear in `published` AT ALL, the flag must
        be exactly True, and that hour may hold no second point.

    ⚠️ WHAT REPLACED "AT MOST ONE POINT PER HOUR", AND WHY IT IS STRICTER. This
    audit began as a `seen` set: any repeated `_utc_hour` raised, because a
    duplicated hour is how a DST day gets mispriced. A bare hour count cannot
    express a quarter-hourly feed, where four distinct published points share
    one hour legitimately - so the count was replaced by the two rules above,
    which subsume it. A repeated CARRIED hour still raises; so does the same
    published point emitted twice; and a carried point may no longer share an
    hour with a published one, which the count never checked at all.

    ⚠️ RAISES, NEVER REPAIRS. A module that silently corrects its own
    provenance is one whose provenance means nothing; the only safe response to
    "this point's flag disagrees with where it came from" is to refuse to emit
    the series at all.
    """
    carried_hours = set()
    emitted = set()
    for p in points:
        u = _utc_hour(p.start)
        src = published.get(u)
        if src is None:
            if u in carried_hours:
                raise ValueError(
                    f"the filled series repeats the carried hour "
                    f"{u.isoformat()}; a duplicated hour is how a DST day gets "
                    f"mispriced")
            carried_hours.add(u)
            if p.provisional is not True:
                raise ValueError(
                    f"{u.isoformat()} is {Provenance.CARRIED.value} and is "
                    f"flagged provisional={p.provisional!r}. §8.3: an hour "
                    f"filled from yesterday's profile is provisional whether "
                    f"or not the number turns out to be right - the flag is a "
                    f"claim about PROVENANCE, not about accuracy.")
        else:
            if not any(p is q for q in src):
                raise ValueError(
                    f"{u.isoformat()} is {Provenance.PUBLISHED.value} but the "
                    f"series carries a COPY of that point rather than the "
                    f"point itself. A copy is where the provisional flag gets "
                    f"lost; published points pass through by identity.")
            if id(p) in emitted:
                raise ValueError(
                    f"the filled series emits the published point at "
                    f"{p.start.isoformat()} TWICE. The provider's own points "
                    f"are strictly increasing (`_validate_raw`), so a repeat "
                    f"is never a second print - it is a duplicated hour, which "
                    f"is how a DST day gets mispriced")
            emitted.add(id(p))


def fill_from_yesterday(read: PriceRead, *, yesterday: PriceRead,
                        since: datetime, until: datetime,
                        stamp: datetime) -> PriceRead:
    """§8.3: fill the hours the feed has not published with yesterday's profile.

    Every argument is required and keyword-only past the first. In particular
    `yesterday` has NO DEFAULT: a fill with no yesterday to fill from is a
    different situation from a fill that found nothing useful in one, and a
    default `None` would make the two arrive looking identical.

    Returns a `PriceRead`:

      * a series, carrying the provider's own objects - ALL of them, which on
        a quarter-hourly feed is four inside one hour - for the hours it
        published, and one `_carried()` point for each remaining whole hour,
        ordered by instant;
      * `no_data` when neither source can produce a single hour - never an
        empty series, which is not constructible (price.py trap 3).

    ⚠️ THE WINDOW IS WHOLE HOURS AT BOTH ENDS. `since` is floored to the hour
    that CONTAINS it, because the current hour is still a tradeable hour; and
    so is `until`, which is the half FCST-04's first draft left out. With
    `since` floored and `until` compared raw, a 48 h horizon asked for at 08:17
    emitted 49 hours - the hour of `until` was still `< until` - and the count
    was one too many for every `now` that is not exactly on the hour, which is
    every `now` a running controller has.

    ⚠️ IDEMPOTENT AND RE-RUNNABLE, WHICH IS THE §5.4 LATE-PUBLICATION CONTRACT.
    Day-ahead publication at 14-16 h CET can be late or fail, so the fetcher
    retries and this runs again against the fresher read. Published hours always
    win over carried ones, so a re-run REPLACES a carried point with the real
    print and the hour becomes firm - the flag and the number move together
    because the whole series is rebuilt rather than patched in place.
    """
    for name, value in (("since", since), ("until", until), ("stamp", stamp)):
        if value.tzinfo is None or value.utcoffset() is None:
            raise ValueError(f"{name} must be tz-aware; a naive instant is "
                             f"ambiguous for one hour every October")

    published = _published_by_hour(read)

    # Yesterday's profile, keyed by BRUSSELS WALL-CLOCK HOUR. First point wins
    # when yesterday itself was a 25-hour day and printed local hour 2 twice.
    by_local_hour = {}
    yesterday.match(
        series=lambda r: [by_local_hour.setdefault(brussels(p.start).hour, p)
                          for p in r.points],
        no_data=lambda _r: [])

    out = []
    u = _utc_hour(since)
    end = _utc_hour(until)
    while u < end:
        got = published.get(u)
        if got:
            out.extend(got)              # BY IDENTITY, and ALL of them. No copy,
        else:                            # and no decimation to one an hour.
            src = by_local_hour.get(brussels(u).hour)
            if src is not None:
                out.append(_carried(u, src.eur_per_kwh))
        u += timedelta(hours=1)

    if not out:
        reason = read.match(
            series=lambda r: (f"the {len(r.points)} published hours all fall "
                              f"outside {since.isoformat()}..{until.isoformat()}"),
            no_data=lambda r: r.reason)
        return PriceRead.no_data(
            f"§8.3 carry-forward produced no hours: {reason}, and yesterday's "
            f"profile has nothing to fill them with either. §11.4 applies: "
            f"static profile, plan provisional, no arbitrage")

    _audit(out, published)
    carried = sum(1 for p in out if _utc_hour(p.start) not in published)
    if carried:
        _LOG.info("§8.3: %d of %d hours carried forward from yesterday's "
                  "profile and flagged provisional; %d published",
                  carried, len(out), len(out) - carried)

    product = read.match(series=lambda r: r.product,
                         no_data=lambda _r: yesterday.product)
    provider = read.match(series=lambda r: r.provider,
                          no_data=lambda r: yesterday.provider)
    return PriceRead.series(out, product, provider, stamp)


# --------------------------------------------------------------------------- #
# AC2: irreversibility as a property of the ACTION, and the gate that reads it
# --------------------------------------------------------------------------- #

@dataclass(frozen=True)
class Irreversible:
    """An action that cannot be undone once started, and how far it commits.

    `commits_h` is the horizon the action ties up - the hours whose prices must
    therefore be firm before it may be justified by a price at all.
    """

    name: str            # the §10.2 reason code
    commits_h: float

    def __post_init__(self):
        if not self.name or not isinstance(self.name, str):
            raise ValueError("an irreversible action needs a reason code (§10.2)")
        if isinstance(self.commits_h, bool) or \
                not isinstance(self.commits_h, (int, float)) or \
                not self.commits_h > 0:
            raise ValueError(
                f"commits_h must be a positive number of hours, not "
                f"{self.commits_h!r}. An action that commits nothing is a "
                f"reversible one and does not belong here.")


# ⚠️ DERIVED FROM THE PHASE MACHINE'S OWN BOUNDS, NOT TYPED HERE. maintenance.py
# already states how long each phase may run before it is a fault; a second copy
# of those numbers here would be a place for them to drift apart, and the drift
# would show up as a gate that checked the wrong hours.
MAINTENANCE_DRAIN = Irreversible("MAINTENANCE_DRAIN", DRAIN_TIMEOUT_MIN / 60.0)
MAINTENANCE_CHARGE = Irreversible("MAINTENANCE_CHARGE", CHARGE_TIMEOUT_MIN / 60.0)


class Gate(_Outcome):
    """May an irreversible action proceed on the price evidence available?

      * `permitted` - carries `.action` and `.hours`, the firm points that were
                      actually checked. The evidence exists only on this arm.
      * `blocked`   - carries `.reason`. Nothing else; `.hours` raises `Absence`.

    No `__bool__`, both `match()` arms required and keyword-only, for the reason
    price.py's `_Outcome` gives: `if gate:` must not look like a guard while
    being none.
    """

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

    @staticmethod
    def permit(action, hours):
        return _Permitted(action, hours)

    @staticmethod
    def block(reason):
        if not reason or not isinstance(reason, str):
            raise ValueError(
                "Gate.block() takes a reason a human can act on. §6 requires "
                "every decision to explain itself, and 'the maintenance cycle "
                "did not start' with no reason is indistinguishable from a bug.")
        return _Blocked(reason)


class _Permitted(Gate):
    kind = "permitted"

    def __init__(self, action, hours):
        if not isinstance(action, Irreversible):
            raise TypeError(f"action must be an Irreversible, not {action!r}")
        pts = tuple(hours)
        if not pts:
            raise ValueError(
                "a permission granted over ZERO checked hours is a permission "
                "granted on no evidence - the shape this whole file exists to "
                "refuse. Block instead.")
        if any(p.provisional for p in pts):
            raise ValueError(
                "a permitted gate cannot carry a provisional hour: §8.3 is the "
                "one rule this type has.")
        self.action = action
        self.hours = pts
        self.words = (f"{action.name} permitted on {len(pts)} firm hours "
                      f"({pts[0].start.isoformat()} ..)")


class _Blocked(Gate):
    kind = "blocked"

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


def committed_hours(now: datetime, action: Irreversible) -> tuple:
    """The whole hours `action` started at `now` would run through.

    Split out because the gate and its caller must agree on what "the hours it
    commits" means, and because a window that quietly excluded the last hour
    would be a gate that permits an action on an hour it never looked at.
    """
    if not isinstance(action, Irreversible):
        raise TypeError(f"action must be an Irreversible, not {action!r}")
    if now.tzinfo is None or now.utcoffset() is None:
        raise ValueError("now must be tz-aware")
    out = []
    u = _utc_hour(now)
    end = now.astimezone(timezone.utc) + timedelta(hours=action.commits_h)
    while u < end:
        out.append(u)
        u += timedelta(hours=1)
    return tuple(out)


def price_permits(read: PriceRead, now: datetime,
                  action: Irreversible) -> Gate:
    """§8.3's rule, as a value: may this irreversible action rest on this price?

    Blocks when the read observed nothing (§11.4 - an action justified by a
    price nobody has is an action justified by nothing), when any hour it
    commits has no price at all, and when any of those hours is provisional.
    Names the offending hour, so an operator reading the log knows which one.
    """
    hours = committed_hours(now, action)
    return read.match(
        series=lambda r: _walk(r, hours, action),
        no_data=lambda r: Gate.block(
            f"{action.name} is gated on a price and there is no price data: "
            f"{r.reason}. §11.4 - and a missing price is not a cheap one"))


def _walk(series, hours, action) -> Gate:
    pts = []
    for u in hours:
        if not series.covers(u):
            return Gate.block(
                f"{action.name} commits {action.commits_h:g} h from "
                f"{hours[0].isoformat()} and no price covers "
                f"{u.isoformat()}. An hour outside the series is an hour with "
                f"no price (§11.4 for that hour), not a free one")
        p = series.at(u)
        if p.provisional:
            return Gate.block(
                f"{action.name} commits {action.commits_h:g} h from "
                f"{hours[0].isoformat()} and {u.isoformat()} is PROVISIONAL - "
                f"carried from yesterday's profile or standing in for a dead "
                f"feed. §8.3: no irreversible action may be taken on a "
                f"provisional price. It becomes available when the day-ahead "
                f"print lands (~14-16 h CET, §5.4)")
        pts.append(p)
    return Gate.permit(action, pts)


def begin_phase(maint, now: datetime, read: PriceRead,
                action: Irreversible) -> Gate:
    """Start a maintenance phase BECAUSE OF A PRICE, or refuse and say why.

    The AC2 example made real: `maint.force_start(now)` - the call that moves
    the phase machine out of `idle` and commits the pack to hours of forced
    discharge - happens on the permitting arm and cannot happen on the other
    one, because the blocking arm has no code path to it.

    ⚠️ READ THE MODULE DOCSTRING'S LAST WARNING BEFORE WIRING THIS IN. This is
    the price-justified entry point, not the calendar one; a site with no price
    provider must keep starting its monthly cycle the way it does today, or
    §12's opening sentence comes true - a cycle that silently never fires.
    """
    gate = price_permits(read, now, action)
    return gate.match(permitted=lambda g: _go(maint, now, g),
                      blocked=lambda g: _no(g))


def _go(maint, now, gate) -> Gate:
    maint.force_start(now)
    _LOG.info("%s: %s", gate.action.name, gate.words)
    return gate


def _no(gate) -> Gate:
    _LOG.warning("irreversible action refused - %s", gate.reason)
    return gate