goodwe-addon/goodwe_controller/test_price_forecast.py

"""Runnable check for FCST-03 - provisional prices.  `python3 test_price_forecast.py`

WHAT THIS FILE IS DEFENDING, in the order the ticket's acceptance criteria name
it:

  AC1  Every price point carries a `provisional` flag MATCHING ITS TRUE
       PROVENANCE - a point filled from yesterday's profile is provisional even
       when the number happens to be right, a point the feed published keeps
       whatever flag the feed gave it, and an unmarked carry-forward is not
       constructible rather than merely discouraged.
  AC2  An irreversible action gated on a provisional price is REJECTED, with the
       spec's own example - a maintenance-phase transition - actually refused:
       the phase machine is still in `idle` afterwards.
  AC3  Real-time / imbalance prices get MEASURABLY WIDER thresholds. Measured,
       by finding the smallest spread each regime will accept and comparing the
       two numbers - not by reading a constant back out of the module that
       defines it.

THREE RULES THIS FILE HOLDS ITSELF TO, because the reviewer will:

  * EVERY VERDICT IS DERIVED. The carried/published split is re-derived here
    from the set of hours the fixture published, by different arithmetic than
    app/price_forecast.py uses; the DST day lengths come from the offset
    difference rather than from a 23 or a 25 typed into a check; the widening
    factor is MEASURED by bisecting the accept/refuse boundary and then
    compared with the declared one.

  * NO STIMULUS IS EVALUATED WHERE THE QUANTITY IT GRADES IS DEGENERATE, and
    this ticket is full of that trap:
      - YESTERDAY AND TODAY MUST PRICE EVERY HOUR DIFFERENTLY, or a fill that
        carried the wrong day forward is invisible. Asserted first, as a check.
      - THE PROVISIONAL HOURS MUST CONTAIN THE CHEAPEST HOUR, or "arbitrage
        ignores provisional hours" grades nothing: the firm-only spread and the
        whole-series spread would be the same number. Asserted before it is
        used, and it is the realistic shape of the danger - yesterday's cheap
        hour luring a charge that today's prices do not justify.
      - THE TWO WIDENING FACTORS MUST DIFFER, or `max`, `sum` and `product` are
        indistinguishable combination rules. Asserted, then all three are
        computed and only one matches what the code does.
      - THE LOCAL-HOUR FILL AND A NAIVE `t - 24 h` FILL MUST DISAGREE on the
        transition days, or the DST check passes for both implementations.
        Asserted, then the correct one is graded.
      - PRICES ARE WRITTEN TO SIX DECIMALS with distinct 3-, 4- and 6-dp
        roundings, so an implementation that rounds differently is visible.
        (TARIFF-03's fixtures were 3-dp, where those three are one number.)

  * A FAILED READ IS NOT AN OBSERVATION. A missing price is not a zero price
    and a stale one is not a fresh one: the gate blocks on `no_data`, blocks on
    an hour the series does not cover, and blocks on a provisional hour - three
    different reasons, none of them an answer.
"""

import ast
import inspect
import os
import sys
from datetime import datetime, timedelta, timezone

from app import price as _price
from app import price_forecast as _fc
from app import static_tariff as _st
from app.maintenance import (
    CHARGE_TIMEOUT_MIN, DRAIN_TIMEOUT_MIN, DRAIN, IDLE, MaintConfig, Maintenance,
)
from app.price import (
    Absence, PricePoint, PriceProduct, PriceRead, PROVIDERS, Uncertainty,
    WIDEN_FACTOR, brussels, brussels_offset, plan_is_provisional,
    required_spread, uncertainty_of, widen_factor,
)
from app.price_forecast import (
    MAINTENANCE_CHARGE, MAINTENANCE_DRAIN, Gate, Irreversible, committed_hours,
    fill_from_yesterday, price_permits, begin_phase,
)
from app.arbitrage import ArbitrageDecision
from app.static_tariff import Arbitrage, TariffPlan

UTC = timezone.utc
HERE = os.path.dirname(os.path.abspath(__file__))
fails = []


def check(name, cond):
    print(f"  {'ok  ' if cond else 'FAIL'} {name}")
    if not cond:
        fails.append(name)


def raises(fn, exc):
    """True only if fn() raises EXACTLY `exc`. A different exception is False -
    'it was refused' and 'it broke on the way to being refused' must not grade
    the same."""
    try:
        fn()
    except exc:
        return True
    except Exception:
        return False
    return False


class _Raised:
    def __init__(self, e):
        self.e = e

    def __eq__(self, other):
        return False

    def __hash__(self):
        return id(self)

    def __bool__(self):
        return False

    def __repr__(self):
        return f"<raised {type(self.e).__name__}: {self.e}>"


def derived(fn):
    """A value a check grades. An exception becomes `_Raised`, which equals
    nothing and is falsy - NOT a default."""
    try:
        return fn()
    except Exception as e:                                    # noqa: BLE001
        return _Raised(e)


def near(a, b, tol=1e-9):
    try:
        return abs(float(a) - float(b)) <= tol
    except (TypeError, ValueError):
        return False


# --------------------------------------------------------------------------- #
# Fixtures.
#
# ⚠️ SIX-DECIMAL PRICES, DELIBERATELY. TARIFF-03's fixtures were 0.3410 /
# 0.2570, where rounding to 3, 4 and 6 decimals gives the same number - so an
# implementation that rounded differently from the one being graded survived
# every check. These are chosen so those three roundings differ.
# --------------------------------------------------------------------------- #

def y_price(h):
    """Yesterday's all-in EUR/kWh for local hour h."""
    return round(0.123437 + 0.010700 * h, 6)


def t_price(h):
    """Today's PUBLISHED all-in EUR/kWh for local hour h. Different from
    yesterday's at every hour, on purpose - see the degeneracy note above."""
    return round(0.206562 + 0.009300 * h, 6)


