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.

⚠️ MUTATING THE MODULE THIS FILE GRADES? PURGE `__pycache__` FIRST - or run
`tools/mutation_run.sh`, which does. `python3 -B` alone does NOT save you: it
stops bytecode being WRITTEN, not a stale one being READ, and a size-preserving
edit reverted inside one second is invisible to CPython's staleness test. That
is TEST-15, and it has already turned one live mutation into a false survivor.
Read MUTATION.md before you report a survivor.
"""

import ast
import inspect
import itertools
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))
# ⚠️ `published` MAPS AN HOUR TO A TUPLE, not to a point (FCST-04 H1). An hour
# may legitimately hold four published points - see the quarter-hourly block
# below - so the audit takes the whole set and checks membership by identity.
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. A real-time / imbalance product's
# current interval is settled after the fact and says so per point (§9.2).
# "Published" is not "firm", and §9.3 carries the flag PER POINT for exactly
# this. No shipped plugin declares REALTIME since TARIFF-07, so the series is
# declared here - which is the only place the case can come from now, and is
# why this stimulus did not go with the plugin.
_rt = series_of(DAY[:2], lambda u: t_price(brussels(u).hour),
                PriceProduct.REALTIME, "_realtime_not_shipped",
                provisional=True)
_eddy_filled = derived(lambda: fill_from_yesterday(
    _rt, 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 _rt.points[0])

# ⚠️ AND A SUB-HOURLY FEED IS NOT DECIMATED TO ONE POINT AN HOUR (FCST-04 H1).
# BELGIAN DAY-AHEAD ITSELF SETTLES PER QUARTER - the 15-minute MTU of
# 2025-10-01, which `generic_dayahead` reads as 96 points a day - and
# `_validate_raw()` accepts any strictly-increasing spacing, so a published
# hour can hold four points. The
# fill walks WHOLE HOURS to decide what needs carrying; that must not become a
# rule about what a published hour may contain. Measured on FCST-04's first
# draft, which keyed the published set `{_utc_hour(p.start): p ...}`: a
# 96-point feed came out 15 points long and its arbitrage spread fell from
# 0.143 to 0.014 EUR/kWh, on an ALL-FIRM, fully-published series.
_QN = 16                                             # four hours of quarters
_QUARTERS = PriceRead.series(
    [PricePoint(start=brussels(D0 + timedelta(minutes=15 * n)),
                eur_per_kwh=round(0.100000 + 0.011300 * n, 6),
                provisional=False)
     for n in range(_QN)],
    PriceProduct.DAY_AHEAD, "generic_dayahead", D0)
_QHOURS = {_fc._utc_hour(p.start) for p in _QUARTERS.points}
check(f"the quarter-hourly fixture really is sub-hourly - {_QN} points across "
      f"{len(_QHOURS)} hours - or every check below passes for an hourly feed "
      f"too and grades nothing",
      len(_QUARTERS.points) == _QN and len(_QHOURS) == _QN // 4)
_qfill = derived(lambda: fill_from_yesterday(_QUARTERS, yesterday=YESTERDAY,
                                             since=D0, until=D1, stamp=D0))
_qfirm = derived(lambda: [p for p in _qfill.points if not p.provisional])
check(f"every one of the {_QN} published points survives the fill BY IDENTITY "
      f"({len(_qfirm)} kept). One-per-hour keying would leave {_QN // 4}, and "
      f"the spread over the hours the feed actually published would collapse "
      f"with them",
      len(_qfirm) == _QN
      and all(any(p is q for q in _QUARTERS.points) for p in _qfirm))
check(f"...and the {len(DAY) - len(_QHOURS)} whole hours it published NOTHING "
      f"for are carried, ONE point each: a carried hour is hourly whatever the "
      f"feed's spacing is, because yesterday's profile is only known per "
      f"wall-clock hour",
      _qfill.kind == "series"
      and len([p for p in _qfill.points if p.provisional])
      == len(DAY) - len(_QHOURS))
check("...and the filled series is still ordered by instant across the two "
      "kinds, so `at()` and `window()` are not reading a shuffled list",
      _qfill.kind == "series"
      and [p.start.astimezone(UTC) for p in _qfill.points]
      == sorted(p.start.astimezone(UTC) for p in _qfill.points))

_PUB_Q = _fc._published_by_hour(_QUARTERS)
# ⚠️ `derived`, NOT A RAW CALL. Under the decimating mutation this check
# exists for, `_audit` RAISES here - and a harness that dies on the mutation it
# is grading reports a traceback instead of a FAIL, which MUTATION.md forbids.
check("the audit ACCEPTS four published points inside one hour, which is the "
      "case the repeated-hour COUNT it replaced could not express",
      derived(lambda: _fc._audit(list(_QUARTERS.points), _PUB_Q)) is None)
check("...and still REFUSES a repeated CARRIED hour, which is the DST "
      "mispricing that count existed to catch",
      raises(lambda: _fc._audit([_fc._carried(D0, 0.11),
                                 _fc._carried(D0, 0.12)], {}), ValueError))
check("...and STILL refuses the SAME published point emitted twice - the "
      "provider's points are strictly increasing, so a repeat is never a "
      "second print. The old hour COUNT already refused this (one point twice "
      "is one hour twice); what is graded here is that the relaxation did not "
      "lose it, not that it added it",
      raises(lambda: _fc._audit([_QUARTERS.points[0], _QUARTERS.points[0]],
                                _PUB_Q), ValueError))
check("...and STILL refuses a CARRIED point standing in an hour the feed DID "
      "publish. The old COUNT never looked at it; the old IDENTITY rule did, "
      "and raised - so this is preserved behaviour, not new strictness. "
      "FCST-04's text called it new here and in `_audit`'s docstring, and a "
      "measured differential says otherwise",
      raises(lambda: _fc._audit([_fc._carried(D0, 0.11)], _PUB_Q), ValueError))

# ⚠️ AND WHAT THE AUDIT DOES NOT DO, STATED WHERE A READER OF THESE THREE
# CHECKS WOULD OTHERWISE ASSUME IT (FCST-05, MUTATION.md habit 16). `_audit` is
# a PROVENANCE post-condition, not a COMPLETENESS one: it grades every point
# that IS emitted and says nothing about points that are not. So it accepts the
# very decimation FCST-04 was raised for. That is not a regression - FCST-03's
# audit was equally blind - but it means the audit is NOT the thing that stops
# H1 coming back. The count and identity checks above this line are.
check("the audit ACCEPTS the H1 decimation itself - four published points in "
      "one hour, only the last emitted - and accepts an EMPTY output. It is a "
      "provenance check, not a completeness one, and the checks that guard "
      "provenance completeness are the COUNT and IDENTITY ones above",
      derived(lambda: _fc._audit([_QUARTERS.points[3]], _PUB_Q)) is None
      and derived(lambda: _fc._audit([], _PUB_Q)) is None)


# --------------------------------------------------------------------------- #
# ⚠️ AND THE CLAIM "STRICTER" IS ITSELF A CHECK NOW (FCST-05).
#
# FCST-04 rewrote `_audit` and wrote, in the module and in the three checks
# above, that the result was STRICTER than FCST-03's. It is not. Two of the
# stated reasons were false - the old `seen` count already refused the same
# point emitted twice, and the old IDENTITY rule already refused a carried
# point in a published hour - and a right decision with a fabricated reason is
# a defect in this repository (MUTATION.md, and the reason DECISIONS files
# exist). So the claim is measured here rather than asserted in prose:
# FCST-03's audit is reproduced below from dd40c56 - its RULES verbatim, its
# message TEXT shortened, and nothing here reads a message - and the two are
# run over the whole domain `_audit` can express, in BOTH directions.
#
# THE DOMAIN, AND ITS SIZE. Every published mask over 3 hours (2**3) crossed
# with every output series of length 0-3 drawn from 3 hours x 5 point kinds.
# The nominal cross-product is 28 928; three of those kinds - published,
# published-provisional, copy-of-published - DO NOT EXIST for an hour the feed
# did not publish, and those combinations are SKIPPED rather than faked, which
# leaves 11 990 realisable comparisons. A fabricated "published point" for an
# unpublished hour is not in either audit's domain, and inventing one would
# grade the fixture rather than the code.
#
# ⚠️ THE OLD COPY IS A LITERAL, NOT A CALL. MUTATION.md habit 2: a differential
# whose two sides both come from the shipped module compares a thing against
# itself and moves with every edit. This side is frozen text.
def _audit_fcst03(points, published):
    """FCST-03's `_audit`, verbatim from dd40c56. Its `published` maps a UTC
    hour to ONE point, which is exactly the expressiveness this ticket relaxed.
    """
    seen = set()
    for p in points:
        u = _fc._utc_hour(p.start)
        if u in seen:
            raise ValueError("the filled series repeats " + u.isoformat())
        seen.add(u)
        src = published.get(u)
        if src is None:
            if p.provisional is not True:
                raise ValueError(u.isoformat() + " is carried and is flagged "
                                 "provisional=" + repr(p.provisional))
        else:
            if p is not src:
                raise ValueError(u.isoformat() + " is published but the series "
                                 "carries a COPY of that point")


def _verdict(fn, pts, pub):
    try:
        fn(list(pts), pub)
        return "accept"
    except ValueError:
        return "raise"


_DH = [D0 + timedelta(hours=6 + i) for i in range(3)]
_weaker, _stronger, _compared = [], [], 0
for _mask in itertools.product((0, 1), repeat=3):
    _pub = {u: PricePoint(start=u, eur_per_kwh=0.10 + i, provisional=False)
            for i, (u, _on) in enumerate(zip(_DH, _mask)) if _on}
    _old_pub = dict(_pub)
    _new_pub = {u: (q,) for u, q in _pub.items()}
    # The five shapes an output point can take at a given hour. Three of them
    # do not exist for an hour the feed did not publish, and those combinations
    # are skipped rather than faked - a fabricated "published point" for an
    # unpublished hour is not in either audit's domain.
    _kinds = []
    for _u in _DH:
        _src = _pub.get(_u)
        _kinds.append(("published", _src))
        _kinds.append(("published-provisional",
                       None if _src is None else
                       PricePoint(start=_src.start,
                                  eur_per_kwh=_src.eur_per_kwh,
                                  provisional=True)))
        _kinds.append(("carried", _fc._carried(_u, 0.5)))
        _kinds.append(("carried-unflagged",
                       PricePoint(start=_u, eur_per_kwh=0.5,
                                  provisional=False)))
        _kinds.append(("copy-of-published",
                       None if _src is None else
                       PricePoint(start=_src.start,
                                  eur_per_kwh=_src.eur_per_kwh,
                                  provisional=_src.provisional)))
    for _n in range(4):
        for _combo in itertools.product(_kinds, repeat=_n):
            if any(k[1] is None for k in _combo):
                continue
            _pts = [k[1] for k in _combo]
            _compared += 1
            _o = _verdict(_audit_fcst03, _pts, _old_pub)
            _nw = _verdict(_fc._audit, _pts, _new_pub)
            if _o == "raise" and _nw == "accept":
                _weaker.append((_mask, [k[0] for k in _combo]))
            if _o == "accept" and _nw == "raise":
                _stronger.append((_mask, [k[0] for k in _combo]))

# ⚠️ THE PRECONDITION FIRST (MUTATION.md habit 1). A differential over a
# domain that cannot reach the two disputed cases would score 0-0 for the wrong
# reason and read exactly like this one, so the two cases are driven against
# the OLD audit by name before the totals are graded.
_probe = PricePoint(start=_DH[0], eur_per_kwh=0.1, provisional=False)
check(f"the differential really covers the two claims it is here to settle, "
      f"or it is {_compared} runs of nothing: FCST-03's own audit raises on "
      f"BOTH 'the same published point twice' and 'a carried point in a "
      f"published hour', which is what FCST-04 said it did not",
      _compared > 3000
      and _verdict(_audit_fcst03, [_probe, _probe], {_DH[0]: _probe}) == "raise"
      and _verdict(_audit_fcst03, [_fc._carried(_DH[0], 0.5)],
                   {_DH[0]: _probe}) == "raise")
check(f"...and over all {_compared} of them the reworked audit is NOT WEAKER "
      f"than FCST-03's: {len(_weaker)} case(s) the old one raised and this one "
      f"accepts {_weaker[:2]}",
      _weaker == [])
check(f"...and NOT STRONGER either: {len(_stronger)} case(s) the old one "
      f"accepted and this one raises {_stronger[:2]}. FCST-04's docstring and "
      f"two of these checks said it was stricter and named two reasons, and "
      f"BOTH were already refused by the old count and the old identity rule. "
      f"The only behavioural change is the RELAXATION below, which is what the "
      f"rework was for",
      _stronger == [])
_QP = [PricePoint(start=_DH[0] + timedelta(minutes=15 * i),
                  eur_per_kwh=0.1 + i, provisional=False) for i in range(4)]
check("...and the relaxation is real and is NOT expressible to the old audit: "
      "four distinct published points inside one hour, which FCST-03's count "
      "refused as a repeated hour and this one accepts",
      _verdict(_fc._audit, _QP, {_DH[0]: tuple(_QP)}) == "accept"
      and _verdict(_audit_fcst03, _QP, {_DH[0]: _QP[0]}) == "raise")

# ⚠️ THE HORIZON IS WHOLE HOURS AT BOTH ENDS (FCST-04 L8). `since` is floored
# and `until` was not, so a 48 h request made at any instant off the hour came
# back 49 hours long. Graded at an OFF-THE-HOUR instant, because on the hour
# the two implementations agree and the check would grade nothing.
_off = D0 + timedelta(hours=8, minutes=17, seconds=43)
for _h in (1, 12, 48):
    _n = derived(lambda h=_h: len(fill_from_yesterday(
        TODAY_PART, yesterday=YESTERDAY, since=_off,
        until=_off + timedelta(hours=h), stamp=_off).points))
    check(f"a {_h} h horizon asked for at {_off.isoformat()} - 17 min 43 s "
          f"past the hour, the shape a running controller always has - is "
          f"{_h} points long, not {_h + 1}",
          _n == _h)

# 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 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))
# ⚠️ 'eneco_eddy' ABOVE IS NOW A RETIRED ID AND THE CHECK IS STRONGER FOR IT.
# It was chosen because it was the one plugin that really did declare REALTIME,
# so the check read "the name does not override the declaration". It declares
# nothing now, which makes the string a pure name with no product behind it -
# exactly the input an id-sniffing implementation would still widen.
check("...and NO shipped plugin declares REALTIME since TARIFF-07, so the "
      "widening path is reached by declaration alone - kept deliberately, see "
      "DECISIONS/TARIFF-07-withdrawing-three-plugins.md",
      [p for p, c in PROVIDERS.items()
       if c.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"]
# TARIFF-15 added the fifth, and came here to say so as this comment instructs:
# `price_formula` is the `custom_formula` provider's only option - one JSON
# string carrying one contract's coefficients - and it lives in this dict and
# in config.yaml and nowhere else, which is the arrangement `static_profile`
# already has for the same reason.
_T15_KEYS = ["price_formula"]
# TARIFF-17 added the sixth, and comes here to say so as the comment above
# instructs. `price_taxes_profile` is the hour-varying spelling of
# `price_taxes_eur_kwh` - a period table of the TAXES component, for the Walloon
# and Brussels DSOs whose distribution tariff really does have a day and a night
# band (ORES 13,27 / 7,39 c EUR/kWh). It is an ADDITION and not a replacement:
# the scalar is untouched, every site on it is unaffected, and validate.py
# refuses the configuration that sets both. It lives in this dict and in
# config.yaml and nowhere else, which is `static_profile`'s arrangement for
# `static_profile`'s reason.
_T17_KEYS = ["price_taxes_profile"]
check(f"the in-code mirror holds exactly TARIFF-02's, TARIFF-03's, OPT-04's, "
      f"TARIFF-15's and TARIFF-17's keys - FCST-03 still added none of its own "
      f"{_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 + _T15_KEYS + _T17_KEYS))
# ...and the later keys 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, TARIFF-15's one and TARIFF-17's one 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) - set(_T15_KEYS)
              - set(_T17_KEYS)) == 9)
# ⚠️ AND THE ONE TARIFF-17 IS THE HOUR-VARYING TWIN OF IS STILL THERE, BY NAME.
# The line above counts nine, and a rename of `price_taxes_eur_kwh` to
# `price_taxes_profile` would also count nine. The scalar path being unchanged
# is this card's first acceptance criterion, and a count cannot grade it.
# ⚠️ THE LITERAL 0.11736 IS TARIFF-18's, AND IT IS TYPED HERE ON PURPOSE. The
# claim is "the scalar is still its own key, holding its own real number", and
# reading that number out of the dict under test would compare it against
# itself. TARIFF-18 raised it from 0.06; if it moves again this line goes red
# and whoever moved it comes here and says which card.
check("...and `price_taxes_eur_kwh` is still its own key beside the table, so "
      "TARIFF-17 widened the framework rather than renaming the scalar - and "
      "it still holds TARIFF-18's Flemish figure",
      isinstance(_defaults, list)
      and "price_taxes_eur_kwh" in _defaults
      and "price_taxes_profile" in _defaults
      and _price.PRICE_OPTION_DEFAULTS["price_taxes_eur_kwh"] == 0.11736
      and _price.PRICE_OPTION_DEFAULTS["price_taxes_profile"] == "")
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)