def local_midnight(y, m, d) -> datetime:
    """The UTC instant of Brussels local midnight on that date, derived from
    `brussels_offset` rather than from a table of dates."""
    naive = datetime(y, m, d, 0, 0)
    for hours in (1, 2):
        guess = naive.replace(tzinfo=UTC) - timedelta(hours=hours)
        if brussels_offset(guess).utcoffset(None) == timedelta(hours=hours):
            return guess
    raise AssertionError(f"no consistent Brussels offset for {naive}")


def hours_between(a, b):
    out, u = [], a
    while u < b:
        out.append(u)
        u += timedelta(hours=1)
    return out


def series_of(instants, price_of, product, provider, *, provisional):
    return PriceRead.series(
        [PricePoint(start=brussels(u), eur_per_kwh=price_of(u),
                    provisional=provisional) for u in instants],
        product, provider, instants[0])


# A plain winter day: local 2026-01-15, 24 hours, no transition in sight.
D0 = local_midnight(2026, 1, 15)
D1 = local_midnight(2026, 1, 16)
YEST0 = local_midnight(2026, 1, 14)
DAY = hours_between(D0, D1)
YDAY = hours_between(YEST0, D0)

YESTERDAY = series_of(YDAY, lambda u: y_price(brussels(u).hour),
                      PriceProduct.DAY_AHEAD, "generic_dayahead",
                      provisional=False)
# The feed has published the first 12 hours of today and nothing after - the
# ordinary state of a dynamic site before the 14-16 h CET print (§5.4).
PUBLISHED_N = 12
TODAY_PART = series_of(DAY[:PUBLISHED_N], lambda u: t_price(brussels(u).hour),
                       PriceProduct.DAY_AHEAD, "generic_dayahead",
                       provisional=False)


print()
print("0. the fixtures are not degenerate (checked before anything is graded)")

check("yesterday and today price EVERY shared hour differently, so a fill that "
      "carried the wrong day forward would be visible in the value and not "
      "only in the flag",
      all(y_price(h) != t_price(h) for h in range(24)))
_roundings = derived(lambda: {round(y_price(3), n) for n in (3, 4, 6)})
check(f"...and a fixture price rounds to three DIFFERENT numbers at 3, 4 and 6 "
      f"decimals {sorted(_roundings)}, so a divergent rounding cannot hide",
      len(_roundings) == 3)
check("...and both widening factors differ, so max / sum / product are three "
      "distinguishable answers",
      len(set(WIDEN_FACTOR.values())) == len(WIDEN_FACTOR) >= 2)


# --------------------------------------------------------------------------- #
print()
print("1. AC1 - the flag matches PROVENANCE, and an unmarked fill is not "
      "constructible")

FILLED = fill_from_yesterday(TODAY_PART, yesterday=YESTERDAY,
                             since=D0, until=D1, stamp=D0)

_pub_starts = {u for u in DAY[:PUBLISHED_N]}
_expected_prov = {u: (u not in _pub_starts) for u in DAY}
_got_prov = derived(lambda: {p.start.astimezone(UTC): p.provisional
                             for p in FILLED.points})
check(f"the filled series covers every hour of the day "
      f"{len(FILLED.points) if FILLED.kind == 'series' else FILLED}",
      FILLED.kind == "series" and len(FILLED.points) == len(DAY))
check("...and EVERY point's flag equals what its provenance says it should be, "
      "re-derived here from the set of hours the fixture published",
      _got_prov == _expected_prov)
_carried = derived(lambda: [p for p in FILLED.points if p.provisional])
check(f"...which is {len(DAY) - PUBLISHED_N} carried hours and {PUBLISHED_N} "
      f"published ones, counted rather than asserted",
      len(_carried) == len(DAY) - PUBLISHED_N)
check("...and the carried hours carry YESTERDAY's number for the same wall "
      "clock hour, not today's and not a neighbour's",
      all(near(p.eur_per_kwh, y_price(brussels(p.start).hour))
          for p in _carried))
check("...while the published hours carry TODAY's, so the two halves are not "
      "the same data wearing two flags",
      all(near(p.eur_per_kwh, t_price(brussels(p.start).hour))
          for p in FILLED.points if not p.provisional))

# ⚠️ THE COPY IS THE DEFECT. A published point that survives as a COPY is one
# whose flag could have been rewritten on the way through; identity is the only
# property that rules that out for every field at once.
check("a published point is passed through BY IDENTITY - the object the "
      "provider produced is the object in the filled series",
      all(any(p is q for q in TODAY_PART.points)
          for p in FILLED.points if not p.provisional))
check("...and the canonical §9.3 dict carries the flag out to the wire for "
      "both kinds",
      FILLED.canonical()["prices"][0]["provisional"] is False
      and FILLED.canonical()["prices"][-1]["provisional"] is True)
check("...and plan_is_provisional() sees the series as provisional overall, "
      "because at least one hour is",
      plan_is_provisional(FILLED) is True)

# The audit is a post-condition that RAISES. Feed it the exact defect.
_stale_copy = PricePoint(start=FILLED.points[-1].start,
                         eur_per_kwh=FILLED.points[-1].eur_per_kwh,
                         provisional=False)         # the copy-the-flag bug
check("the provenance audit REFUSES a carried hour flagged non-provisional - "
      "the 'stale but plausible' point, which is SAFETY-18's frozen-but-fresh "
      "meter one layer up",
      raises(lambda: _fc._audit([_stale_copy], {}), ValueError))
check("...and refuses a published hour that arrived as a COPY rather than as "
      "the published object, which is where a flag gets lost",
      raises(lambda: _fc._audit(
          [PricePoint(start=TODAY_PART.points[0].start,
                      eur_per_kwh=TODAY_PART.points[0].eur_per_kwh,
                      provisional=TODAY_PART.points[0].provisional)],
          {D0: TODAY_PART.points[0]}), ValueError))
check("...and never repairs a flag: the audit's only outcomes are silence and "
      "an exception, so a series it passed was already correct",
      _fc._audit(list(FILLED.points), {u: TODAY_PART.points[i]
                                       for i, u in enumerate(DAY[:PUBLISHED_N])})
      is None)

# ⚠️ A PUBLISHED POINT MAY BE PROVISIONAL. eneco_eddy's current-quarter print
# is settled after the fact and says so (§9.2). "Published" is not "firm".
_eddy = series_of(DAY[:2], lambda u: t_price(brussels(u).hour),
                  PriceProduct.REALTIME, "eneco_eddy", provisional=True)
_eddy_filled = derived(lambda: fill_from_yesterday(
    _eddy, yesterday=YESTERDAY, since=D0, until=D1, stamp=D0))
check("a PUBLISHED point that the feed itself flagged provisional keeps that "
      "flag - the audit checks provenance, and does not force published hours "
      "to False",
      _eddy_filled.kind == "series"
      and all(p.provisional for p in _eddy_filled.points[:2])
      and _eddy_filled.points[0] is _eddy.points[0])

# The structural half: the constructor has nothing to pass, and the file is
# scanned so the pass-through idiom cannot come back looking ordinary.
check("_carried() has NO `provisional` parameter, so a caller has nothing to "
      "forget, default, or copy from the point being carried",
      "provisional" not in inspect.signature(_fc._carried).parameters)
_SRC = derived(lambda: open(os.path.join(HERE, "app", "price_forecast.py"),
                            encoding="utf-8").read())


def _nonliteral_provisional(src):
    """Every `PricePoint(...)` in the module whose `provisional` is not a
    literal. AST, not a substring: the module docstring quotes the defect it
    exists to prevent, and a text scan would match its own documentation."""
    out = []
    for node in ast.walk(ast.parse(src)):
        if isinstance(node, ast.Call) and \
                getattr(node.func, "id", None) == "PricePoint":
            kw = {k.arg: k.value for k in node.keywords}
            v = kw.get("provisional")
            if not isinstance(v, ast.Constant):
                out.append(ast.dump(node)[:80])
    return out


_leaks = derived(lambda: _nonliteral_provisional(_SRC))
check(f"no PricePoint built in app/price_forecast.py takes its `provisional` "
      f"from an expression - it is a literal at every construction site, so "
      f"`provisional=y.provisional` cannot reappear in a later edit {_leaks}",
      _leaks == [])
# The scan must be able to fail, or it grades nothing.
check("...and that scan FAILS on the defect it names, so it is a check and not "
      "a decoration",
      _nonliteral_provisional(
          "PricePoint(start=t, eur_per_kwh=v, provisional=y.provisional)") != [])


# --------------------------------------------------------------------------- #
print()
print("2. the fill keys on the WALL CLOCK, so the 23- and 25-hour days are right")

for label, (y, m, d) in (("spring forward (23 h)", (2026, 3, 29)),
                         ("fall back (25 h)", (2026, 10, 25))):
    a = local_midnight(y, m, d)
    b = local_midnight(y, m, d + 1)
    yb = local_midnight(y, m, d - 1)
    yesterday = series_of(hours_between(yb, a),
                          lambda u: y_price(brussels(u).hour),
                          PriceProduct.DAY_AHEAD, "generic_dayahead",
                          provisional=False)
    # Nothing published at all yet: the whole day is carried forward.
    nothing = PriceRead.no_data("the day-ahead print has not landed yet")
    got = derived(lambda: fill_from_yesterday(nothing, yesterday=yesterday,
                                              since=a, until=b, stamp=a))
    # ⚠️ DERIVED FROM THE OFFSETS, not from a 23 or a 25 typed here.
    off_a = brussels_offset(a).utcoffset(None).total_seconds() / 3600
    off_b = brussels_offset(b - timedelta(seconds=1)).utcoffset(None) \
        .total_seconds() / 3600
    expect = int(24 - (off_b - off_a))
    check(f"{label}: the day is {expect} hours long, derived from the offset "
          f"change {off_a:+.0f} -> {off_b:+.0f}, and the fill emits exactly "
          f"that many",
          got.kind == "series" and len(got.points) == expect)
    check(f"{label}: every hour of it is flagged provisional - nothing was "
          f"published",
          got.kind == "series" and all(p.provisional for p in got.points))
    # The discriminating part: a naive `t - 24 h` fill would give a DIFFERENT
    # answer here, so this check separates the two implementations.
    y_by_utc = {u: y_price(brussels(u).hour) for u in hours_between(yb, a)}
    naive = [y_by_utc.get(u - timedelta(hours=24)) for u in hours_between(a, b)]
    wall = [y_price(brussels(u).hour) for u in hours_between(a, b)]
    check(f"{label}: the wall-clock fill and a naive t-24h fill DISAGREE, so "
          f"the check below can tell them apart",
          naive != wall)
    check(f"{label}: ...and the emitted values are the WALL-CLOCK ones",
          got.kind == "series"
          and [round(p.eur_per_kwh, 6) for p in got.points] == wall)

# The 25-hour day's two 02:00 hours are two distinct instants, both filled.
_oct = local_midnight(2026, 10, 25)
_oct_end = local_midnight(2026, 10, 26)
_oct_fill = fill_from_yesterday(
    PriceRead.no_data("nothing yet"),
    yesterday=series_of(hours_between(local_midnight(2026, 10, 24), _oct),
                        lambda u: y_price(brussels(u).hour),
                        PriceProduct.DAY_AHEAD, "generic_dayahead",
                        provisional=False),
    since=_oct, until=_oct_end, stamp=_oct)
_twos = [p for p in _oct_fill.points if brussels(p.start).hour == 2]
check("the 25-hour day has TWO local 02:00 points, at different offsets, and "
      "both are filled and both are flagged - a 24-slot implementation would "
      "collide them into one",
      len(_twos) == 2
      and _twos[0].start.utcoffset() != _twos[1].start.utcoffset()
      and all(p.provisional for p in _twos))


# --------------------------------------------------------------------------- #
print()
print("3. §5.4 - a late publication REPLACES a carried hour and re-flags it")

_late = fill_from_yesterday(
    series_of(DAY, lambda u: t_price(brussels(u).hour),
              PriceProduct.DAY_AHEAD, "generic_dayahead", provisional=False),
    yesterday=YESTERDAY, since=D0, until=D1, stamp=D1)
_h20 = DAY[20]
_before = derived(lambda: FILLED.at(_h20))
_after = derived(lambda: _late.at(_h20))
check("before the print, hour 20 is provisional and carries YESTERDAY's number",
      _before.provisional is True and near(_before.eur_per_kwh, y_price(20)))
check("...after it, the same hour is FIRM and carries today's - the flag and "
      "the number move together, because the series is rebuilt and not patched",
      _after.provisional is False and near(_after.eur_per_kwh, t_price(20)))
check("...and the two numbers actually differ, so 'it was replaced' is a "
      "measurement and not a restatement",
      not near(_before.eur_per_kwh, _after.eur_per_kwh))
check("...and re-running the fill over an already-complete series changes "
      "nothing: it is idempotent",
      [p.canonical() for p in fill_from_yesterday(
          _late, yesterday=YESTERDAY, since=D0, until=D1, stamp=D1).points]
      == [p.canonical() for p in _late.points])

# Nothing to fill from, and nothing published: no_data, never an empty series.
_nothing = derived(lambda: fill_from_yesterday(
    PriceRead.no_data("the feed timed out"),
    yesterday=PriceRead.no_data("and there is no yesterday either"),
    since=D0, until=D1, stamp=D0))
check("with neither a published hour nor a yesterday, the result is no_data - "
      "not an empty series, and not a series of zeros",
      _nothing.kind == "no_data")
check("...so it has no .points to iterate and no .get() to default out of",
      raises(lambda: _nothing.points, Absence)
      and raises(lambda: _nothing.get("points", []), Absence))
check("...and its reason names §11.4 for an operator",
      "§11.4" in _nothing.reason)
check("a naive `since` is refused rather than silently interpreted as UTC",
      raises(lambda: fill_from_yesterday(TODAY_PART, yesterday=YESTERDAY,
                                         since=datetime(2026, 1, 15),
                                         until=D1, stamp=D0), ValueError))
check("...and `yesterday` has no default, so a fill with nothing to fill from "
      "cannot be written by accident",
      raises(lambda: fill_from_yesterday(TODAY_PART, since=D0, until=D1,
                                         stamp=D0), TypeError))


# --------------------------------------------------------------------------- #
print()
print("4. AC2 - an irreversible action is REJECTED on a provisional price")


class FakeStore:
    """Same surface as app/store.Store, no disk. Copied from test_maintenance."""

    def __init__(self):
        self.data = {"phase": IDLE, "phase_started": None,
                     "last_completed": None, "last_start_attempt": None}

    def set(self, k, v):
        self.data[k] = v

    def set_time(self, k, when):
        self.data[k] = when

    def get_time(self, k):
        return self.data.get(k)

    def save(self):
        pass


def maint():
    return Maintenance(MaintConfig(enabled=True), FakeStore())


check("the drain's commitment window is DERIVED from the phase machine's own "
      "timeout, so the gate and maintenance.py cannot drift apart",
      near(MAINTENANCE_DRAIN.commits_h, DRAIN_TIMEOUT_MIN / 60.0)
      and near(MAINTENANCE_CHARGE.commits_h, CHARGE_TIMEOUT_MIN / 60.0))
# The shipped `maintenance_start_hour` is 10:00 local, so a drain started on
# schedule runs into the evening - hours a morning-only day-ahead print has not
# covered. That is the ticket's scenario and it is where the window is placed.
START = D0 + timedelta(hours=MaintConfig().start_hour)
_hours = derived(lambda: committed_hours(START, MAINTENANCE_DRAIN))
check(f"...and it commits {len(_hours) if not isinstance(_hours, _Raised) else _hours} "
      f"whole hours, one per hour of the timeout",
      len(_hours) == int(DRAIN_TIMEOUT_MIN / 60))
check("an action that commits nothing is not an irreversible one and cannot be "
      "declared", raises(lambda: Irreversible("X", 0), ValueError)
      and raises(lambda: Irreversible("", 4), ValueError))

# The stimulus: the first hours are firm, the drain's window reaches into the
# carried ones. THIS is the ticket's scenario - a cycle started at 10:00 on
# prices that were published for the morning and guessed for the evening.
_m = maint()
_gate_bad = derived(lambda: begin_phase(_m, START, FILLED, MAINTENANCE_DRAIN))
_first_prov = derived(lambda: next(u for u in _hours if FILLED.at(u).provisional))
check("the drain's window really does reach past the published hours while "
      "staying INSIDE the series, so the refusal below is about provenance and "
      "not about a short series",
      FILLED.covers(_hours[-1])
      and any(not FILLED.at(u).provisional for u in _hours)
      and any(FILLED.at(u).provisional for u in _hours))
check("a maintenance DRAIN whose window includes a carried-forward hour is "
      "BLOCKED", _gate_bad.kind == "blocked")
check("...and the phase machine did not move: it is still idle, so the "
      "rejection is of the ACTION and not merely of a verdict object",
      _m.phase == IDLE)
check("...and the reason names §8.3, the offending hour, and when the real "
      "price is expected",
      "§8.3" in _gate_bad.reason
      and _first_prov.isoformat() in _gate_bad.reason
      and "14-16" in _gate_bad.reason)

# The positive arm, so the block above is discriminating rather than universal.
_firm_long = series_of(hours_between(START,
                                     START + timedelta(hours=len(_hours) + 2)),
                       lambda u: t_price(brussels(u).hour % 24),
                       PriceProduct.DAY_AHEAD, "generic_dayahead",
                       provisional=False)
_m2 = maint()
_gate_ok = derived(lambda: begin_phase(_m2, START, _firm_long,
                                       MAINTENANCE_DRAIN))
check("...while the SAME transition on a fully published series is PERMITTED, "
      "and the phase machine actually moves to drain",
      _gate_ok.kind == "permitted" and _m2.phase == DRAIN)
check("...and the permitting arm carries the evidence it checked - one firm "
      "point per committed hour",
      _gate_ok.kind == "permitted" and len(_gate_ok.hours) == len(_hours)
      and not any(p.provisional for p in _gate_ok.hours))

# The other two ways a price can fail to be an observation.
_m3 = maint()
_gate_none = begin_phase(_m3, START, PriceRead.no_data("the feed timed out"),
                         MAINTENANCE_DRAIN)
check("no price data at all BLOCKS an irreversible action too - a missing "
      "price is not a cheap one (§11.4)",
      _gate_none.kind == "blocked" and "§11.4" in _gate_none.reason
      and _m3.phase == IDLE)
_short = series_of(DAY[:4], lambda u: t_price(brussels(u).hour),
                   PriceProduct.DAY_AHEAD, "generic_dayahead",
                   provisional=False)
_gate_short = price_permits(_short, D0, MAINTENANCE_DRAIN)
check("...and so does an hour the series does not cover, which is §11.4 for "
      "that hour rather than a reason to reach for a neighbour",
      _gate_short.kind == "blocked" and "no price covers" in _gate_short.reason)

# The variant type itself, held to the same shape as PriceRead and Arbitrage.
check("a BLOCKED gate carries a reason and nothing else: .hours and .action "
      "raise Absence rather than answering with an empty tuple",
      raises(lambda: _gate_bad.hours, Absence)
      and raises(lambda: _gate_bad.action, Absence))
check("...it is not iterable and has no .get(), the other two routes to an "
      "empty, harmless-looking loop",
      raises(lambda: list(_gate_bad), Absence)
      and raises(lambda: _gate_bad.get("hours", []), Absence))
check("...match() demands BOTH arms, so the block cannot be fallen through",
      raises(lambda: _gate_bad.match(permitted=lambda g: 1), TypeError))
check("...and there is no falsy variant, so `if gate:` cannot look like a "
      "guard while being none",
      bool(_gate_bad) is True and bool(_gate_ok) is True)
# A genuinely carried point - the flagged one, not the mis-flagged copy above.
_prov_pt = FILLED.points[-1]
check("the point the two constructor checks below are handed really is "
      "provisional, or they would be grading nothing",
      _prov_pt.provisional is True)
check("a permitted gate cannot be built over a provisional hour or over no "
      "hours at all - §8.3 is the one rule this type has",
      raises(lambda: Gate.permit(MAINTENANCE_DRAIN, [_prov_pt]), ValueError)
      and raises(lambda: Gate.permit(MAINTENANCE_DRAIN, []), ValueError))

# TEST-14, PORTED. A default on any of these constructors reopens the exact
# hole `_NoData.reason` had - and static_tariff's `_Refused.reason` still has,
# because nothing scanned it. Scanned by SIGNATURE, so a default added later is
# red without anyone remembering to write a new case.
def _defaulted(fn):
    try:
        ps = list(inspect.signature(fn).parameters.values())
    except (TypeError, ValueError) as e:                      # pragma: no cover
        return _Raised(e)
    return [q.name for q in ps
            if q.name != "self" and q.default is not inspect.Parameter.empty]


_ctors = {"Gate.permit": Gate.permit, "Gate.block": Gate.block,
          "Gate.match": Gate.match,
          "_Permitted.__init__": _fc._Permitted.__init__,
          "_Blocked.__init__": _fc._Blocked.__init__,
          # ⚠️ static_tariff's outcome type is scanned HERE because nothing
          # scanned it before: TEST-14 records `_Refused.reason` gaining a
          # default as a mutation that survived every check in that file.
          "Arbitrage.allow": Arbitrage.allow, "Arbitrage.refuse": Arbitrage.refuse,
          "Arbitrage.match": Arbitrage.match,
          "_Allowed.__init__": _st._Allowed.__init__,
          "_Refused.__init__": _st._Refused.__init__,
          # ⚠️ OPT-04's outcome type is scanned HERE, in the same dict, rather
          # than by a hand-written case in test_arbitrage.py - which is TEST-14's
          # actual instruction. A new outcome type extends the SCAN; a
          # hand-written case is the thing that gets forgotten, and the whole
          # reason this dict exists instead of five separate checks.
          "ArbitrageDecision.__init__": ArbitrageDecision.__init__}
_with_defaults = derived(lambda: {n: d for n, d in
                                  ((n, _defaulted(f)) for n, f in _ctors.items())
                                  if d})
check(f"no constructor or handler of an outcome type - Gate's OR Arbitrage's - "
      f"carries a DEFAULT, so a verdict can never be built out of nothing "
      f"{_with_defaults}", _with_defaults == {})
check("...so a block with no reason is a TypeError at the call site, not a "
      "blank status line",
      raises(lambda: Gate.block(), TypeError)
      and raises(lambda: Gate.block(""), ValueError))
check("...and the same holds for the refusal FCST-03 added to Arbitrage",
      raises(lambda: Arbitrage.refuse(), TypeError))
# ⚠️ ONE CALL PER ROLE, AND THE FIRST DRAFT SHOWED WHY. Passing the provisional
# point as BOTH `cheapest` and `dearest` is a stimulus that cannot distinguish
# "the constructor guards the cheapest hour" from "the constructor guards both
# hours": a guard on either one alone raises, and the check goes green for an
# implementation that leaves the sell hour unguarded. That is the degeneracy
# rule this file states at the top, in a new dress - so the two roles are
# stimulated SEPARATELY, each with a firm point in the other seat.
_firm_pt = TODAY_PART.points[0]
check("the two points these constructor checks use really differ in provenance, "
      "or neither of them grades anything",
      _prov_pt.provisional is True and _firm_pt.provisional is False)
check("an ALLOWED arbitrage verdict cannot be constructed with a provisional "
      "CHEAPEST hour - the buy hour - even directly: §8.3 is enforced at the "
      "constructor, not only in the ladder above it",
      raises(lambda: Arbitrage.allow(0.1, _prov_pt, _firm_pt, "normal",
                                     PriceProduct.DAY_AHEAD, 0.0, frozenset()),
             ValueError))
check("...nor with a provisional DEAREST hour - the sell hour - which a guard "
      "written for `cheapest` alone would have let through",
      raises(lambda: Arbitrage.allow(0.1, _firm_pt, _prov_pt, "normal",
                                     PriceProduct.DAY_AHEAD, 0.0, frozenset()),
             ValueError))
check("...while two firm points build the verdict perfectly well, so the two "
      "refusals above are discriminating rather than universal",
      Arbitrage.allow(0.1, _firm_pt, _firm_pt, "normal",
                      PriceProduct.DAY_AHEAD, 0.0, frozenset()).kind
      == "allowed")


# --------------------------------------------------------------------------- #
print()
print("5. AC3 - real-time thresholds are MEASURABLY wider (measured, not read)")


def plan_with_spread(spread, product, provider):
    """A two-hour series with exactly `spread` between its two prices."""
    lo, hi = 0.150000, round(0.150000 + spread, 6)
    pts = [PricePoint(start=brussels(D0), eur_per_kwh=lo, provisional=False),
           PricePoint(start=brussels(D0 + timedelta(hours=1)),
                      eur_per_kwh=hi, provisional=False)]
    return TariffPlan(PriceRead.series(pts, product, provider, D0), None,
                      "provider")


def min_accepted_spread(product, sources, base, provider="generic_dayahead"):
    """The SMALLEST spread this configuration will trade on, found by bisecting
    the accept/refuse boundary of the real verdict. Nothing here reads a
    factor out of the module it is grading."""
    lo, hi = 0.0, 4.0
    for _ in range(60):
        mid = (lo + hi) / 2
        v = plan_with_spread(mid, product, provider).arbitrage(
            peak_guard_active=False, min_spread_eur_kwh=base,
            uncertainty=sources)
        if v.kind == "allowed":
            hi = mid
        else:
            lo = mid
    return hi


BASE = 0.104000        # a caller's own round-trip cost threshold, EUR/kWh
_normal = derived(lambda: min_accepted_spread(PriceProduct.DAY_AHEAD,
                                              frozenset(), BASE))
_wide = derived(lambda: min_accepted_spread(PriceProduct.REALTIME,
                                            frozenset(), BASE))
_weather = derived(lambda: min_accepted_spread(
    PriceProduct.DAY_AHEAD, frozenset({Uncertainty.NO_WEATHER}), BASE))
_both = derived(lambda: min_accepted_spread(
    PriceProduct.REALTIME, frozenset({Uncertainty.NO_WEATHER}), BASE))

check(f"a day-ahead product trades on any spread above its own threshold "
      f"{_normal:.6f}, which is the base {BASE} it was given",
      near(_normal, BASE, 1e-6))
check(f"a REAL-TIME / imbalance product needs {_wide:.6f} - MEASURABLY wider, "
      f"by a factor of {(_wide / _normal) if _normal else 0:.3f} measured from "
      f"two verdict boundaries and not read from any constant",
      _wide > _normal + 1e-6)
check(f"...and that measured factor equals the declared one for "
      f"REALTIME_PRODUCT ({WIDEN_FACTOR[Uncertainty.REALTIME_PRODUCT]}), so "
      f"the declaration and the behaviour are the same thing",
      near(_wide / _normal, WIDEN_FACTOR[Uncertainty.REALTIME_PRODUCT], 1e-4))
check(f"§11.4's no-weather row reaches the SAME widening path by a different "
      f"trigger, and lands at {_weather:.6f}",
      near(_weather / _normal, WIDEN_FACTOR[Uncertainty.NO_WEATHER], 1e-4))

# ⚠️ THE COMBINATION RULE, ON A STIMULUS WHERE THE THREE CANDIDATES DIFFER.
_f1 = WIDEN_FACTOR[Uncertainty.REALTIME_PRODUCT]
_f2 = WIDEN_FACTOR[Uncertainty.NO_WEATHER]
check(f"with BOTH sources active the threshold is max(f1,f2)={max(_f1, _f2)} x "
      f"base and NOT the product ({_f1 * _f2}) or the sum ({_f1 + _f2}) - two "
      f"independent doubts must not compound into a threshold nothing can "
      f"clear",
      near(_both / _normal, max(_f1, _f2), 1e-4)
      and not near(_both / _normal, _f1 * _f2, 1e-3)
      and not near(_both / _normal, _f1 + _f2, 1e-3))

# The regime must follow the DECLARED product, never the plugin id.
_wide_named_dayahead = derived(lambda: min_accepted_spread(
    PriceProduct.REALTIME, frozenset(), BASE, provider="generic_dayahead"))
_normal_named_eddy = derived(lambda: min_accepted_spread(
    PriceProduct.DAY_AHEAD, frozenset(), BASE, provider="eneco_eddy"))
check("the widening follows the DECLARED PriceProduct even when the provider "
      "is called 'generic_dayahead', and does NOT follow the name "
      "'eneco_eddy' when the declared product is day-ahead - so a sixth "
      "supplier is widened by its declaration and not by an id in a list",
      near(_wide_named_dayahead, _wide, 1e-9)
      and near(_normal_named_eddy, _normal, 1e-9))
check("...and the shipped plugin §9.2 names IS the one that declares REALTIME, "
      "so the mechanism is pointed at the real product",
      PROVIDERS["eneco_eddy"].capabilities().product is PriceProduct.REALTIME
      and uncertainty_of(PriceProduct.REALTIME)
      == frozenset({Uncertainty.REALTIME_PRODUCT})
      and uncertainty_of(PriceProduct.DAY_AHEAD) == frozenset())
check("a string in the uncertainty set is REFUSED rather than widening by 1.0 "
      "and reporting that it widened",
      raises(lambda: widen_factor({"realtime"}), TypeError)
      and raises(lambda: required_spread(0.1, {"realtime"}), TypeError))
check("...and a negative or non-finite base threshold is refused, since it is "
      "one every spread clears",
      raises(lambda: required_spread(-0.01, frozenset()), ValueError)
      and raises(lambda: required_spread(float("nan"), frozenset()), ValueError))
check("required_spread() with nothing uncertain is the base itself, so the "
      "widening is additive machinery and not a permanent surcharge",
      near(required_spread(BASE, frozenset()), BASE))


# --------------------------------------------------------------------------- #
print()
print("6. the refusal ladder - provisional hours are DROPPED, not traded on")

# ⚠️ THE PROVISIONAL HOURS CARRY THE CHEAPEST PRICE. That is the realistic
# danger (yesterday's cheap night luring a charge) AND the non-degenerate
# stimulus: the firm-only spread and the whole-series spread differ, so a check
# on the spread can tell which set was used.
_cheap_stale = [PricePoint(start=brussels(DAY[i]),
                           eur_per_kwh=0.041250, provisional=True)
                for i in range(3)]
_firm_rest = [PricePoint(start=brussels(DAY[i]),
                         eur_per_kwh=round(0.212500 + 0.013750 * i, 6),
                         provisional=False) for i in range(3, 12)]
_mixed = TariffPlan(PriceRead.series(_cheap_stale + _firm_rest,
                                     PriceProduct.DAY_AHEAD,
                                     "generic_dayahead", D0), None, "provider")
_all_spread = derived(lambda: _mixed.spread_eur_kwh())
_firm_spread = derived(lambda: round(
    max(p.eur_per_kwh for p in _firm_rest)
    - min(p.eur_per_kwh for p in _firm_rest), 6))
check(f"the fixture is non-degenerate: the whole-series spread "
      f"{_all_spread:.6f} and the firm-only spread {_firm_spread:.6f} are "
      f"different numbers, so a verdict can be traced to one of them",
      not near(_all_spread, _firm_spread, 1e-6))
_v = derived(lambda: _mixed.arbitrage(peak_guard_active=False,
                                      min_spread_eur_kwh=0.0,
                                      uncertainty=frozenset()))
check("a PART-published series still allows arbitrage - §8.3 forbids acting on "
      "a provisional price, not trading at all until 16:00",
      _v.kind == "allowed")
check(f"...on the FIRM spread {_firm_spread:.6f}, not the one yesterday's "
      f"carried-forward cheap hour would have invented",
      near(_v.spread_eur_kwh, _firm_spread, 1e-9))
check("...and the cheapest and dearest hours it hands an optimiser are both "
      "PUBLISHED hours - the two points ARE the trade",
      _v.cheapest.provisional is False and _v.dearest.provisional is False)
check("...and it reports the threshold it cleared and why, rather than leaving "
      "OPT-04 to re-derive it from a product enum",
      near(_v.required_spread_eur_kwh, 0.0) and _v.uncertainty == frozenset())

_all_prov = TariffPlan(PriceRead.series(_cheap_stale, PriceProduct.DAY_AHEAD,
                                        "generic_dayahead", D0), None,
                       "fallback")
_vp = derived(lambda: _all_prov.arbitrage(peak_guard_active=False,
                                          min_spread_eur_kwh=0.0,
                                          uncertainty=frozenset()))
check("a WHOLLY provisional series - the §11.4 fallback, or a day before any "
      "print - is still refused, and the reason counts the firm hours",
      _vp.kind == "refused" and "0 of 3" in _vp.reason and "§8.3" in _vp.reason)
check("...and the refusal carries no numbers to trade on",
      raises(lambda: _vp.spread_eur_kwh, Absence)
      and raises(lambda: _vp.cheapest, Absence))
_one_firm = TariffPlan(
    PriceRead.series(_cheap_stale + _firm_rest[:1], PriceProduct.DAY_AHEAD,
                     "generic_dayahead", D0), None, "provider")
check("one firm hour is not enough either: a buy hour and a sell hour must "
      "BOTH be published before a cycle can be committed",
      _one_firm.arbitrage(peak_guard_active=False, min_spread_eur_kwh=0.0,
                          uncertainty=frozenset()).kind == "refused")
check("§10.1 still holds above all of it: the peak guard refuses the same "
      "series that is otherwise allowed",
      _mixed.arbitrage(peak_guard_active=True, min_spread_eur_kwh=0.0,
                       uncertainty=frozenset()).kind == "refused")
check("the three questions cannot be skipped: all of peak_guard_active, "
      "min_spread_eur_kwh and uncertainty are required and keyword-only",
      raises(lambda: _mixed.arbitrage(peak_guard_active=False), TypeError)
      and raises(lambda: _mixed.arbitrage(peak_guard_active=False,
                                          min_spread_eur_kwh=0.0), TypeError)
      and raises(lambda: _mixed.arbitrage(False, 0.0, frozenset()), TypeError))
check("...nor answered with a truthy 'unknown' or a negative threshold",
      raises(lambda: _mixed.arbitrage(peak_guard_active=None,
                                      min_spread_eur_kwh=0.0,
                                      uncertainty=frozenset()), TypeError)
      and raises(lambda: _mixed.arbitrage(peak_guard_active=False,
                                          min_spread_eur_kwh=-0.1,
                                          uncertainty=frozenset()), TypeError))

# The cheap-window flag is the other consumer of provisionality, and it is the
# one a live install's maintenance charge floor already reads.
_firm_hour = _firm_rest[0].start
_prov_hour = _cheap_stale[0].start
check("the cheap-window flag is decided per HOUR: a provisional hour is never "
      "cheap, whatever its (stale) number says",
      _mixed.cheap_now(_prov_hour) is False)
check("...while a firm hour below the firm mean still is, so carry-forward "
      "filling does not switch the maintenance charge floor off every morning",
      _mixed.cheap_now(_firm_hour) is True)
# ⚠️ GRADED DIRECTLY, BECAUSE ITS CONSUMERS CANNOT GRADE IT. The §11.4 refusal
# and `cheap_now` both guard provenance independently, so `firm_points()`
# returning the carried hours TOO still leaves both of them answering correctly
# - a stimulus that cannot distinguish an over-reporting accessor from a
# correct one. So the tuple itself is compared with the published points, by
# identity and in order, rather than only the mean it feeds.
check("firm_points() returns EXACTLY the published points, in order - not the "
      "carried ones as well, which its two consumers would have absorbed "
      "without either of them changing its answer",
      list(_mixed.firm_points()) == _firm_rest
      and all(a is b for a, b in zip(_mixed.firm_points(), _firm_rest)))
check("...and the firm mean it compares against is derived from published "
      "hours only",
      near(sum(p.eur_per_kwh for p in _mixed.firm_points())
           / len(_mixed.firm_points()),
           sum(p.eur_per_kwh for p in _firm_rest) / len(_firm_rest)))


# --------------------------------------------------------------------------- #
print()
print("7. CFG-03 and validate.py rule 7 - what this ticket did NOT widen")

_defaults = derived(lambda: sorted(_price.PRICE_OPTION_DEFAULTS))
_cfg = derived(lambda: open(os.path.join(HERE, "config.yaml"),
                            encoding="utf-8").read())
# ⚠️ THIS LIST IS A DRIFT PIN, NOT A BAN ON EVER ADDING AN OPTION. FCST-03 added
# none, and said so here. OPT-04 added FOUR - and added them to this ONE dict
# and to config.yaml, which is precisely the behaviour CFG-03 asks for: the
# alternative is a fourth freehand copy in main.py's `o.get(k, X)` fallbacks.
# The pin is what makes such an addition a decision somebody signs rather than a
# thing that appears; whoever adds the fifth comes here and says which card and
# why, as this line does.
_OPT04_KEYS = ["arbitrage_enabled", "cycle_cost_threshold_eur",
               "arbitrage_soc_floor", "arbitrage_soc_ceiling"]
check(f"the in-code mirror holds exactly TARIFF-02's, TARIFF-03's and OPT-04's "
      f"keys - FCST-03 still added none of its own {_defaults}",
      _defaults == sorted(["price_provider", "price_markup_eur_kwh",
                           "price_taxes_eur_kwh", "export_price_eur_kwh",
                           "price_csv_path", "price_feed_url",
                           "price_horizon_h", "price_timeout_s",
                           "static_profile"] + _OPT04_KEYS))
# ...and the four are genuinely NEW, not a rename of something that was already
# there, or the line above would pass while the mirror silently lost a key.
check("...and OPT-04's four keys are additions rather than replacements: the "
      "nine TARIFF-02/03 keys are all still present",
      isinstance(_defaults, list)
      and len(set(_defaults) - set(_OPT04_KEYS)) == 9)
check("...and every one of them is still in config.yaml, which stays the "
      "source of truth",
      isinstance(_cfg, str)
      and all(f"\n  {k}:" in _cfg for k in _price.PRICE_OPTION_DEFAULTS))


def _names(path):
    try:
        tree = ast.parse(open(path, encoding="utf-8").read())
    except (OSError, SyntaxError):                            # pragma: no cover
        return set()
    out = set()
    for node in ast.walk(tree):
        if isinstance(node, ast.ImportFrom):
            out |= {a.name for a in node.names}
        if isinstance(node, ast.Name):
            out.add(node.id)
        if isinstance(node, ast.Attribute):
            out.add(node.attr)
    return out


_FC = os.path.join(HERE, "app", "price_forecast.py")
_used = derived(lambda: _names(_FC))
check("app/price_forecast.py reads NO options at all - it takes a PriceRead "
      "and instants, so there is no fourth place for an option default to live",
      isinstance(_used, set)
      and not ({"price_option", "PRICE_OPTION_DEFAULTS", "PluginOptions"}
               & _used))
# ⚠️ RULE 7'S TRIPWIRE, RESPECTED RATHER THAN TRIPPED. test_price.py section 12
# pins the set of modules that consume `load_prices()` to exactly
# ["static_tariff.py"], and goes red when a second one appears. This module
# deliberately does NOT fetch: it transforms a read somebody else obtained, so
# rule 7's message about what a configured provider does or does not drive is
# unchanged by this ticket and needed no rewording.
check("...and it does not consume load_prices(), so validate.py rule 7's "
      "pinned consumer list is still true and its wording still accurate",
      "load_prices" not in _used)


print()
print("FAILED" if fails else "OK")
for f in fails:
    print("  -", f)
sys.exit(1 if fails else 0)